[FEATURE][TOOLS-CDC]:add flink Command line support

flink-connector-oceanbase-tools-cdc/pom.xml

@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Copyright 2024 OceanBase.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+  http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>com.oceanbase</groupId>
+        <artifactId>flink-connector-oceanbase-parent</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>flink-connector-oceanbase-tools-cdc</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.oceanbase</groupId>
+            <artifactId>flink-connector-oceanbase</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-connector-mysql-cdc</artifactId>
+            <version>${flink.sql.cdc.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-connector-oracle-cdc</artifactId>
+            <version>${flink.sql.cdc.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-connector-postgres-cdc</artifactId>
+            <version>${flink.sql.cdc.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-connector-sqlserver-cdc</artifactId>
+            <version>${flink.sql.cdc.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-connector-db2-cdc</artifactId>
+            <version>${flink.sql.cdc.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>com.oracle.ojdbc</groupId>
+            <artifactId>ojdbc8</artifactId>
+            <version>${ojdbc.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+</project>

flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseToolsConnectProvider.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2024 OceanBase.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.oceanbase.connector.flink.connection;
+
+import com.oceanbase.connector.flink.OceanBaseConnectorOptions;
+import com.oceanbase.connector.flink.tools.catalog.TableSchema;
+import com.oceanbase.connector.flink.utils.OceanBaseToolsJdbcUtils;
+
+public class OceanBaseToolsConnectProvider extends OceanBaseConnectionProvider {
+
+    public OceanBaseToolsConnectProvider(OceanBaseConnectorOptions options) {
+        super(options);
+    }
+
+    public boolean databaseExists(String database) {
+        return OceanBaseToolsJdbcUtils.databaseExists(database, this::getConnection);
+    }
+
+    public void createDatabase(String database) {
+        OceanBaseToolsJdbcUtils.createDatabase(database, this::getConnection);
+    }
+
+    public boolean tableExists(String database, String table) {
+        return OceanBaseToolsJdbcUtils.tableExists(database, table, this::getConnection);
+    }
+
+    public void createTable(TableSchema schema) {
+        OceanBaseToolsJdbcUtils.createTable(schema, this::getConnection);
+    }
+}

flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2024 OceanBase.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.oceanbase.connector.flink.table;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.Date;
+import java.sql.Time;
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class OceanBaseJsonSerializationSchema extends AbstractRecordSerializationSchema<String> {
+
+    private static final long serialVersionUID = 1L;
+    private static final Logger log =
+            LoggerFactory.getLogger(OceanBaseJsonSerializationSchema.class);
+
+    private final TableInfo tableInfo;
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+
+    public OceanBaseJsonSerializationSchema(TableInfo tableInfo) {
+        this.tableInfo = tableInfo;
+    }
+
+    @Override
+    public Record serialize(String rowDataStr) {
+        try {
+            JsonNode rowDataNode = objectMapper.readTree(rowDataStr);
+            DataChangeRecord.Type type;
+            String op = rowDataNode.path("op").asText();
+            if ("r".equals(op) || "c".equals(op)) {
+                type = DataChangeRecord.Type.UPSERT;
+            } else if ("d".equals(op)) {
+                type = DataChangeRecord.Type.DELETE;
+            } else {
+                throw new IllegalArgumentException("Unknown operation type: " + op);
+            }
+            int size = tableInfo.getFieldNames().size();
+            Object[] values = new Object[size];
+            for (int i = 0; i < size; i++) {
+                String fieldName = tableInfo.getFieldNames().get(i);
+                JsonNode fieldNode = rowDataNode.path("after").path(fieldName);
+                values[i] = objectMapper.convertValue(fieldNode, new TypeReference<Object>() {});
+            }
+
+            return new DataChangeRecord(tableInfo, type, values);
+        } catch (IOException e) {
+            log.error("Failed to parse rowData JSON: {}", rowDataStr, e);
+            return null;
+        }
+    }
+
+    @Override
+    protected SerializationRuntimeConverter createNotNullConverter(LogicalType type) {
+        switch (type.getTypeRoot()) {
+            case BOOLEAN:
+            case TINYINT:
+            case SMALLINT:
+            case INTEGER:
+            case INTERVAL_YEAR_MONTH:
+            case BIGINT:
+            case INTERVAL_DAY_TIME:
+            case FLOAT:
+            case DOUBLE:
+            case BINARY:
+            case VARBINARY:
+                return data -> data;
+            case CHAR:
+            case VARCHAR:
+                return Object::toString;
+            case DATE:
+                return data -> Date.valueOf(LocalDate.ofEpochDay((int) data));
+            case TIME_WITHOUT_TIME_ZONE:
+                return data -> Time.valueOf(LocalTime.ofNanoOfDay((int) data * 1_000_000L));
+            case TIMESTAMP_WITHOUT_TIME_ZONE:
+                return data -> ((TimestampData) data).toTimestamp();
+            case DECIMAL:
+                return data -> ((DecimalData) data).toBigDecimal();
+            case ARRAY:
+                return data -> {
+                    ArrayData arrayData = (ArrayData) data;
+                    return IntStream.range(0, arrayData.size())
+                            .mapToObj(i -> arrayData.getString(i).toString())
+                            .collect(Collectors.joining(","));
+                };
+            default:
+                throw new UnsupportedOperationException("Unsupported type:" + type);
+        }
+    }
+}

flink-connector-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/FieldSchema.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2024 OceanBase.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.oceanbase.connector.flink.tools.catalog;
+
+public class FieldSchema {
+    private String name;
+    private String typeString;
+    private String defaultValue;
+    private String comment;
+
+    public Boolean getNullable() {
+        return nullable;
+    }
+
+    public void setNullable(Boolean nullable) {
+        this.nullable = nullable;
+    }
+
+    private Boolean nullable;
+
+    public FieldSchema() {}
+
+    public FieldSchema(String name, String typeString, String comment, Boolean nullable) {
+        this.name = name;
+        this.typeString = typeString;
+        this.comment = comment;
+        this.nullable = nullable;
+    }
+
+    public FieldSchema(
+            String name, String typeString, String defaultValue, String comment, Boolean nullable) {
+        this.name = name;
+        this.typeString = typeString;
+        this.defaultValue = defaultValue;
+        this.comment = comment;
+        this.nullable = nullable;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getTypeString() {
+        return typeString;
+    }
+
+    public void setTypeString(String typeString) {
+        this.typeString = typeString;
+    }
+
+    public String getComment() {
+        return comment;
+    }
+
+    public void setComment(String comment) {
+        this.comment = comment;
+    }
+
+    public String getDefaultValue() {
+        return defaultValue;
+    }
+
+    public void setDefaultValue(String defaultValue) {
+        this.defaultValue = defaultValue;
+    }
+
+    @Override
+    public String toString() {
+        return "FieldSchema{"
+                + "name='"
+                + name
+                + '\''
+                + ", typeString='"
+                + typeString
+                + '\''
+                + ", defaultValue='"
+                + defaultValue
+                + '\''
+                + ", comment='"
+                + comment
+                + '\''
+                + ", nullable='"
+                + nullable
+                + '\''
+                + '}';
+    }
+}