This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 2c401acbc [cdc] Fix bug that computed columns work abnormally with
case-insensitive (#2485)
2c401acbc is described below
commit 2c401acbc357b5432d9d60f4e951b5dfb7a74bb5
Author: yuzelin <[email protected]>
AuthorDate: Tue Dec 12 16:36:46 2023 +0800
[cdc] Fix bug that computed columns work abnormally with case-insensitive
(#2485)
---
.../flink/action/cdc/CdcActionCommonUtils.java | 62 +++++---
.../flink/action/cdc/ComputedColumnUtils.java | 13 +-
.../cdc/MessageQueueSyncTableActionBase.java | 2 +
.../flink/action/cdc/SyncTableActionBase.java | 10 +-
.../flink/action/cdc/format/RecordParser.java | 23 +--
.../action/cdc/mongodb/MongoDBSyncTableAction.java | 2 +-
.../action/cdc/mongodb/MongodbSchemaUtils.java | 27 +---
.../mongodb/strategy/Mongo4VersionStrategy.java | 34 ++---
.../cdc/mongodb/strategy/MongoVersionStrategy.java | 43 ++----
.../flink/action/cdc/mysql/MySqlActionUtils.java | 6 +-
.../action/cdc/mysql/MySqlSyncDatabaseAction.java | 6 +-
.../action/cdc/mysql/MySqlSyncTableAction.java | 6 +-
.../action/cdc/mysql/schema/MySqlSchemaUtils.java | 17 +--
.../flink/action/cdc/CdcActionITCaseBase.java | 164 +++++++++++++++++----
.../action/cdc/kafka/KafkaActionITCaseBase.java | 66 +--------
.../cdc/kafka/KafkaCanalSyncTableActionITCase.java | 57 +++++++
.../cdc/mongodb/MongoDBActionITCaseBase.java | 75 +---------
.../cdc/mongodb/MongoDBSyncTableActionITCase.java | 35 ++++-
.../action/cdc/mongodb/MongodbSchemaITCase.java | 15 +-
.../action/cdc/mysql/MySqlActionITCaseBase.java | 61 +-------
.../cdc/mysql/MySqlSyncTableActionITCase.java | 61 +++++++-
.../action/cdc/pulsar/PulsarActionITCaseBase.java | 66 +--------
.../canal/table/computedcolumn/canal-data-2.txt | 19 +++
.../mongodb/table/computedcolumn/test-table-2.js | 24 +++
.../src/test/resources/mysql/sync_table_setup.sql | 27 ++++
25 files changed, 477 insertions(+), 444 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
index d54b1ea36..b3fdab390 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
@@ -32,6 +32,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -162,12 +163,14 @@ public class CdcActionCommonUtils {
}
public static Schema buildPaimonSchema(
+ String tableName,
List<String> specifiedPartitionKeys,
List<String> specifiedPrimaryKeys,
List<ComputedColumn> computedColumns,
Map<String, String> tableConfig,
Schema sourceSchema,
CdcMetadataConverter[] metadataConverters,
+ boolean caseSensitive,
boolean requirePrimaryKeys) {
Schema.Builder builder = Schema.newBuilder();
@@ -176,40 +179,53 @@ public class CdcActionCommonUtils {
builder.options(sourceSchema.options());
// fields
- sourceSchema
- .fields()
- .forEach(
- dataField ->
- builder.column(
- dataField.name(),
- dataField.type(),
- dataField.description()));
+ Set<String> existedFields = new HashSet<>();
+ Function<String, String> columnDuplicateErrMsg =
columnDuplicateErrMsg(tableName);
+
+ for (DataField field : sourceSchema.fields()) {
+ String fieldName =
+ columnCaseConvertAndDuplicateCheck(
+ field.name(), existedFields, caseSensitive,
columnDuplicateErrMsg);
+ builder.column(fieldName, field.type(), field.description());
+ }
for (ComputedColumn computedColumn : computedColumns) {
- builder.column(computedColumn.columnName(),
computedColumn.columnType());
+ String computedColumnName =
+ columnCaseConvertAndDuplicateCheck(
+ computedColumn.columnName(),
+ existedFields,
+ caseSensitive,
+ columnDuplicateErrMsg);
+ builder.column(computedColumnName, computedColumn.columnType());
}
for (CdcMetadataConverter metadataConverter : metadataConverters) {
- builder.column(metadataConverter.columnName(),
metadataConverter.dataType());
+ String metadataColumnName =
+ columnCaseConvertAndDuplicateCheck(
+ metadataConverter.columnName(),
+ existedFields,
+ caseSensitive,
+ columnDuplicateErrMsg);
+ builder.column(metadataColumnName, metadataConverter.dataType());
}
// primary keys
if (!specifiedPrimaryKeys.isEmpty()) {
- Map<String, Integer> sourceColumns =
- sourceSchema.fields().stream()
- .collect(Collectors.toMap(DataField::name, entity
-> 1));
+ Set<String> sourceColumns =
+
sourceSchema.fields().stream().map(DataField::name).collect(Collectors.toSet());
+ sourceColumns.addAll(
+ computedColumns.stream()
+ .map(ComputedColumn::columnName)
+ .collect(Collectors.toSet()));
for (String key : specifiedPrimaryKeys) {
- if (!sourceColumns.containsKey(key)
- && computedColumns.stream().noneMatch(c ->
c.columnName().equals(key))) {
- throw new IllegalArgumentException(
- "Specified primary key '"
- + key
- + "' does not exist in source tables or
computed columns.");
- }
+ checkArgument(
+ sourceColumns.contains(key),
+ "Specified primary key '%s' does not exist in source
tables or computed columns.",
+ key);
}
- builder.primaryKey(specifiedPrimaryKeys);
+ builder.primaryKey(listCaseConvert(specifiedPrimaryKeys,
caseSensitive));
} else if (!sourceSchema.primaryKeys().isEmpty()) {
- builder.primaryKey(sourceSchema.primaryKeys());
+ builder.primaryKey(listCaseConvert(sourceSchema.primaryKeys(),
caseSensitive));
} else if (requirePrimaryKeys) {
throw new IllegalArgumentException(
"Primary keys are not specified. "
@@ -219,7 +235,7 @@ public class CdcActionCommonUtils {
// partition keys
if (!specifiedPartitionKeys.isEmpty()) {
- builder.partitionKeys(specifiedPartitionKeys);
+ builder.partitionKeys(listCaseConvert(specifiedPartitionKeys,
caseSensitive));
}
// comment
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtils.java
index ed3c48d5f..70b1898c0 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtils.java
@@ -21,6 +21,7 @@ package org.apache.paimon.flink.action.cdc;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.StringUtils;
import java.util.ArrayList;
import java.util.Arrays;
@@ -35,6 +36,12 @@ public class ComputedColumnUtils {
public static List<ComputedColumn> buildComputedColumns(
List<String> computedColumnArgs, List<DataField> physicFields) {
+ return buildComputedColumns(computedColumnArgs, physicFields, true);
+ }
+
+ /** The caseSensitive only affects check. We don't change field names at
building phase. */
+ public static List<ComputedColumn> buildComputedColumns(
+ List<String> computedColumnArgs, List<DataField> physicFields,
boolean caseSensitive) {
Map<String, DataType> typeMapping =
physicFields.stream()
.collect(
@@ -67,11 +74,13 @@ public class ComputedColumnUtils {
String fieldReference = args[0].trim();
String[] literals =
Arrays.stream(args).skip(1).map(String::trim).toArray(String[]::new);
+ String fieldReferenceCheckForm =
+ StringUtils.caseSensitiveConversion(fieldReference,
caseSensitive);
checkArgument(
- typeMapping.containsKey(fieldReference),
+ typeMapping.containsKey(fieldReferenceCheckForm),
String.format(
"Referenced field '%s' is not in given fields:
%s.",
- fieldReference, typeMapping.keySet()));
+ fieldReferenceCheckForm, typeMapping.keySet()));
computedColumns.add(
new ComputedColumn(
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
index c3c6d49ff..fb505bb8b 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
@@ -72,12 +72,14 @@ public abstract class MessageQueueSyncTableActionBase
extends SyncTableActionBas
@Override
protected Schema buildPaimonSchema(Schema retrievedSchema) {
return CdcActionCommonUtils.buildPaimonSchema(
+ table,
partitionKeys,
primaryKeys,
computedColumns,
tableConfig,
retrievedSchema,
metadataConverters,
+ caseSensitive,
false);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
index 317304b28..a5aab32d0 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
@@ -100,12 +100,14 @@ public abstract class SyncTableActionBase extends
SynchronizationActionBase {
protected Schema buildPaimonSchema(Schema retrievedSchema) {
return CdcActionCommonUtils.buildPaimonSchema(
+ table,
partitionKeys,
primaryKeys,
computedColumns,
tableConfig,
retrievedSchema,
metadataConverters,
+ caseSensitive,
true);
}
@@ -113,8 +115,6 @@ public abstract class SyncTableActionBase extends
SynchronizationActionBase {
protected void validateCaseSensitivity() {
AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database",
database);
AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table", table);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Partition
keys", partitionKeys);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Primary keys",
primaryKeys);
}
@Override
@@ -124,7 +124,6 @@ public abstract class SyncTableActionBase extends
SynchronizationActionBase {
if (catalog.tableExists(identifier)) {
fileStoreTable = (FileStoreTable)
catalog.getTable(identifier).copy(tableConfig);
try {
- // TODO: test case insensitive with computed columns
Schema retrievedSchema = retrieveSchema();
computedColumns =
buildComputedColumns(computedColumnArgs,
retrievedSchema.fields());
@@ -138,7 +137,10 @@ public abstract class SyncTableActionBase extends
SynchronizationActionBase {
+ "the Paimon table has defined all the
argument columns used for computed columns.");
// schema evolution will add the computed columns
computedColumns =
- buildComputedColumns(computedColumnArgs,
fileStoreTable.schema().fields());
+ buildComputedColumns(
+ computedColumnArgs,
+ fileStoreTable.schema().fields(),
+ caseSensitive);
// check partition keys and primary keys in case that user
specified them
checkConstraints();
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
index 261d26308..0593fc1e9 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
@@ -43,16 +43,12 @@ import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
-import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
-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;
@@ -97,22 +93,9 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
return null;
}
- LinkedHashMap<String, DataType> paimonFieldTypes =
extractPaimonFieldTypes();
Schema.Builder builder = Schema.newBuilder();
- Set<String> existedFields = new HashSet<>();
- Function<String, String> columnDuplicateErrMsg =
columnDuplicateErrMsg(getTableName());
- for (Map.Entry<String, DataType> entry :
paimonFieldTypes.entrySet()) {
- builder.column(
- columnCaseConvertAndDuplicateCheck(
- entry.getKey(),
- existedFields,
- caseSensitive,
- columnDuplicateErrMsg),
- entry.getValue());
- }
-
- builder.primaryKey(listCaseConvert(extractPrimaryKeys(),
caseSensitive));
-
+ extractPaimonFieldTypes().forEach(builder::column);
+ builder.primaryKey(extractPrimaryKeys());
return builder.build();
} catch (Exception e) {
logInvalidJsonString(record);
@@ -145,7 +128,7 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
}
@Override
- public void flatMap(String value, Collector<RichCdcMultiplexRecord> out)
throws Exception {
+ public void flatMap(String value, Collector<RichCdcMultiplexRecord> out) {
try {
setRoot(value);
extractRecords().forEach(out::collect);
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
index 7ee875f8d..32596863a 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
@@ -65,7 +65,7 @@ public class MongoDBSyncTableAction extends
SyncTableActionBase {
@Override
protected Schema retrieveSchema() {
- return MongodbSchemaUtils.getMongodbSchema(cdcSourceConfig,
caseSensitive);
+ return MongodbSchemaUtils.getMongodbSchema(cdcSourceConfig);
}
@Override
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaUtils.java
index f81bdfaf7..28ad91ba0 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaUtils.java
@@ -37,16 +37,11 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
import java.util.Objects;
-import java.util.Set;
-import java.util.function.Function;
import static
com.ververica.cdc.connectors.mongodb.internal.MongoDBEnvelope.encodeValue;
import static org.apache.flink.util.Preconditions.checkNotNull;
-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.mongodb.MongoDBActionUtils.FIELD_NAME;
import static
org.apache.paimon.flink.action.cdc.mongodb.MongoDBActionUtils.START_MODE;
@@ -77,7 +72,7 @@ public class MongodbSchemaUtils {
* name, and optionally, the username and password for authentication. For
the SPECIFIED mode,
* the field names should also be specified in the configuration.
*/
- public static Schema getMongodbSchema(Configuration mongodbConfig, boolean
caseSensitive) {
+ public static Schema getMongodbSchema(Configuration mongodbConfig) {
SchemaAcquisitionMode mode = getModeFromConfig(mongodbConfig);
String databaseName =
Objects.requireNonNull(
@@ -95,7 +90,7 @@ public class MongodbSchemaUtils {
mongodbConfig.get(FIELD_NAME), "Field
names cannot be null")
.split(",");
- return createMongodbSchema(collectionName, columnNames,
caseSensitive);
+ return createMongodbSchema(columnNames);
case DYNAMIC:
String hosts =
Objects.requireNonNull(
@@ -126,8 +121,7 @@ public class MongodbSchemaUtils {
"No documents in collection to infer schema");
}
- return createMongodbSchema(
- collectionName, getColumnNames(firstDocument),
caseSensitive);
+ return createMongodbSchema(getColumnNames(firstDocument));
} catch (Exception e) {
throw new RuntimeException(
"Failed to create schema from MongoDB collection",
e);
@@ -166,21 +160,14 @@ public class MongodbSchemaUtils {
return document != null ? new ArrayList<>(document.keySet()) :
Collections.emptyList();
}
- private static Schema createMongodbSchema(
- String collectionName, String[] columnNames, boolean
caseSensitive) {
- return createMongodbSchema(collectionName, Arrays.asList(columnNames),
caseSensitive);
+ private static Schema createMongodbSchema(String[] columnNames) {
+ return createMongodbSchema(Arrays.asList(columnNames));
}
- private static Schema createMongodbSchema(
- String collectionName, List<String> columnNames, boolean
caseSensitive) {
+ private static Schema createMongodbSchema(List<String> columnNames) {
Schema.Builder builder = Schema.newBuilder();
- Set<String> existedFields = new HashSet<>();
- Function<String, String> columnDuplicateErrMsg =
columnDuplicateErrMsg(collectionName);
for (String column : columnNames) {
- builder.column(
- columnCaseConvertAndDuplicateCheck(
- column, existedFields, caseSensitive,
columnDuplicateErrMsg),
- DataTypes.STRING());
+ builder.column(column, DataTypes.STRING());
}
builder.primaryKey(ID_FIELD);
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
index 53cd8a309..395ec939e 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
@@ -34,6 +34,10 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.columnDuplicateErrMsg;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.mapKeyCaseConvert;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.recordKeyDuplicateErrMsg;
+
/**
* Implementation class for extracting records from MongoDB versions greater
than 4.x and less than
* 6.x.
@@ -93,22 +97,21 @@ public class Mongo4VersionStrategy implements
MongoVersionStrategy {
private List<RichCdcMultiplexRecord> handleOperation(
String op, JsonNode fullDocument, JsonNode documentKey) throws
JsonProcessingException {
List<RichCdcMultiplexRecord> records = new ArrayList<>();
- LinkedHashMap<String, DataType> paimonFieldTypes = new
LinkedHashMap<>();
switch (op) {
case OP_INSERT:
- records.add(processRecord(fullDocument, paimonFieldTypes,
RowKind.INSERT));
+ records.add(processRecord(fullDocument, RowKind.INSERT));
break;
case OP_REPLACE:
case OP_UPDATE:
// Before version 6.0 of MongoDB, it was not possible to
obtain 'Update Before'
// information. Therefore, data is first deleted using the
primary key '_id', and
// then inserted.
- records.add(processRecord(documentKey, paimonFieldTypes,
RowKind.DELETE));
- records.add(processRecord(fullDocument, paimonFieldTypes,
RowKind.INSERT));
+ records.add(processRecord(documentKey, RowKind.DELETE));
+ records.add(processRecord(fullDocument, RowKind.INSERT));
break;
case OP_DELETE:
- records.add(processRecord(documentKey, paimonFieldTypes,
RowKind.DELETE));
+ records.add(processRecord(documentKey, RowKind.DELETE));
break;
default:
throw new UnsupportedOperationException("Unknown record type:
" + op);
@@ -121,24 +124,21 @@ public class Mongo4VersionStrategy implements
MongoVersionStrategy {
* RichCdcMultiplexRecord object.
*
* @param fullDocument the JSON node containing the full document to be
processed.
- * @param paimonFieldTypes a LinkedHashMap containing the field types to
be used in the
- * processing.
* @param rowKind the kind of row to be processed (e.g., insert, update,
delete).
* @throws JsonProcessingException if there is an error in processing the
JSON document.
* @return a RichCdcMultiplexRecord object that contains the processed
record information.
*/
- private RichCdcMultiplexRecord processRecord(
- JsonNode fullDocument,
- LinkedHashMap<String, DataType> paimonFieldTypes,
- RowKind rowKind)
+ private RichCdcMultiplexRecord processRecord(JsonNode fullDocument,
RowKind rowKind)
throws JsonProcessingException {
+ LinkedHashMap<String, DataType> paimonFieldTypes = new
LinkedHashMap<>();
Map<String, String> record =
- getExtractRow(
- fullDocument,
- paimonFieldTypes,
- caseSensitive,
- computedColumns,
- mongodbConfig);
+ getExtractRow(fullDocument, paimonFieldTypes, computedColumns,
mongodbConfig);
+
+ record = mapKeyCaseConvert(record, caseSensitive,
recordKeyDuplicateErrMsg(record));
+ paimonFieldTypes =
+ mapKeyCaseConvert(
+ paimonFieldTypes, caseSensitive,
columnDuplicateErrMsg(collection));
+
return new RichCdcMultiplexRecord(
databaseName,
collection,
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
index d4388d7bd..24d2be85e 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
@@ -39,8 +39,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
-import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.mapKeyCaseConvert;
-import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.recordKeyDuplicateErrMsg;
import static
org.apache.paimon.flink.action.cdc.mongodb.MongoDBActionUtils.DEFAULT_ID_GENERATION;
import static
org.apache.paimon.flink.action.cdc.mongodb.MongoDBActionUtils.FIELD_NAME;
import static
org.apache.paimon.flink.action.cdc.mongodb.MongoDBActionUtils.PARSER_PATH;
@@ -75,7 +73,6 @@ public interface MongoVersionStrategy {
*
* @param jsonNode The JsonNode representing the MongoDB document.
* @param paimonFieldTypes A map to store the field types.
- * @param caseSensitive Flag indicating if the extraction should be
case-sensitive.
* @param mongodbConfig Configuration for the MongoDB connection.
* @return A map representing the extracted row.
* @throws JsonProcessingException If there's an error during JSON
processing.
@@ -83,7 +80,6 @@ public interface MongoVersionStrategy {
default Map<String, String> getExtractRow(
JsonNode jsonNode,
LinkedHashMap<String, DataType> paimonFieldTypes,
- boolean caseSensitive,
List<ComputedColumn> computedColumns,
Configuration mongodbConfig)
throws JsonProcessingException {
@@ -100,39 +96,28 @@ public interface MongoVersionStrategy {
mongodbConfig.getBoolean(DEFAULT_ID_GENERATION)
? objectNode.set(ID_FIELD, idNode.get(OID_FIELD))
: objectNode;
- Map<String, String> row;
switch (mode) {
case SPECIFIED:
- row =
- parseFieldsFromJsonRecord(
- document.toString(),
- mongodbConfig.getString(PARSER_PATH),
- mongodbConfig.getString(FIELD_NAME),
- computedColumns,
- paimonFieldTypes);
- break;
+ return parseFieldsFromJsonRecord(
+ document.toString(),
+ mongodbConfig.getString(PARSER_PATH),
+ mongodbConfig.getString(FIELD_NAME),
+ computedColumns,
+ paimonFieldTypes);
case DYNAMIC:
- row =
- parseAndTypeJsonRow(
- document.toString(),
- paimonFieldTypes,
- computedColumns,
- caseSensitive);
- break;
+ return parseAndTypeJsonRow(document.toString(),
paimonFieldTypes, computedColumns);
default:
throw new RuntimeException("Unsupported extraction mode: " +
mode);
}
- return mapKeyCaseConvert(row, caseSensitive,
recordKeyDuplicateErrMsg(row));
}
/** Parses and types a JSON row based on the given parameters. */
default Map<String, String> parseAndTypeJsonRow(
String evaluate,
LinkedHashMap<String, DataType> paimonFieldTypes,
- List<ComputedColumn> computedColumns,
- boolean caseSensitive) {
+ List<ComputedColumn> computedColumns) {
Map<String, String> parsedRow = JsonSerdeUtil.parseJsonMap(evaluate,
String.class);
- return processParsedData(parsedRow, paimonFieldTypes, computedColumns,
caseSensitive);
+ return processParsedData(parsedRow, paimonFieldTypes, computedColumns);
}
/** Parses fields from a JSON record based on the given parameters. */
@@ -151,23 +136,21 @@ public interface MongoVersionStrategy {
parsedRow.put(columnNames[i],
Optional.ofNullable(evaluate).orElse("{}"));
}
- return processParsedData(parsedRow, fieldTypes, computedColumns,
false);
+ return processParsedData(parsedRow, fieldTypes, computedColumns);
}
/** Processes the parsed data to generate the result map and update field
types. */
static Map<String, String> processParsedData(
Map<String, String> parsedRow,
LinkedHashMap<String, DataType> fieldTypes,
- List<ComputedColumn> computedColumns,
- boolean caseSensitive) {
+ List<ComputedColumn> computedColumns) {
int initialCapacity = parsedRow.size() + computedColumns.size();
Map<String, String> resultMap = new HashMap<>(initialCapacity);
parsedRow.forEach(
(column, value) -> {
- String key = caseSensitive ? column : column.toLowerCase();
- fieldTypes.putIfAbsent(key, DataTypes.STRING());
- resultMap.put(key, value);
+ fieldTypes.put(column, DataTypes.STRING());
+ resultMap.put(column, value);
});
computedColumns.forEach(
computedColumn -> {
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
index 9cfa8eca8..b44291fc4 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
@@ -88,8 +88,7 @@ public class MySqlActionUtils {
Configuration mySqlConfig,
Predicate<String> monitorTablePredication,
List<Identifier> excludedTables,
- TypeMapping typeMapping,
- boolean caseSensitive)
+ TypeMapping typeMapping)
throws Exception {
Pattern databasePattern =
Pattern.compile(mySqlConfig.get(MySqlSourceOptions.DATABASE_NAME));
@@ -113,8 +112,7 @@ public class MySqlActionUtils {
databaseName,
tableName,
tableComment,
- typeMapping,
- caseSensitive);
+ typeMapping);
Identifier identifier =
Identifier.create(databaseName, tableName);
if (monitorTablePredication.test(tableName)) {
mySqlSchemasInfo.addSchema(identifier,
schema);
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index 67bf7927b..ca7678b41 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -122,7 +122,6 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
@Override
protected void beforeBuildingSourceSink() throws Exception {
- boolean caseSensitive = catalog.caseSensitive();
Pattern includingPattern = Pattern.compile(includingTables);
Pattern excludingPattern =
excludingTables == null ? null :
Pattern.compile(excludingTables);
@@ -132,8 +131,7 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
tableName ->
shouldMonitorTable(tableName,
includingPattern, excludingPattern),
excludedTables,
- typeMapping,
- caseSensitive);
+ typeMapping);
logNonPkTables(mySqlSchemasInfo.nonPkTables());
List<MySqlTableInfo> mySqlTableInfos =
mySqlSchemasInfo.toMySqlTableInfos(mergeShards);
@@ -153,12 +151,14 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
FileStoreTable table;
Schema fromMySql =
CdcActionCommonUtils.buildPaimonSchema(
+ identifier.getFullName(),
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
tableConfig,
tableInfo.schema(),
metadataConverters,
+ caseSensitive,
true);
try {
table = (FileStoreTable) catalog.getTable(identifier);
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
index b7510d08a..f8551a48b 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -99,11 +99,7 @@ public class MySqlSyncTableAction extends
SyncTableActionBase {
protected Schema retrieveSchema() throws Exception {
this.mySqlSchemasInfo =
MySqlActionUtils.getMySqlTableInfos(
- cdcSourceConfig,
- monitorTablePredication(),
- new ArrayList<>(),
- typeMapping,
- caseSensitive);
+ cdcSourceConfig, monitorTablePredication(), new
ArrayList<>(), typeMapping);
validateMySqlTableInfos(mySqlSchemasInfo);
MySqlTableInfo tableInfo = mySqlSchemasInfo.mergeAll();
return tableInfo.schema();
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/schema/MySqlSchemaUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/schema/MySqlSchemaUtils.java
index 45762d550..a9eb2c995 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/schema/MySqlSchemaUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/schema/MySqlSchemaUtils.java
@@ -32,17 +32,11 @@ import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Objects;
-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.TypeMapping.TypeMappingMode.TO_NULLABLE;
-import static org.apache.paimon.utils.StringUtils.caseSensitiveConversion;
/** Utility class to load MySQL table schema with JDBC. */
public class MySqlSchemaUtils {
@@ -54,13 +48,10 @@ public class MySqlSchemaUtils {
String databaseName,
String tableName,
String tableComment,
- TypeMapping typeMapping,
- boolean caseSensitive)
+ TypeMapping typeMapping)
throws SQLException {
Schema.Builder builder = Schema.newBuilder();
try (ResultSet rs = metaData.getColumns(databaseName, null, tableName,
null)) {
- Set<String> existedFields = new HashSet<>();
- Function<String, String> columnDuplicateErrMsg =
columnDuplicateErrMsg(tableName);
while (rs.next()) {
String fieldName = rs.getString("COLUMN_NAME");
String fieldType = rs.getString("TYPE_NAME");
@@ -82,10 +73,6 @@ public class MySqlSchemaUtils {
MySqlTypeUtils.toDataType(fieldType, precision, scale,
typeMapping)
.copy(isNullable);
- fieldName =
- columnCaseConvertAndDuplicateCheck(
- fieldName, existedFields, caseSensitive,
columnDuplicateErrMsg);
-
builder.column(fieldName, paimonType, fieldComment);
}
}
@@ -95,7 +82,7 @@ public class MySqlSchemaUtils {
try (ResultSet rs = metaData.getPrimaryKeys(databaseName, null,
tableName)) {
while (rs.next()) {
String fieldName = rs.getString("COLUMN_NAME");
- primaryKeys.add(caseSensitiveConversion(fieldName,
caseSensitive));
+ primaryKeys.add(fieldName);
}
}
builder.primaryKey(primaryKeys);
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
index 7f759584b..427f66399 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
@@ -20,6 +20,14 @@ package org.apache.paimon.flink.action.cdc;
import org.apache.paimon.flink.action.ActionBase;
import org.apache.paimon.flink.action.ActionITCaseBase;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSyncDatabaseActionFactory;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSyncTableActionFactory;
+import
org.apache.paimon.flink.action.cdc.mongodb.MongoDBSyncDatabaseActionFactory;
+import
org.apache.paimon.flink.action.cdc.mongodb.MongoDBSyncTableActionFactory;
+import org.apache.paimon.flink.action.cdc.mysql.MySqlSyncDatabaseActionFactory;
+import org.apache.paimon.flink.action.cdc.mysql.MySqlSyncTableActionFactory;
+import
org.apache.paimon.flink.action.cdc.pulsar.PulsarSyncDatabaseActionFactory;
+import org.apache.paimon.flink.action.cdc.pulsar.PulsarSyncTableActionFactory;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.TableScan;
@@ -202,20 +210,66 @@ public class CdcActionITCaseBase extends ActionITCaseBase
{
}
}
+ private <T> String getActionName(Class<T> clazz) {
+ switch (clazz.getSimpleName()) {
+ case "MySqlSyncTableAction":
+ return MySqlSyncTableActionFactory.IDENTIFIER;
+ case "MySqlSyncDatabaseAction":
+ return MySqlSyncDatabaseActionFactory.IDENTIFIER;
+ case "KafkaSyncTableAction":
+ return KafkaSyncTableActionFactory.IDENTIFIER;
+ case "KafkaSyncDatabaseAction":
+ return KafkaSyncDatabaseActionFactory.IDENTIFIER;
+ case "MongoDBSyncTableAction":
+ return MongoDBSyncTableActionFactory.IDENTIFIER;
+ case "MongoDBSyncDatabaseAction":
+ return MongoDBSyncDatabaseActionFactory.IDENTIFIER;
+ case "PulsarSyncTableAction":
+ return PulsarSyncTableActionFactory.IDENTIFIER;
+ case "PulsarSyncDatabaseAction":
+ return PulsarSyncDatabaseActionFactory.IDENTIFIER;
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown sync action: " + clazz.getSimpleName());
+ }
+ }
+
+ private <T> String getConfKey(Class<T> clazz) {
+ switch (clazz.getSimpleName()) {
+ case "MySqlSyncTableAction":
+ case "MySqlSyncDatabaseAction":
+ return "--" + CdcActionCommonUtils.MYSQL_CONF;
+ case "KafkaSyncTableAction":
+ case "KafkaSyncDatabaseAction":
+ return "--" + CdcActionCommonUtils.KAFKA_CONF;
+ case "MongoDBSyncTableAction":
+ case "MongoDBSyncDatabaseAction":
+ return "--" + CdcActionCommonUtils.MONGODB_CONF;
+ case "PulsarSyncTableAction":
+ case "PulsarSyncDatabaseAction":
+ return "--" + CdcActionCommonUtils.PULSAR_CONF;
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown sync action: " + clazz.getSimpleName());
+ }
+ }
+
/** Base builder to build table synchronization action from action
arguments. */
- protected abstract static class SyncTableActionBuilder<T> {
+ protected abstract class SyncTableActionBuilder<T extends
SynchronizationActionBase> {
- protected final Map<String, String> sourceConfig;
- protected Map<String, String> catalogConfig = Collections.emptyMap();
- protected Map<String, String> tableConfig = Collections.emptyMap();
+ private final Class<T> clazz;
+ private final Map<String, String> sourceConfig;
- protected final List<String> partitionKeys = new ArrayList<>();
- protected final List<String> primaryKeys = new ArrayList<>();
- protected final List<String> computedColumnArgs = new ArrayList<>();
- protected final List<String> typeMappingModes = new ArrayList<>();
- protected final List<String> metadataColumns = new ArrayList<>();
+ private Map<String, String> catalogConfig = Collections.emptyMap();
+ private Map<String, String> tableConfig = Collections.emptyMap();
+ private final List<String> partitionKeys = new ArrayList<>();
+ private final List<String> primaryKeys = new ArrayList<>();
+ private final List<String> computedColumnArgs = new ArrayList<>();
+ private final List<String> typeMappingModes = new ArrayList<>();
+ private final List<String> metadataColumns = new ArrayList<>();
- public SyncTableActionBuilder(Map<String, String> sourceConfig) {
+ public SyncTableActionBuilder(Class<T> clazz, Map<String, String>
sourceConfig) {
+ this.clazz = clazz;
this.sourceConfig = sourceConfig;
}
@@ -258,27 +312,53 @@ public class CdcActionITCaseBase extends ActionITCaseBase
{
return this;
}
- public abstract T build();
+ public T build() {
+ List<String> args =
+ new ArrayList<>(
+ Arrays.asList(
+ getActionName(clazz),
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName));
+
+ args.addAll(mapToArgs(getConfKey(clazz), sourceConfig));
+ args.addAll(mapToArgs("--catalog-conf", catalogConfig));
+ args.addAll(mapToArgs("--table-conf", tableConfig));
+
+ args.addAll(listToArgs("--partition-keys", partitionKeys));
+ args.addAll(listToArgs("--primary-keys", primaryKeys));
+ args.addAll(listToArgs("--type-mapping", typeMappingModes));
+
+ args.addAll(listToMultiArgs("--computed-column",
computedColumnArgs));
+ args.addAll(listToMultiArgs("--metadata-column", metadataColumns));
+
+ return createAction(clazz, args);
+ }
}
/** Base Builder to build database synchronization from action arguments.
*/
- protected abstract static class SyncDatabaseActionBuilder<T> {
-
- protected final Map<String, String> sourceConfig;
- protected Map<String, String> catalogConfig = Collections.emptyMap();
- protected Map<String, String> tableConfig = Collections.emptyMap();
-
- @Nullable protected Boolean ignoreIncompatible;
- @Nullable protected Boolean mergeShards;
- @Nullable protected String tablePrefix;
- @Nullable protected String tableSuffix;
- @Nullable protected String includingTables;
- @Nullable protected String excludingTables;
- @Nullable protected String mode;
- protected final List<String> typeMappingModes = new ArrayList<>();
- protected final List<String> metadataColumn = new ArrayList<>();
-
- public SyncDatabaseActionBuilder(Map<String, String> sourceConfig) {
+ protected abstract class SyncDatabaseActionBuilder<T extends
SynchronizationActionBase> {
+
+ private final Class<T> clazz;
+ private final Map<String, String> sourceConfig;
+
+ private Map<String, String> catalogConfig = Collections.emptyMap();
+ private Map<String, String> tableConfig = Collections.emptyMap();
+ @Nullable private Boolean ignoreIncompatible;
+ @Nullable private Boolean mergeShards;
+ @Nullable private String tablePrefix;
+ @Nullable private String tableSuffix;
+ @Nullable private String includingTables;
+ @Nullable private String excludingTables;
+ @Nullable private String mode;
+ private final List<String> typeMappingModes = new ArrayList<>();
+ private final List<String> metadataColumn = new ArrayList<>();
+
+ public SyncDatabaseActionBuilder(Class<T> clazz, Map<String, String>
sourceConfig) {
+ this.clazz = clazz;
this.sourceConfig = sourceConfig;
}
@@ -337,6 +417,32 @@ public class CdcActionITCaseBase extends ActionITCaseBase {
return this;
}
- public abstract T build();
+ public T build() {
+ List<String> args =
+ new ArrayList<>(
+ Arrays.asList(
+ getActionName(clazz),
+ "--warehouse",
+ warehouse,
+ "--database",
+ database));
+
+ args.addAll(mapToArgs(getConfKey(clazz), sourceConfig));
+ args.addAll(mapToArgs("--catalog-conf", catalogConfig));
+ args.addAll(mapToArgs("--table-conf", tableConfig));
+
+ args.addAll(nullableToArgs("--ignore-incompatible",
ignoreIncompatible));
+ args.addAll(nullableToArgs("--merge-shards", mergeShards));
+ args.addAll(nullableToArgs("--table-prefix", tablePrefix));
+ args.addAll(nullableToArgs("--table-suffix", tableSuffix));
+ args.addAll(nullableToArgs("--including-tables", includingTables));
+ args.addAll(nullableToArgs("--excluding-tables", excludingTables));
+ args.addAll(nullableToArgs("--mode", mode));
+
+ args.addAll(listToArgs("--type-mapping", typeMappingModes));
+ args.addAll(listToArgs("--metadata-column", metadataColumn));
+
+ return createAction(clazz, args);
+ }
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
index d044b38dd..9db87a556 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
@@ -55,7 +55,6 @@ import java.net.URL;
import java.nio.file.Files;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -239,32 +238,7 @@ public abstract class KafkaActionITCaseBase extends
CdcActionITCaseBase {
extends SyncTableActionBuilder<KafkaSyncTableAction> {
public KafkaSyncTableActionBuilder(Map<String, String> kafkaConfig) {
- super(kafkaConfig);
- }
-
- public KafkaSyncTableAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "kafka_sync_table",
- "--warehouse",
- warehouse,
- "--database",
- database,
- "--table",
- tableName));
-
- args.addAll(mapToArgs("--kafka-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
-
- args.addAll(listToArgs("--partition-keys", partitionKeys));
- args.addAll(listToArgs("--primary-keys", primaryKeys));
- args.addAll(listToArgs("--type-mapping", typeMappingModes));
-
- args.addAll(listToMultiArgs("--computed-column",
computedColumnArgs));
-
- return createAction(KafkaSyncTableAction.class, args);
+ super(KafkaSyncTableAction.class, kafkaConfig);
}
}
@@ -273,43 +247,7 @@ public abstract class KafkaActionITCaseBase extends
CdcActionITCaseBase {
extends SyncDatabaseActionBuilder<KafkaSyncDatabaseAction> {
public KafkaSyncDatabaseActionBuilder(Map<String, String> kafkaConfig)
{
- super(kafkaConfig);
- }
-
- public KafkaSyncDatabaseActionBuilder ignoreIncompatible(boolean
ignoreIncompatible) {
- throw new UnsupportedOperationException();
- }
-
- public KafkaSyncDatabaseActionBuilder mergeShards(boolean mergeShards)
{
- throw new UnsupportedOperationException();
- }
-
- public KafkaSyncDatabaseActionBuilder withMode(String mode) {
- throw new UnsupportedOperationException();
- }
-
- public KafkaSyncDatabaseAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "kafka_sync_database",
- "--warehouse",
- warehouse,
- "--database",
- database));
-
- args.addAll(mapToArgs("--kafka-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
-
- args.addAll(nullableToArgs("--table-prefix", tablePrefix));
- args.addAll(nullableToArgs("--table-suffix", tableSuffix));
- args.addAll(nullableToArgs("--including-tables", includingTables));
- args.addAll(nullableToArgs("--excluding-tables", excludingTables));
-
- args.addAll(listToArgs("--type-mapping", typeMappingModes));
-
- return createAction(KafkaSyncDatabaseAction.class, args);
+ super(KafkaSyncDatabaseAction.class, kafkaConfig);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
index 1af4e117f..237b4366b 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.action.cdc.kafka;
+import org.apache.paimon.catalog.FileSystemCatalogOptions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -26,6 +27,8 @@ import org.apache.paimon.types.RowType;
import org.apache.flink.core.execution.JobClient;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
import java.util.Collections;
@@ -1171,4 +1174,58 @@ public class KafkaCanalSyncTableActionITCase extends
KafkaActionITCaseBase {
rowType,
Collections.singletonList("k"));
}
+
+ @ParameterizedTest(name = "triggerSchemaRetrievalException = {0}")
+ @ValueSource(booleans = {true, false})
+ @Timeout(60)
+ public void testComputedColumnWithCaseInsensitive(boolean
triggerSchemaRetrievalException)
+ throws Exception {
+ String topic = "computed_column_with_case_insensitive" +
UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+
+ if (triggerSchemaRetrievalException) {
+ createFileStoreTable(
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(), DataTypes.DATE(),
DataTypes.INT(),
+ },
+ new String[] {"_id", "_date", "_year"}),
+ Collections.emptyList(),
+ Collections.singletonList("_id"),
+ Collections.emptyMap());
+ } else {
+ List<String> lines =
readLines("kafka/canal/table/computedcolumn/canal-data-2.txt");
+ writeRecordsToKafka(topic, lines);
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put(VALUE_FORMAT.key(), "canal-json");
+ kafkaConfig.put(TOPIC.key(), topic);
+ KafkaSyncTableAction action =
+ syncTableActionBuilder(kafkaConfig)
+ .withTableConfig(getBasicTableConfig())
+ .withCatalogConfig(
+ Collections.singletonMap(
+
FileSystemCatalogOptions.CASE_SENSITIVE.key(), "false"))
+ .withComputedColumnArgs("_YEAR=year(_DATE)")
+ .build();
+ runActionWithDefaultEnv(action);
+
+ if (triggerSchemaRetrievalException) {
+ List<String> lines =
readLines("kafka/canal/table/computedcolumn/canal-data-2.txt");
+ writeRecordsToKafka(topic, lines);
+ }
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(), DataTypes.DATE(),
DataTypes.INT()
+ },
+ new String[] {"_id", "_date", "_year"});
+ waitForResult(
+ Arrays.asList("+I[1, 19439, 2023]", "+I[2, NULL, NULL]"),
+ getFileStoreTable(tableName),
+ rowType,
+ Collections.singletonList("_id"));
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionITCaseBase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionITCaseBase.java
index b4c5eed77..c99865dd7 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionITCaseBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionITCaseBase.java
@@ -30,10 +30,7 @@ import org.slf4j.LoggerFactory;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
@@ -95,37 +92,7 @@ public abstract class MongoDBActionITCaseBase extends
CdcActionITCaseBase {
extends SyncTableActionBuilder<MongoDBSyncTableAction> {
public MongoDBSyncTableActionBuilder(Map<String, String>
mongodbConfig) {
- super(mongodbConfig);
- }
-
- public MongoDBSyncTableActionBuilder withPrimaryKeys(String...
primaryKeys) {
- throw new UnsupportedOperationException();
- }
-
- public MongoDBSyncTableActionBuilder withTypeMappingModes(String...
typeMappingModes) {
- throw new UnsupportedOperationException();
- }
-
- public MongoDBSyncTableAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "mongodb_sync_table",
- "--warehouse",
- warehouse,
- "--database",
- database,
- "--table",
- tableName));
-
- args.addAll(mapToArgs("--mongodb-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
- args.addAll(listToArgs("--computed-column", computedColumnArgs));
-
- args.addAll(listToArgs("--partition-keys", partitionKeys));
-
- return createAction(MongoDBSyncTableAction.class, args);
+ super(MongoDBSyncTableAction.class, mongodbConfig);
}
}
@@ -134,45 +101,7 @@ public abstract class MongoDBActionITCaseBase extends
CdcActionITCaseBase {
extends SyncDatabaseActionBuilder<MongoDBSyncDatabaseAction> {
public MongoDBSyncDatabaseActionBuilder(Map<String, String>
mongodbConfig) {
- super(mongodbConfig);
- }
-
- public MongoDBSyncDatabaseActionBuilder ignoreIncompatible(boolean
ignoreIncompatible) {
- throw new UnsupportedOperationException();
- }
-
- public MongoDBSyncDatabaseActionBuilder mergeShards(boolean
mergeShards) {
- throw new UnsupportedOperationException();
- }
-
- public MongoDBSyncDatabaseActionBuilder withMode(String mode) {
- throw new UnsupportedOperationException();
- }
-
- public MongoDBSyncDatabaseActionBuilder withTypeMappingModes(String...
typeMappingModes) {
- throw new UnsupportedOperationException();
- }
-
- public MongoDBSyncDatabaseAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "mongodb_sync_database",
- "--warehouse",
- warehouse,
- "--database",
- database));
-
- args.addAll(mapToArgs("--mongodb-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
-
- args.addAll(nullableToArgs("--table-prefix", tablePrefix));
- args.addAll(nullableToArgs("--table-suffix", tableSuffix));
- args.addAll(nullableToArgs("--including-tables", includingTables));
- args.addAll(nullableToArgs("--excluding-tables", excludingTables));
-
- return createAction(MongoDBSyncDatabaseAction.class, args);
+ super(MongoDBSyncDatabaseAction.class, mongodbConfig);
}
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionITCase.java
index 485e9876d..d43b4e925 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.action.cdc.mongodb;
+import org.apache.paimon.catalog.FileSystemCatalogOptions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -35,7 +36,7 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
-/** IT cases for {@link MongoDBSyncTableActionITCase}. */
+/** IT cases for {@link MongoDBSyncTableAction}. */
public class MongoDBSyncTableActionITCase extends MongoDBActionITCaseBase {
@Test
@@ -329,4 +330,36 @@ public class MongoDBSyncTableActionITCase extends
MongoDBActionITCaseBase {
"+I[100000000000000000000103, 12-pack drill bits,
12-pack of drill bits with sizes ranging from #40 to #3, 0.8]");
waitForResult(expectedInsert, table, rowType, primaryKeys);
}
+
+ @Test
+ @Timeout(60)
+ public void testComputedColumnWithCaseInsensitive() throws Exception {
+ writeRecordsToMongoDB("test-table-2", database,
"table/computedcolumn");
+ Map<String, String> mongodbConfig = getBasicMongoDBConfig();
+ mongodbConfig.put("database", database);
+ mongodbConfig.put("collection",
"computed_column_with_case_insensitive");
+
+ MongoDBSyncTableAction action =
+ syncTableActionBuilder(mongodbConfig)
+ .withTableConfig(getBasicTableConfig())
+ .withCatalogConfig(
+ Collections.singletonMap(
+
FileSystemCatalogOptions.CASE_SENSITIVE.key(), "false"))
+ .withComputedColumnArgs("_YEAR=year(_DATE)")
+ .build();
+ runActionWithDefaultEnv(action);
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(), DataTypes.STRING(),
DataTypes.INT()
+ },
+ new String[] {"_id", "_date", "_year"});
+ waitForResult(
+ Arrays.asList(
+ "+I[100000000000000000000101, 2023-12-11, 2023]",
+ "+I[100000000000000000000102, NULL, NULL]"),
+ getFileStoreTable(tableName),
+ rowType,
+ Collections.singletonList("_id"));
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaITCase.java
index 1881747d1..eb207f8c0 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongodbSchemaITCase.java
@@ -83,7 +83,7 @@ public class MongodbSchemaITCase extends
MongoDBActionITCaseBase {
mongodbConfig.setString(MongoDBSourceOptions.CONNECTION_OPTIONS,
"authSource=admin");
mongodbConfig.setString(MongoDBSourceOptions.DATABASE, "testDatabase");
mongodbConfig.setString(MongoDBSourceOptions.COLLECTION,
"testCollection");
- Schema schema = MongodbSchemaUtils.getMongodbSchema(mongodbConfig,
true);
+ Schema schema = MongodbSchemaUtils.getMongodbSchema(mongodbConfig);
assertNotNull(schema);
}
@@ -99,8 +99,7 @@ public class MongodbSchemaITCase extends
MongoDBActionITCaseBase {
mongodbConfig.setString(MongoDBSourceOptions.COLLECTION,
"testCollection");
assertThrows(
- RuntimeException.class,
- () -> MongodbSchemaUtils.getMongodbSchema(mongodbConfig,
true));
+ RuntimeException.class, () ->
MongodbSchemaUtils.getMongodbSchema(mongodbConfig));
}
@Test
@@ -111,7 +110,7 @@ public class MongodbSchemaITCase extends
MongoDBActionITCaseBase {
// Expect an exception to be thrown due to missing necessary settings
assertThrows(
NullPointerException.class,
- () -> MongodbSchemaUtils.getMongodbSchema(mongodbConfig,
true));
+ () -> MongodbSchemaUtils.getMongodbSchema(mongodbConfig));
}
@Test
@@ -127,7 +126,7 @@ public class MongodbSchemaITCase extends
MongoDBActionITCaseBase {
mongodbConfig.setString(MongoDBSourceOptions.COLLECTION,
"testCollection");
// Call the method and check the results
- Schema schema = MongodbSchemaUtils.getMongodbSchema(mongodbConfig,
true);
+ Schema schema = MongodbSchemaUtils.getMongodbSchema(mongodbConfig);
// Verify the schema
assertNotNull(schema);
@@ -152,8 +151,7 @@ public class MongodbSchemaITCase extends
MongoDBActionITCaseBase {
mongodbConfig.setString(MongoDBSourceOptions.COLLECTION,
"testCollection");
assertThrows(
- RuntimeException.class,
- () -> MongodbSchemaUtils.getMongodbSchema(mongodbConfig,
true));
+ RuntimeException.class, () ->
MongodbSchemaUtils.getMongodbSchema(mongodbConfig));
}
@Test
@@ -168,7 +166,6 @@ public class MongodbSchemaITCase extends
MongoDBActionITCaseBase {
mongodbConfig.setString(MongoDBSourceOptions.COLLECTION,
"invalidCollection");
assertThrows(
- RuntimeException.class,
- () -> MongodbSchemaUtils.getMongodbSchema(mongodbConfig,
true));
+ RuntimeException.class, () ->
MongodbSchemaUtils.getMongodbSchema(mongodbConfig));
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionITCaseBase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionITCaseBase.java
index b9f91b3b4..7da281b99 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionITCaseBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionITCaseBase.java
@@ -31,10 +31,7 @@ import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.ZoneId;
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
@@ -104,33 +101,7 @@ public class MySqlActionITCaseBase extends
CdcActionITCaseBase {
extends SyncTableActionBuilder<MySqlSyncTableAction> {
public MySqlSyncTableActionBuilder(Map<String, String> mySqlConfig) {
- super(mySqlConfig);
- }
-
- public MySqlSyncTableAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "mysql_sync_table",
- "--warehouse",
- warehouse,
- "--database",
- database,
- "--table",
- tableName));
-
- args.addAll(mapToArgs("--mysql-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
-
- args.addAll(listToArgs("--partition-keys", partitionKeys));
- args.addAll(listToArgs("--primary-keys", primaryKeys));
- args.addAll(listToArgs("--type-mapping", typeMappingModes));
-
- args.addAll(listToMultiArgs("--computed-column",
computedColumnArgs));
- args.addAll(listToMultiArgs("--metadata-column", metadataColumns));
-
- return createAction(MySqlSyncTableAction.class, args);
+ super(MySqlSyncTableAction.class, mySqlConfig);
}
}
@@ -139,35 +110,7 @@ public class MySqlActionITCaseBase extends
CdcActionITCaseBase {
extends SyncDatabaseActionBuilder<MySqlSyncDatabaseAction> {
public MySqlSyncDatabaseActionBuilder(Map<String, String> mySqlConfig)
{
- super(mySqlConfig);
- }
-
- public MySqlSyncDatabaseAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "mysql_sync_database",
- "--warehouse",
- warehouse,
- "--database",
- database));
-
- args.addAll(mapToArgs("--mysql-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
-
- args.addAll(nullableToArgs("--ignore-incompatible",
ignoreIncompatible));
- args.addAll(nullableToArgs("--merge-shards", mergeShards));
- args.addAll(nullableToArgs("--table-prefix", tablePrefix));
- args.addAll(nullableToArgs("--table-suffix", tableSuffix));
- args.addAll(nullableToArgs("--including-tables", includingTables));
- args.addAll(nullableToArgs("--excluding-tables", excludingTables));
- args.addAll(nullableToArgs("--mode", mode));
-
- args.addAll(listToArgs("--type-mapping", typeMappingModes));
- args.addAll(listToArgs("--metadata-column", metadataColumn));
-
- return createAction(MySqlSyncDatabaseAction.class, args);
+ super(MySqlSyncDatabaseAction.class, mySqlConfig);
}
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
index e7abc8902..30e200b48 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.action.cdc.mysql;
+import org.apache.paimon.catalog.FileSystemCatalogOptions;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
@@ -654,7 +655,7 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
}
@Test
- public void testInvalidPrimaryKey() throws Exception {
+ public void testInvalidPrimaryKey() {
Map<String, String> mySqlConfig = getBasicMySqlConfig();
mySqlConfig.put("database-name", DATABASE_NAME);
mySqlConfig.put("table-name", "schema_evolution_\\d+");
@@ -998,4 +999,62 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
thread.interrupt();
env.close();
}
+
+ @Test
+ @Timeout(60)
+ public void testComputedColumnWithCaseInsensitive() throws Exception {
+ Map<String, String> mySqlConfig = getBasicMySqlConfig();
+ mySqlConfig.put("database-name",
"computed_column_with_case_insensitive");
+ mySqlConfig.put("table-name", "t");
+
+ MySqlSyncTableAction action =
+ syncTableActionBuilder(mySqlConfig)
+ .withCatalogConfig(
+ Collections.singletonMap(
+
FileSystemCatalogOptions.CASE_SENSITIVE.key(), "false"))
+
.withComputedColumnArgs("SUBSTRING=substring(UPPERCASE_STRING,2)")
+ .build();
+ runActionWithDefaultEnv(action);
+
+ try (Statement statement = getStatement()) {
+ statement.execute("USE computed_column_with_case_insensitive");
+ statement.executeUpdate("INSERT INTO t VALUES (1, 'apache')");
+ statement.executeUpdate("INSERT INTO t VALUES (2, null)");
+ }
+
+ FileStoreTable table = getFileStoreTable();
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(), DataTypes.VARCHAR(10),
DataTypes.STRING(),
+ },
+ new String[] {"pk", "uppercase_string", "substring"});
+ waitForResult(
+ Arrays.asList("+I[1, apache, ache]", "+I[2, NULL, NULL]"),
+ table,
+ rowType,
+ Collections.singletonList("pk"));
+ }
+
+ @Test
+ @Timeout(60)
+ public void testSpecifyKeysWithCaseInsensitive() throws Exception {
+ Map<String, String> mySqlConfig = getBasicMySqlConfig();
+ mySqlConfig.put("database-name", "specify_key_with_case_insensitive");
+ mySqlConfig.put("table-name", "t");
+
+ MySqlSyncTableAction action =
+ syncTableActionBuilder(mySqlConfig)
+ .withCatalogConfig(
+ Collections.singletonMap(
+
FileSystemCatalogOptions.CASE_SENSITIVE.key(), "false"))
+ .withPrimaryKeys("ID1", "PART")
+ .withPartitionKeys("PART")
+ .build();
+ runActionWithDefaultEnv(action);
+
+ FileStoreTable table = getFileStoreTable();
+ assertThat(table.primaryKeys()).containsExactly("id1", "part");
+ assertThat(table.partitionKeys()).containsExactly("part");
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
index 5e96b3a5f..969a9416f 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
@@ -48,7 +48,6 @@ import java.net.URL;
import java.nio.file.Files;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -269,32 +268,7 @@ public class PulsarActionITCaseBase extends
CdcActionITCaseBase {
extends SyncTableActionBuilder<PulsarSyncTableAction> {
public PulsarSyncTableActionBuilder(Map<String, String> pulsarConfig) {
- super(pulsarConfig);
- }
-
- public PulsarSyncTableAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "pulsar_sync_table",
- "--warehouse",
- warehouse,
- "--database",
- database,
- "--table",
- tableName));
-
- args.addAll(mapToArgs("--pulsar-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
-
- args.addAll(listToArgs("--partition-keys", partitionKeys));
- args.addAll(listToArgs("--primary-keys", primaryKeys));
- args.addAll(listToArgs("--type-mapping", typeMappingModes));
-
- args.addAll(listToMultiArgs("--computed-column",
computedColumnArgs));
-
- return createAction(PulsarSyncTableAction.class, args);
+ super(PulsarSyncTableAction.class, pulsarConfig);
}
}
@@ -303,43 +277,7 @@ public class PulsarActionITCaseBase extends
CdcActionITCaseBase {
extends SyncDatabaseActionBuilder<PulsarSyncDatabaseAction> {
public PulsarSyncDatabaseActionBuilder(Map<String, String>
pulsarConfig) {
- super(pulsarConfig);
- }
-
- public PulsarSyncDatabaseActionBuilder ignoreIncompatible(boolean
ignoreIncompatible) {
- throw new UnsupportedOperationException();
- }
-
- public PulsarSyncDatabaseActionBuilder mergeShards(boolean
mergeShards) {
- throw new UnsupportedOperationException();
- }
-
- public PulsarSyncDatabaseActionBuilder withMode(String mode) {
- throw new UnsupportedOperationException();
- }
-
- public PulsarSyncDatabaseAction build() {
- List<String> args =
- new ArrayList<>(
- Arrays.asList(
- "pulsar_sync_database",
- "--warehouse",
- warehouse,
- "--database",
- database));
-
- args.addAll(mapToArgs("--pulsar-conf", sourceConfig));
- args.addAll(mapToArgs("--catalog-conf", catalogConfig));
- args.addAll(mapToArgs("--table-conf", tableConfig));
-
- args.addAll(nullableToArgs("--table-prefix", tablePrefix));
- args.addAll(nullableToArgs("--table-suffix", tableSuffix));
- args.addAll(nullableToArgs("--including-tables", includingTables));
- args.addAll(nullableToArgs("--excluding-tables", excludingTables));
-
- args.addAll(listToArgs("--type-mapping", typeMappingModes));
-
- return createAction(PulsarSyncDatabaseAction.class, args);
+ super(PulsarSyncDatabaseAction.class, pulsarConfig);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/canal/table/computedcolumn/canal-data-2.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/canal/table/computedcolumn/canal-data-2.txt
new file mode 100644
index 000000000..473382cd9
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/canal/table/computedcolumn/canal-data-2.txt
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+{"data":[{"_ID":"1","_DATE":"2023-03-23"},{"_ID":"2","_DATE":null}],"database":"paimon_sync_table","es":1683006706000,"id":92,"isDdl":false,"mysqlType":{"_ID":"INT","_DATE":"DATE"},"old":null,"pkNames":["_ID"],"sql":"","sqlType":{"_ID":4,"_DATE":91},"table":"computed_column_with_case_insensitive","ts":1683006706728,"type":"INSERT"}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/mongodb/table/computedcolumn/test-table-2.js
b/paimon-flink/paimon-flink-cdc/src/test/resources/mongodb/table/computedcolumn/test-table-2.js
new file mode 100644
index 000000000..17ee3deaa
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/mongodb/table/computedcolumn/test-table-2.js
@@ -0,0 +1,24 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// -- this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+db.getCollection('computed_column_with_case_insensitive').insertMany([
+ {
+ "_id": ObjectId("100000000000000000000101"),
+ "_DATE": "2023-12-11"
+ },
+ {
+ "_id": ObjectId("100000000000000000000102")
+ }
+]);
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
index 7990a0330..814ba8f2e 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
@@ -347,4 +347,31 @@ CREATE TABLE t (
pk INT,
v VARCHAR(10),
PRIMARY KEY (pk)
+);
+
+--
################################################################################
+-- testComputedColumnWithCaseInsensitive
+--
################################################################################
+
+CREATE DATABASE computed_column_with_case_insensitive;
+USE computed_column_with_case_insensitive;
+
+CREATE TABLE t (
+ PK INT,
+ UPPERCASE_STRING VARCHAR(10),
+ PRIMARY KEY (pk)
+);
+
+--
################################################################################
+-- testSpecifyKeysWithCaseInsensitive
+--
################################################################################
+
+CREATE DATABASE specify_key_with_case_insensitive;
+USE specify_key_with_case_insensitive;
+
+CREATE TABLE t (
+ ID0 INT,
+ ID1 INT,
+ PART INT,
+ PRIMARY KEY (ID0)
);
\ No newline at end of file
