(beam) branch master updated: [YAML] Add JDBC branding transforms (#29171)

[ro...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

robertwb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new e4bd6183b70 [YAML] Add JDBC branding transforms (#29171)
e4bd6183b70 is described below

commit e4bd6183b70c4bcb9db1b86aafaa3be0a0e75a7a
Author: Jeff Kinard <35...@users.noreply.github.com>
AuthorDate: Tue Oct 31 16:24:36 2023 -0400

    [YAML] Add JDBC branding transforms (#29171)
    
    Add specific jdbc io's for mysql, postgres, oracle and sql server.
---
 .../schemaio-expansion-service/build.gradle        |  4 +++
 sdks/python/apache_beam/yaml/standard_io.yaml      | 41 ++++++++++++++++++++++
 sdks/python/apache_beam/yaml/yaml_provider.py      | 18 +++++++++-
 3 files changed, 62 insertions(+), 1 deletion(-)

diff --git a/sdks/java/extensions/schemaio-expansion-service/build.gradle b/sdks/java/extensions/schemaio-expansion-service/build.gradle
index 246c0c155cb..ae6599dfbb9 100644
--- a/sdks/java/extensions/schemaio-expansion-service/build.gradle
+++ b/sdks/java/extensions/schemaio-expansion-service/build.gradle
@@ -46,6 +46,10 @@ dependencies {
     implementation project(path: ":sdks:java:core", configuration: "shadow")
     implementation library.java.vendored_grpc_1_54_0
     implementation library.java.vendored_guava_32_1_2_jre
+    // Stage default drivers for JdbcIO schema transforms
+    implementation 'mysql:mysql-connector-java:8.0.22'
+    implementation 'com.oracle.database.jdbc:ojdbc8:23.2.0.0'
+    implementation 'com.microsoft.sqlserver:mssql-jdbc:12.2.0.jre11'
     testImplementation library.java.junit
     testImplementation library.java.mockito_core
 }
diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml
index 6499ffc7e55..236046abb9b 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -138,6 +138,14 @@
   transforms:
     'ReadFromJdbc': 'ReadFromJdbc'
     'WriteToJdbc': 'WriteToJdbc'
+    'ReadFromMySql': 'ReadFromJdbc'
+    'WriteToMySql': 'WriteToJdbc'
+    'ReadFromPostgres': 'ReadFromJdbc'
+    'WriteToPostgres': 'WriteToJdbc'
+    'ReadFromOracle': 'ReadFromJdbc'
+    'WriteToOracle': 'WriteToJdbc'
+    'ReadFromSqlServer': 'ReadFromJdbc'
+    'WriteToSqlServer': 'WriteToJdbc'
   config:
     mappings:
       'ReadFromJdbc':
@@ -159,10 +167,43 @@
         driver_jars: 'driverJars'
         connection_properties: 'connectionProperties'
         connection_init_sql: 'connectionInitSql'
+      'ReadFromMySql': 'ReadFromJdbc'
+      'WriteToMySql': 'WriteToJdbc'
+      'ReadFromPostgres': 'ReadFromJdbc'
+      'WriteToPostgres': 'WriteToJdbc'
+      'ReadFromOracle':  'ReadFromJdbc'
+      'WriteToOracle': 'WriteToJdbc'
+      'ReadFromSqlServer': 'ReadFromJdbc'
+      'WriteToSqlServer': 'WriteToJdbc'
+    defaults:
+      'ReadFromMySql':
+        driverClassName: 'com.mysql.jdbc.Driver'
+      'WriteToMySql':
+        driverClassName: 'com.mysql.jdbc.Driver'
+      'ReadFromPostgres':
+        driverClassName: 'org.postgresql.Driver'
+      'WriteToPostgres':
+        driverClassName: 'org.postgresql.Driver'
+      'ReadFromOracle':
+        driverClassName: 'oracle.jdbc.driver.OracleDriver'
+      'WriteToOracle':
+        driverClassName: 'oracle.jdbc.driver.OracleDriver'
+      'ReadFromSqlServer':
+        driverClassName: 'com.microsoft.sqlserver.jdbc.SQLServerDriver'
+      'WriteToSqlServer':
+        driverClassName: 'com.microsoft.sqlserver.jdbc.SQLServerDriver'
     underlying_provider:
       type: beamJar
       transforms:
         'ReadFromJdbc': 'beam:schematransform:org.apache.beam:jdbc_read:v1'
         'WriteToJdbc': 'beam:schematransform:org.apache.beam:jdbc_write:v1'
+        'ReadFromMySql': 'beam:schematransform:org.apache.beam:jdbc_read:v1'
+        'WriteToMySql': 'beam:schematransform:org.apache.beam:jdbc_write:v1'
+        'ReadFromPostgres': 'beam:schematransform:org.apache.beam:jdbc_read:v1'
+        'WriteToPostgres': 'beam:schematransform:org.apache.beam:jdbc_write:v1'
+        'ReadFromOracle': 'beam:schematransform:org.apache.beam:jdbc_read:v1'
+        'WriteToOracle': 'beam:schematransform:org.apache.beam:jdbc_write:v1'
+        'ReadFromSqlServer': 'beam:schematransform:org.apache.beam:jdbc_read:v1'
+        'WriteToSqlServer': 'beam:schematransform:org.apache.beam:jdbc_write:v1'
       config:
         gradle_target: 'sdks:java:extensions:schemaio-expansion-service:shadowJar'
diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py
index a21db09b50c..2d7471e506b 100644
--- a/sdks/python/apache_beam/yaml/yaml_provider.py
+++ b/sdks/python/apache_beam/yaml/yaml_provider.py
@@ -739,9 +739,25 @@ class RenamingProvider(Provider):
     for transform in transforms.keys():
       if transform not in mappings:
         raise ValueError(f'Missing transform {transform} in mappings.')
-    self._mappings = mappings
+    self._mappings = self.expand_mappings(mappings)
     self._defaults = defaults or {}
 
+  @staticmethod
+  def expand_mappings(mappings):
+    if not isinstance(mappings, dict):
+      raise ValueError(
+          "RenamingProvider mappings must be dict of transform "
+          "mappings.")
+    for key, value in mappings.items():
+      if isinstance(value, str):
+        if value not in mappings.keys():
+          raise ValueError(
+              "RenamingProvider transform mappings must be dict or "
+              "specify transform that has mappings within same "
+              "provider.")
+        mappings[key] = mappings[value]
+    return mappings
+
   def available(self) -> bool:
     return self._underlying_provider.available()