This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 1d53921f83515e07d3debe3b25397697e868fcc8 Author: Li Guo <[email protected]> AuthorDate: Thu Sep 10 22:52:32 2026 -0700 [cdc] Fix computed column args upper-cased in case-insensitive mode (#9717) --- .../paimon/flink/action/cdc/ComputedColumn.java | 33 +++++++++- .../flink/action/cdc/ComputedColumnUtils.java | 44 ++++++++----- .../action/cdc/format/AbstractRecordParser.java | 3 +- .../cdc/mongodb/strategy/MongoVersionStrategy.java | 3 +- .../flink/action/cdc/mysql/MySqlRecordParser.java | 4 +- .../action/cdc/postgres/PostgresRecordParser.java | 3 +- .../flink/action/cdc/ComputedColumnUtilsTest.java | 74 ++++++++++++++++++++++ .../cdc/kafka/KafkaCanalSyncTableActionITCase.java | 20 ++++-- 8 files changed, 153 insertions(+), 31 deletions(-) diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java index 5e9041a120..2e4fc4e39d 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java @@ -23,6 +23,7 @@ import org.apache.paimon.types.DataType; import javax.annotation.Nullable; import java.io.Serializable; +import java.util.Map; /** * A Computed column's value is computed from input columns. Only expression with at most two inputs @@ -34,10 +35,12 @@ public class ComputedColumn implements Serializable { private final String columnName; private final Expression expression; + private final boolean caseSensitive; - public ComputedColumn(String columnName, Expression expression) { + public ComputedColumn(String columnName, Expression expression, boolean caseSensitive) { this.columnName = columnName; this.expression = expression; + this.caseSensitive = caseSensitive; } public String columnName() { @@ -61,4 +64,32 @@ public class ComputedColumn implements Serializable { } return expression.eval(input); } + + /** + * Evaluates this column against a source record. The referenced field is matched by exact name, + * or ignoring case when the catalog is case-insensitive (record keys keep the case of the + * source system). + */ + @Nullable + public String evalFromRecord(Map<String, String> rowData) { + return eval(referencedValue(rowData)); + } + + @Nullable + private String referencedValue(Map<String, String> rowData) { + String reference = fieldReference(); + if (reference == null) { + return null; + } + String value = rowData.get(reference); + if (caseSensitive || rowData.containsKey(reference)) { + return value; + } + for (Map.Entry<String, String> entry : rowData.entrySet()) { + if (reference.equalsIgnoreCase(entry.getKey())) { + return entry.getValue(); + } + } + return null; + } } 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 7bd9dd561d..c85f33b6aa 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 @@ -26,11 +26,14 @@ import org.apache.paimon.utils.Preconditions; import org.apache.flink.api.java.tuple.Tuple2; import java.util.ArrayList; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import static org.apache.paimon.utils.StringUtils.toLowerCaseIfNeed; + /** Utility methods for {@link ComputedColumn}, such as build. */ public class ComputedColumnUtils { @@ -39,13 +42,21 @@ public class ComputedColumnUtils { return buildComputedColumns(computedColumnArgs, physicFields, true); } - /** The caseSensitive only affects check. We don't change field names at building phase. */ + /** + * Column names and expression arguments are kept as given. When the catalog is + * case-insensitive, names are only case-converted where they are matched: in the dependency + * sort, in the type lookup of referenced fields and in the lookup of the referenced field in + * source records. + */ public static List<ComputedColumn> buildComputedColumns( List<String> computedColumnArgs, List<DataField> physicFields, boolean caseSensitive) { Map<String, DataType> typeMapping = physicFields.stream() .collect( - Collectors.toMap(DataField::name, DataField::type, (v1, v2) -> v2)); + Collectors.toMap( + field -> toLowerCaseIfNeed(field.name(), caseSensitive), + DataField::type, + (v1, v2) -> v2)); // sort computed column args by dependencies LinkedHashMap<String, Tuple2<String, String[]>> sortedArgs = @@ -58,11 +69,12 @@ public class ComputedColumnUtils { String[] args = columnArg.getValue().f1; Expression expr = Expression.create(typeMapping, caseSensitive, exprName, args); - ComputedColumn cmpColumn = new ComputedColumn(columnName, expr); - computedColumns.add(new ComputedColumn(columnName, expr)); + ComputedColumn cmpColumn = new ComputedColumn(columnName, expr, caseSensitive); + computedColumns.add(cmpColumn); - // remember the column type for later reference by other computed columns - typeMapping.put(columnName, cmpColumn.columnType()); + // remember the column type for later reference by other computed columns, in the form + // ReferencedField looks it up + typeMapping.put(toLowerCaseIfNeed(columnName, caseSensitive), cmpColumn.columnType()); } return computedColumns; @@ -70,14 +82,12 @@ public class ComputedColumnUtils { private static LinkedHashMap<String, Tuple2<String, String[]>> sortComputedColumnArgs( List<String> computedColumnArgs, boolean caseSensitive) { - List<String> argList = - computedColumnArgs.stream() - .map(x -> caseSensitive ? x : x.toUpperCase()) - .collect(Collectors.toList()); - + // Only the keys used for the dependency sort are case-converted. Names and arguments stay + // as typed: literals such as the date_format pattern must reach the expression unchanged. LinkedHashMap<String, Tuple2<String, String[]>> eqMap = new LinkedHashMap<>(); LinkedHashMap<String, String> refMap = new LinkedHashMap<>(); - for (String arg : argList) { + Map<String, String> originalNames = new HashMap<>(); + for (String arg : computedColumnArgs) { String[] kv = arg.split("="); if (kv.length != 2) { throw new IllegalArgumentException( @@ -85,6 +95,7 @@ public class ComputedColumnUtils { "Invalid computed column argument: %s. Please use format 'column-name=expr-name(args, ...)'.", arg)); } + String columnName = kv[0].trim(); String expression = kv[1].trim(); // parse expression int left = expression.indexOf('('); @@ -97,9 +108,11 @@ public class ComputedColumnUtils { String exprName = expression.substring(0, left); String[] args = expression.substring(left + 1, right).split(","); + String sortName = toLowerCaseIfNeed(columnName, caseSensitive); + eqMap.put(columnName, Tuple2.of(exprName, args)); // args[0] may be empty string, eg. "cal_col=now()" - eqMap.put(kv[0].trim(), Tuple2.of(exprName, args)); - refMap.put(kv[0].trim(), args[0].trim()); + refMap.put(sortName, toLowerCaseIfNeed(args[0].trim(), caseSensitive)); + originalNames.put(sortName, columnName); } List<String> sortedKeys = DfsSort.sortKeys(refMap); @@ -107,7 +120,8 @@ public class ComputedColumnUtils { LinkedHashMap<String, Tuple2<String, String[]>> sortedMap = new LinkedHashMap<>(refMap.size()); for (String key : sortedKeys) { - sortedMap.put(key, eqMap.get(key)); + String columnName = originalNames.get(key); + sortedMap.put(columnName, eqMap.get(columnName)); } return sortedMap; } diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java index 5c092d881e..d5ea8f9d17 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java @@ -118,8 +118,7 @@ public abstract class AbstractRecordParser Map<String, String> rowData, CdcSchema.Builder schemaBuilder) { computedColumns.forEach( computedColumn -> { - String result = - computedColumn.eval(rowData.get(computedColumn.fieldReference())); + String result = computedColumn.evalFromRecord(rowData); rowData.put(computedColumn.columnName(), result); schemaBuilder.column(computedColumn.columnName(), computedColumn.columnType()); 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 fcabcaffbb..0a72f71160 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 @@ -160,8 +160,7 @@ public interface MongoVersionStrategy { computedColumns.forEach( computedColumn -> { String columnName = computedColumn.columnName(); - String fieldReference = computedColumn.fieldReference(); - String computedValue = computedColumn.eval(parsedRow.get(fieldReference)); + String computedValue = computedColumn.evalFromRecord(parsedRow); resultMap.put(columnName, computedValue); schemaBuilder.column(columnName, computedColumn.columnType()); diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java index 6c8f2ae324..9c4caee5ef 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java @@ -279,9 +279,7 @@ public class MySqlRecordParser implements FlatMapFunction<CdcSourceRecord, RichC // generate values of computed columns for (ComputedColumn computedColumn : computedColumns) { - String refName = computedColumn.fieldReference(); - - resultMap.put(computedColumn.columnName(), computedColumn.eval(resultMap.get(refName))); + resultMap.put(computedColumn.columnName(), computedColumn.evalFromRecord(resultMap)); // remember the computed column data type for later reference by other computed columns schemaBuilder.column(computedColumn.columnName(), computedColumn.columnType()); diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java index 003ae7b529..dc89d9408e 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java @@ -352,8 +352,7 @@ public class PostgresRecordParser // generate values of computed columns for (ComputedColumn computedColumn : computedColumns) { - String refName = computedColumn.fieldReference(); - resultMap.put(computedColumn.columnName(), computedColumn.eval(resultMap.get(refName))); + resultMap.put(computedColumn.columnName(), computedColumn.evalFromRecord(resultMap)); } for (CdcMetadataConverter metadataConverter : metadataConverters) { diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtilsTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtilsTest.java index d04f91053b..a740d41afe 100644 --- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtilsTest.java +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/ComputedColumnUtilsTest.java @@ -25,12 +25,15 @@ import org.junit.jupiter.api.Test; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.stream.Collectors; import static org.apache.paimon.flink.action.cdc.ComputedColumnUtils.buildComputedColumns; import static org.junit.Assert.assertThrows; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; /** Test for ComputedColumnUtils. */ public class ComputedColumnUtilsTest { @@ -53,6 +56,77 @@ public class ComputedColumnUtilsTest { columns.stream().map(ComputedColumn::columnName).collect(Collectors.toList())); } + @Test + public void testCaseInsensitiveKeepsArgumentsAsGiven() { + List<DataField> physicalFields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "create_time", DataTypes.TIMESTAMP(3))); + List<ComputedColumn> columns = + buildComputedColumns( + Arrays.asList( + "dt=date_format(CREATE_TIME,yyyy-MM-dd)", + "tag=cast(hello, STRING)"), + physicalFields, + false); + + assertEquals( + Arrays.asList("dt", "tag"), + columns.stream().map(ComputedColumn::columnName).collect(Collectors.toList())); + // the pattern must not be case-converted: YYYY-MM-DD means week year and day of year + assertEquals("2024-12-30", columns.get(0).eval("2024-12-30 10:00:00.000")); + assertEquals("hello", columns.get(1).eval(null)); + } + + @Test + public void testEvalFromRecordCaseInsensitive() { + List<DataField> physicalFields = + Arrays.asList(new DataField(0, "create_time", DataTypes.TIMESTAMP(3))); + Map<String, String> record = new HashMap<>(); + // the source system keeps upper case names, the user wrote the reference in lower case + record.put("CREATE_TIME", "2024-12-30 10:00:00.000"); + + ComputedColumn caseInsensitive = + buildComputedColumns( + Collections.singletonList("dt=date_format(create_time,yyyy-MM-dd)"), + physicalFields, + false) + .get(0); + assertEquals("2024-12-30", caseInsensitive.evalFromRecord(record)); + + ComputedColumn caseSensitive = + buildComputedColumns( + Collections.singletonList("dt=date_format(create_time,yyyy-MM-dd)"), + physicalFields, + true) + .get(0); + assertNull(caseSensitive.evalFromRecord(record)); + record.put("create_time", "2024-12-31 10:00:00.000"); + assertEquals("2024-12-31", caseSensitive.evalFromRecord(record)); + // an exact match wins over a match that ignores case + assertEquals("2024-12-31", caseInsensitive.evalFromRecord(record)); + // a present key with a null value is a null input, not a reason to look at other columns + record.put("create_time", null); + assertNull(caseInsensitive.evalFromRecord(record)); + } + + @Test + public void testCaseInsensitiveReferenceBetweenComputedColumns() { + List<DataField> physicalFields = Arrays.asList(new DataField(0, "_date", DataTypes.DATE())); + List<ComputedColumn> columns = + buildComputedColumns( + Arrays.asList("_YEAR_STR=substring(_year, 0, 2)", "_year=year(_DATE)"), + physicalFields, + false); + + assertEquals( + Arrays.asList("_year", "_YEAR_STR"), + columns.stream().map(ComputedColumn::columnName).collect(Collectors.toList())); + assertEquals(DataTypes.INT(), columns.get(0).columnType()); + assertEquals(DataTypes.STRING(), columns.get(1).columnType()); + assertEquals("20", columns.get(1).eval(columns.get(0).eval("2023-03-23"))); + } + @Test public void testCycleReference() { List<String> calColArgs = 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 df48d9bd4e..984c0c102b 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 @@ -1109,9 +1109,12 @@ public class KafkaCanalSyncTableActionITCase extends KafkaSyncTableActionITCase createFileStoreTable( RowType.of( new DataType[] { - DataTypes.INT().notNull(), DataTypes.DATE(), DataTypes.INT(), + DataTypes.INT().notNull(), + DataTypes.DATE(), + DataTypes.INT(), + DataTypes.STRING() }, - new String[] {"_id", "_date", "_year"}), + new String[] {"_id", "_date", "_year", "_date_str"}), Collections.emptyList(), Collections.singletonList("_id"), Collections.emptyList(), @@ -1129,7 +1132,9 @@ public class KafkaCanalSyncTableActionITCase extends KafkaSyncTableActionITCase .withCatalogConfig( Collections.singletonMap( CatalogOptions.CASE_SENSITIVE.key(), "false")) - .withComputedColumnArgs("_YEAR=year(_DATE)") + // the pattern literal must survive the case-insensitive handling + .withComputedColumnArgs( + "_YEAR=year(_DATE)", "_DATE_STR=date_format(_DATE,yyyy-MM-dd)") .build(); runActionWithDefaultEnv(action); @@ -1140,11 +1145,14 @@ public class KafkaCanalSyncTableActionITCase extends KafkaSyncTableActionITCase RowType rowType = RowType.of( new DataType[] { - DataTypes.INT().notNull(), DataTypes.DATE(), DataTypes.INT() + DataTypes.INT().notNull(), + DataTypes.DATE(), + DataTypes.INT(), + DataTypes.STRING() }, - new String[] {"_id", "_date", "_year"}); + new String[] {"_id", "_date", "_year", "_date_str"}); waitForResult( - Arrays.asList("+I[1, 19439, 2023]", "+I[2, NULL, NULL]"), + Arrays.asList("+I[1, 19439, 2023, 2023-03-23]", "+I[2, NULL, NULL, NULL]"), getFileStoreTable(tableName), rowType, Collections.singletonList("_id"));
