Add mysql read to managed io

Author: shunping

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto

@@ -80,6 +80,8 @@ message ManagedTransforms {
       "beam:schematransform:org.apache.beam:postgres_read:v1"];
     POSTGRES_WRITE = 8 [(org.apache.beam.model.pipeline.v1.beam_urn) =
       "beam:schematransform:org.apache.beam:postgres_write:v1"];
+    MYSQL_READ = 9 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+      "beam:schematransform:org.apache.beam:mysql_read:v1"];
   }
 }
 

sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/providers/ReadFromMySqlSchemaTransformProvider.java

@@ -18,20 +18,28 @@
 package org.apache.beam.sdk.io.jdbc.providers;
 
 import static org.apache.beam.sdk.io.jdbc.JdbcUtil.MYSQL;
+import static org.apache.beam.sdk.util.construction.BeamUrns.getUrn;
 
 import com.google.auto.service.AutoService;
+import org.apache.beam.model.pipeline.v1.ExternalTransforms;
 import org.apache.beam.sdk.io.jdbc.JdbcReadSchemaTransformProvider;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
 import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
 import org.checkerframework.checker.initialization.qual.Initialized;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @AutoService(SchemaTransformProvider.class)
 public class ReadFromMySqlSchemaTransformProvider extends JdbcReadSchemaTransformProvider {
 
+  private static final Logger LOG =
+      LoggerFactory.getLogger(ReadFromMySqlSchemaTransformProvider.class);
+
   @Override
   public @UnknownKeyFor @NonNull @Initialized String identifier() {
-    return "beam:schematransform:org.apache.beam:mysql_read:v1";
+    return getUrn(ExternalTransforms.ManagedTransforms.Urns.MYSQL_READ);
   }
 
   @Override
@@ -43,4 +51,25 @@ public String description() {
   protected String jdbcType() {
     return MYSQL;
   }
+
+  @Override
+  public @UnknownKeyFor @NonNull @Initialized SchemaTransform from(
+      JdbcReadSchemaTransformConfiguration configuration) {
+    String jdbcType = configuration.getJdbcType();
+    if (jdbcType != null && !jdbcType.equals(jdbcType())) {
+      throw new IllegalArgumentException(
+          String.format("Wrong JDBC type. Expected '%s' but got '%s'", jdbcType(), jdbcType));
+    }
+
+    Integer fetchSize = configuration.getFetchSize();
+    if (fetchSize != null
+        && fetchSize > 0
+        && configuration.getJdbcUrl() != null
+        && !configuration.getJdbcUrl().contains("useCursorFetch=true")) {
+      LOG.warn(
+          "The fetchSize option is ignored. It is required to set useCursorFetch=true"
+              + " in the JDBC URL when using fetchSize for MySQL");
+    }
+    return super.from(configuration);
+  }
 }

sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java

@@ -97,6 +97,7 @@ public class Managed {
   public static final String KAFKA = "kafka";
   public static final String BIGQUERY = "bigquery";
   public static final String POSTGRES = "postgres";
+  public static final String MYSQL = "mysql";
 
   // Supported SchemaTransforms
   public static final Map<String, String> READ_TRANSFORMS =
@@ -106,6 +107,7 @@ public class Managed {
           .put(KAFKA, getUrn(ExternalTransforms.ManagedTransforms.Urns.KAFKA_READ))
           .put(BIGQUERY, getUrn(ExternalTransforms.ManagedTransforms.Urns.BIGQUERY_READ))
           .put(POSTGRES, getUrn(ExternalTransforms.ManagedTransforms.Urns.POSTGRES_READ))
+          .put(MYSQL, getUrn(ExternalTransforms.ManagedTransforms.Urns.MYSQL_READ))
           .build();
   public static final Map<String, String> WRITE_TRANSFORMS =
       ImmutableMap.<String, String>builder()

sdks/python/apache_beam/transforms/external.py

@@ -83,6 +83,7 @@
     ManagedTransforms.Urns.BIGQUERY_WRITE.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,  # pylint: disable=line-too-long
     ManagedTransforms.Urns.POSTGRES_READ.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,
     ManagedTransforms.Urns.POSTGRES_WRITE.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,  # pylint: disable=line-too-long
+    ManagedTransforms.Urns.MYSQL_READ.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,
 }
 
 

sdks/python/apache_beam/transforms/managed.py

@@ -86,6 +86,7 @@
 KAFKA = "kafka"
 BIGQUERY = "bigquery"
 POSTGRES = "postgres"
+MYSQL = "mysql"
 
 __all__ = ["ICEBERG", "KAFKA", "BIGQUERY", "Read", "Write"]
 
@@ -98,6 +99,7 @@ class Read(PTransform):
       KAFKA: ManagedTransforms.Urns.KAFKA_READ.urn,
       BIGQUERY: ManagedTransforms.Urns.BIGQUERY_READ.urn,
       POSTGRES: ManagedTransforms.Urns.POSTGRES_READ.urn,
+      MYSQL: ManagedTransforms.Urns.MYSQL_READ.urn,
   }
 
   def __init__(