This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c408d50f6 [cdc] The newly added table cannot obtain the metadata
column in the database cdc action. (#2784)
c408d50f6 is described below
commit c408d50f6432e8aef8d16e9e61a628700ab90a01
Author: Kerwin <[email protected]>
AuthorDate: Thu Jan 25 09:51:43 2024 +0800
[cdc] The newly added table cannot obtain the metadata column in the
database cdc action. (#2784)
---
.../shortcodes/generated/postgres_sync_table.html | 2 +-
.../flink/action/cdc/SyncDatabaseActionBase.java | 3 +-
.../flink/sink/cdc/NewTableSchemaBuilder.java | 39 +++++++++++++++++-----
.../cdc/mysql/MySqlSyncDatabaseActionITCase.java | 14 ++++++++
.../postgres/PostgresSyncTableActionITCase.java | 2 +-
5 files changed, 49 insertions(+), 11 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/postgres_sync_table.html
b/docs/layouts/shortcodes/generated/postgres_sync_table.html
index ecb085375..407a91f24 100644
--- a/docs/layouts/shortcodes/generated/postgres_sync_table.html
+++ b/docs/layouts/shortcodes/generated/postgres_sync_table.html
@@ -60,7 +60,7 @@ under the License.
</tr>
<tr>
<td><h5>--metadata_column</h5></td>
- <td>--metadata_column is used to specify which metadata columns to
include in the output schema of the connector. Metadata columns provide
additional information related to the source data, for example:
--metadata_column table_name --metadata_column database_name --metadata_column
schema_name --metadata_column op_ts. See its <a
href="https://ververica.github.io/flink-cdc-connectors/master/content/connectors/postgres-cdc.html#available-metadata";>document</a>
for a complete list of [...]
+ <td>--metadata_column is used to specify which metadata columns to
include in the output schema of the connector. Metadata columns provide
additional information related to the source data, for example:
--metadata_column table_name,database_name,schema_name,op_ts. See its <a
href="https://ververica.github.io/flink-cdc-connectors/master/content/connectors/postgres-cdc.html#available-metadata";>document</a>
for a complete list of available metadata.</td>
</tr>
<tr>
<td><h5>--postgres_conf</h5></td>
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
index 74c2f6fc7..ecfca9a17 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
@@ -117,7 +117,8 @@ public abstract class SyncDatabaseActionBase extends
SynchronizationActionBase {
@Override
protected EventParser.Factory<RichCdcMultiplexRecord>
buildEventParserFactory() {
- NewTableSchemaBuilder schemaBuilder = new
NewTableSchemaBuilder(tableConfig, caseSensitive);
+ NewTableSchemaBuilder schemaBuilder =
+ new NewTableSchemaBuilder(tableConfig, caseSensitive,
metadataConverters);
Pattern includingPattern = Pattern.compile(includingTables);
Pattern excludingPattern =
excludingTables == null ? null :
Pattern.compile(excludingTables);
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
index c7352eb26..18324abed 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java
@@ -18,27 +18,35 @@
package org.apache.paimon.flink.sink.cdc;
+import org.apache.paimon.flink.action.cdc.CdcMetadataConverter;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.types.DataType;
import java.io.Serializable;
-import java.util.LinkedHashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.columnCaseConvertAndDuplicateCheck;
import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.columnDuplicateErrMsg;
import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.listCaseConvert;
-import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.mapKeyCaseConvert;
/** Build schema for new table found in database synchronization. */
public class NewTableSchemaBuilder implements Serializable {
private final Map<String, String> tableConfig;
private final boolean caseSensitive;
+ private final CdcMetadataConverter[] metadataConverters;
- public NewTableSchemaBuilder(Map<String, String> tableConfig, boolean
caseSensitive) {
+ public NewTableSchemaBuilder(
+ Map<String, String> tableConfig,
+ boolean caseSensitive,
+ CdcMetadataConverter[] metadataConverters) {
this.tableConfig = tableConfig;
this.caseSensitive = caseSensitive;
+ this.metadataConverters = metadataConverters;
}
public Optional<Schema> build(RichCdcMultiplexRecord record) {
@@ -47,12 +55,27 @@ public class NewTableSchemaBuilder implements Serializable {
String tableName = record.tableName();
tableName = tableName == null ? "UNKNOWN" : tableName;
- LinkedHashMap<String, DataType> fieldTypes =
- mapKeyCaseConvert(
- record.fieldTypes(), caseSensitive,
columnDuplicateErrMsg(tableName));
- for (Map.Entry<String, DataType> entry : fieldTypes.entrySet()) {
- builder.column(entry.getKey(), entry.getValue());
+ // fields
+ Set<String> existedFields = new HashSet<>();
+ Function<String, String> columnDuplicateErrMsg =
columnDuplicateErrMsg(tableName);
+
+ for (Map.Entry<String, DataType> entry :
record.fieldTypes().entrySet()) {
+ String fieldName =
+ columnCaseConvertAndDuplicateCheck(
+ entry.getKey(), existedFields, caseSensitive,
columnDuplicateErrMsg);
+
+ builder.column(fieldName, entry.getValue());
+ }
+
+ for (CdcMetadataConverter metadataConverter : metadataConverters) {
+ String metadataColumnName =
+ columnCaseConvertAndDuplicateCheck(
+ metadataConverter.columnName(),
+ existedFields,
+ caseSensitive,
+ columnDuplicateErrMsg);
+ builder.column(metadataColumnName, metadataConverter.dataType());
}
builder.primaryKey(listCaseConvert(record.primaryKeys(),
caseSensitive));
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
index 5074e759d..84169ffe0 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
@@ -1314,6 +1314,20 @@ public class MySqlSyncDatabaseActionITCase extends
MySqlActionITCaseBase {
table,
rowType,
Collections.singletonList("k"));
+
+ // test newly created table
+ if (mode == COMBINED) {
+ statement.execute("USE " + "metadata");
+ statement.executeUpdate("CREATE TABLE t3 (k INT, v1
VARCHAR(10), PRIMARY KEY (k))");
+ statement.executeUpdate("INSERT INTO t3 VALUES (1, 'Hi')");
+ waitingTables("t3");
+ table = getFileStoreTable("t3");
+ waitForResult(
+ Collections.singletonList("+I[1, Hi, t3, metadata]"),
+ table,
+ rowType,
+ Collections.singletonList("k"));
+ }
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
index a6a34c9bf..cbd6c7bd0 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
@@ -508,7 +508,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
.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
