fixed

apache · Feb 18, 2024 · 62538dd · 62538dd
1 parent 9a62197
commit 62538dd
Show file tree

Hide file tree

Showing 10 changed files with 61 additions and 63 deletions.
diff --git a/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcMetadataProcessor.java b/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcMetadataProcessor.java
@@ -55,7 +55,8 @@ public enum CdcMetadataProcessor {
             SyncJobHandler.SourceType.SQLSERVER,
             new CdcMetadataConverter.DatabaseNameConverter(),
             new CdcMetadataConverter.TableNameConverter(),
-            new CdcMetadataConverter.SchemaNameConverter());
+            new CdcMetadataConverter.SchemaNameConverter(),
+            new CdcMetadataConverter.OpTsConverter());
 
     private final SyncJobHandler.SourceType sourceType;
 

diff --git a/...ink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncJobHandler.java b/...ink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncJobHandler.java
@@ -27,6 +27,7 @@
 import org.apache.paimon.flink.action.cdc.postgres.PostgresRecordParser;
 import org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils;
 import org.apache.paimon.flink.action.cdc.sqlserver.SqlServerRecordParser;
+import org.apache.paimon.flink.action.cdc.sqlserver.SqlServerSourceOptions;
 import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
 
 import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
@@ -47,6 +48,7 @@
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.MYSQL_CONF;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.POSTGRES_CONF;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.PULSAR_CONF;
+import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.SQLSERVER_CONF;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.checkOneRequiredOption;
 import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.checkRequiredOptions;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -139,6 +141,26 @@ public void checkRequiredOption() {
                                     + "use postgres_sync_table instead.");
                 }
                 break;
+            case SQLSERVER:
+                checkRequiredOptions(
+                        cdcSourceConfig,
+                        SQLSERVER_CONF,
+                        SqlServerSourceOptions.HOSTNAME,
+                        SqlServerSourceOptions.USERNAME,
+                        SqlServerSourceOptions.PASSWORD,
+                        SqlServerSourceOptions.DATABASE_NAME);
+                if (isTableSync) {
+                    checkRequiredOptions(
+                            cdcSourceConfig, SQLSERVER_CONF, SqlServerSourceOptions.TABLE_NAME);
+                } else {
+                    checkArgument(
+                            !cdcSourceConfig.contains(SqlServerSourceOptions.TABLE_NAME),
+                            SqlServerSourceOptions.TABLE_NAME.key()
+                                    + " cannot be set for sqlserver_sync_database. "
+                                    + "If you want to sync several SQLServer tables into one Paimon table, "
+                                    + "use sqlserver_sync_table instead.");
+                }
+                break;
             case KAFKA:
                 checkRequiredOptions(
                         cdcSourceConfig,

diff --git a/...-cdc/src/main/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerActionUtils.java b/...-cdc/src/main/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerActionUtils.java
@@ -53,8 +53,8 @@
 import static com.ververica.cdc.debezium.table.DebeziumOptions.getDebeziumProperties;
 import static org.apache.paimon.flink.action.MultiTablesSinkMode.COMBINED;
 import static org.apache.paimon.flink.action.MultiTablesSinkMode.DIVIDED;
-import static org.apache.paimon.flink.action.cdc.mysql.MySqlTypeUtils.toPaimonTypeVisitor;
 import static org.apache.paimon.flink.action.cdc.sqlserver.SqlServerSourceOptions.SCAN_STARTUP_MODE;
+import static org.apache.paimon.flink.action.cdc.sqlserver.SqlServerTypeUtils.toPaimonTypeVisitor;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /** Utility class for SqlServer action. */

diff --git a/...java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncDatabaseActionFactory.java b/...java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncDatabaseActionFactory.java
@@ -38,7 +38,7 @@
 /** Factory to create {@link SqlServerSyncDatabaseAction}. */
 public class SqlServerSyncDatabaseActionFactory implements ActionFactory {
 
-    public static final String IDENTIFIER = "sqlserver-sync-database";
+    public static final String IDENTIFIER = "sqlserver_sync_database";
 
     @Override
     public String identifier() {

diff --git a/...in/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncTableActionFactory.java b/...in/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncTableActionFactory.java
@@ -26,7 +26,7 @@
 /** Factory to create {@link SqlServerSyncTableAction}. */
 public class SqlServerSyncTableActionFactory extends SyncTableActionFactoryBase {
 
-    public static final String IDENTIFIER = "sqlserver-sync-table";
+    public static final String IDENTIFIER = "sqlserver_sync_table";
 
     @Override
     public String identifier() {

diff --git a/...nk-cdc/src/main/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerTypeUtils.java b/...nk-cdc/src/main/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerTypeUtils.java
@@ -18,6 +18,7 @@
 
 package org.apache.paimon.flink.action.cdc.sqlserver;
 
+import org.apache.paimon.flink.action.cdc.JdbcToPaimonTypeVisitor;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypes;
@@ -151,4 +152,23 @@ public static DataType toPaimonDataType(
                         String.format("Don't support SQLServer type '%s' yet.", type));
         }
     }
+
+    public static JdbcToPaimonTypeVisitor toPaimonTypeVisitor() {
+        return SqlServerTypeUtils.SqlServerToPaimonTypeVisitor.INSTANCE;
+    }
+
+    private static class SqlServerToPaimonTypeVisitor implements JdbcToPaimonTypeVisitor {
+
+        private static final SqlServerTypeUtils.SqlServerToPaimonTypeVisitor INSTANCE =
+                new SqlServerTypeUtils.SqlServerToPaimonTypeVisitor();
+
+        @Override
+        public DataType visit(
+                String type,
+                @Nullable Integer length,
+                @Nullable Integer scale,
+                TypeMapping typeMapping) {
+            return toPaimonDataType(type, length, scale, typeMapping);
+        }
+    }
 }
diff --git a/...paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/...paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -25,3 +25,6 @@ org.apache.paimon.flink.action.cdc.pulsar.PulsarSyncDatabaseActionFactory
 org.apache.paimon.flink.action.cdc.mongodb.MongoDBSyncTableActionFactory
 org.apache.paimon.flink.action.cdc.mongodb.MongoDBSyncDatabaseActionFactory
 org.apache.paimon.flink.action.cdc.postgres.PostgresSyncTableActionFactory
+org.apache.paimon.flink.action.cdc.sqlserver.SqlServerSyncDatabaseActionFactory
+org.apache.paimon.flink.action.cdc.sqlserver.SqlServerSyncTableActionFactory
+
diff --git a/...aimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java b/...aimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
@@ -259,6 +259,9 @@ private <T> String getConfKey(Class<T> clazz) {
                 return "--" + CdcActionCommonUtils.PULSAR_CONF;
             case "PostgresSyncTableAction":
                 return "--" + CdcActionCommonUtils.POSTGRES_CONF;
+            case "SqlServerSyncTableAction":
+            case "SqlServerSyncDatabaseAction":
+                return "--" + CdcActionCommonUtils.SQLSERVER_CONF;
             default:
                 throw new UnsupportedOperationException(
                         "Unknown sync action: " + clazz.getSimpleName());

diff --git a/.../java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncDatabaseActionITCase.java b/.../java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncDatabaseActionITCase.java
@@ -111,9 +111,9 @@ public void testSpecifiedSqlServerTable() {
         assertThatThrownBy(action::run)
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage(
-                        "table-name cannot be set for sqlserver-sync-database. "
+                        "table-name cannot be set for sqlserver_sync_database. "
                                 + "If you want to sync several SqlServer tables into one Paimon table, "
-                                + "use sqlserver-sync-table instead.");
+                                + "use sqlserver_sync_table instead.");
     }
 
     @Test
@@ -297,8 +297,8 @@ public void testIgnoreCase() throws Exception {
         FileStoreTable table = getFileStoreTable("t");
         assertThat(JsonSerdeUtil.toFlatJson(table.schema().fields()))
                 .isEqualTo(
-                        "[{\"id\":0,\"name\":\"k\",\"type\":\"INT NOT NULL\",\"description\":\"\"},"
-                                + "{\"id\":1,\"name\":\"uppercase_v0\",\"type\":\"VARCHAR(20)\",\"description\":\"\"}]");
+                        "[{\"id\":0,\"name\":\"k\",\"type\":\"INT NOT NULL\"},"
+                                + "{\"id\":1,\"name\":\"uppercase_v0\",\"type\":\"VARCHAR(20)\"}]");
     }
 
     @Test

diff --git a/...est/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncTableActionITCase.java b/...est/java/org/apache/paimon/flink/action/cdc/sqlserver/SqlServerSyncTableActionITCase.java
@@ -36,7 +36,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ThreadLocalRandom;
 
 import static org.apache.paimon.testutils.assertj.AssertionUtils.anyCauseMatches;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -72,9 +71,9 @@ public void testSyncSqlServerTable() throws Exception {
         runActionWithDefaultEnv(action);
 
         checkTableSchema(
-                "[{\"id\":0,\"name\":\"pt\",\"type\":\"INT NOT NULL\",\"description\":\"\"},"
-                        + "{\"id\":1,\"name\":\"_id\",\"type\":\"INT NOT NULL\",\"description\":\"\"},"
-                        + "{\"id\":2,\"name\":\"v1\",\"type\":\"VARCHAR(10)\",\"description\":\"\"}]");
+                "[{\"id\":0,\"name\":\"pt\",\"type\":\"INT NOT NULL\"},"
+                        + "{\"id\":1,\"name\":\"_id\",\"type\":\"INT NOT NULL\"},"
+                        + "{\"id\":2,\"name\":\"v1\",\"type\":\"VARCHAR(10)\"}]");
 
         try (Statement statement = getStatement()) {
             testSyncSqlServerTableImpl(statement);
@@ -262,7 +261,7 @@ public void testInvalidPrimaryKey() throws Exception {
                 .satisfies(
                         anyCauseMatches(
                                 IllegalArgumentException.class,
-                                "Specified primary key 'pk' does not exist in source tables or computed columns."));
+                                "Specified primary key 'pk' does not exist in source tables or computed columns [pt, _id, v1]."));
     }
 
     @Test
@@ -483,56 +482,6 @@ public void testCatalogAndTableConfig() {
                 .containsExactlyEntriesOf(Collections.singletonMap("table-key", "table-value"));
     }
 
-    @Test
-    @Timeout(60)
-    public void testSyncShardsMultipleSchemas() throws Exception {
-        Map<String, String> sqlServerConfig = getBasicSqlServerConfig();
-
-        // test table list
-        ThreadLocalRandom random = ThreadLocalRandom.current();
-        String schemaPattern = random.nextBoolean() ? "shard_.+" : "shard_1|shard_2";
-        String tblPattern = random.nextBoolean() ? "t.+" : "t1|t2";
-
-        sqlServerConfig.put("database-name", "shard_schema");
-        sqlServerConfig.put("schema-name", schemaPattern);
-        sqlServerConfig.put("table-name", tblPattern);
-
-        SqlServerSyncTableAction action =
-                syncTableActionBuilder(sqlServerConfig)
-                        .withPartitionKeys("pt")
-                        .withPrimaryKeys("pk", "pt")
-                        .withComputedColumnArgs("pt=substring(_date,5)")
-                        .build();
-        runActionWithDefaultEnv(action);
-
-        try (Statement statement = getStatement()) {
-            statement.executeUpdate("USE shard_schema");
-            statement.executeUpdate("INSERT INTO shard_1.t1 VALUES (1, '2023-07-30')");
-            statement.executeUpdate("INSERT INTO shard_1.t2 VALUES (2, '2023-07-30')");
-            statement.executeUpdate("INSERT INTO shard_2.t1 VALUES (3, '2023-07-31')");
-            statement.executeUpdate("INSERT INTO shard_2.t1 VALUES (4, '2023-07-31')");
-        }
-
-        FileStoreTable table = getFileStoreTable();
-        RowType rowType =
-                RowType.of(
-                        new DataType[] {
-                            DataTypes.INT().notNull(),
-                            DataTypes.VARCHAR(10),
-                            DataTypes.STRING().notNull()
-                        },
-                        new String[] {"pk", "_date", "pt"});
-        waitForResult(
-                Arrays.asList(
-                        "+I[1, 2023-07-30, 07-30]",
-                        "+I[2, 2023-07-30, 07-30]",
-                        "+I[3, 2023-07-31, 07-31]",
-                        "+I[4, 2023-07-31, 07-31]"),
-                table,
-                rowType,
-                Arrays.asList("pk", "pt"));
-    }
-
     private FileStoreTable getFileStoreTable() throws Exception {
         return getFileStoreTable(tableName);
     }