diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransform.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransform.java index 742322d7826..03128aeef75 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransform.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransform.java @@ -50,14 +50,14 @@ public SchemaMetadataTransform( if (!StringUtils.isNullOrWhitespaceOnly(primaryKeyString)) { String[] primaryKeyArr = primaryKeyString.split(","); for (int i = 0; i < primaryKeyArr.length; i++) { - primaryKeyArr[i] = primaryKeyArr[i].trim(); + primaryKeyArr[i] = normalizeColumnReference(primaryKeyArr[i]); } primaryKeys = Arrays.asList(primaryKeyArr); } if (!StringUtils.isNullOrWhitespaceOnly(partitionKeyString)) { String[] partitionKeyArr = partitionKeyString.split(","); for (int i = 0; i < partitionKeyArr.length; i++) { - partitionKeyArr[i] = partitionKeyArr[i].trim(); + partitionKeyArr[i] = normalizeColumnReference(partitionKeyArr[i]); } partitionKeys = Arrays.asList(partitionKeyArr); } @@ -86,6 +86,14 @@ public SchemaMetadataTransform( } } + private String normalizeColumnReference(String columnReference) { + String trimmed = columnReference.trim(); + if (trimmed.length() >= 2 && trimmed.startsWith("`") && trimmed.endsWith("`")) { + return trimmed.substring(1, trimmed.length() - 1).replace("``", "`"); + } + return trimmed; + } + public List getPrimaryKeys() { return primaryKeys; } diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransformTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransformTest.java index d12f669fe0b..49b65df07f2 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransformTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransformTest.java @@ -24,6 +24,16 @@ /** Testcases for {@link SchemaMetadataTransform}. */ class SchemaMetadataTransformTest { + @Test + void testPrimaryAndPartitionKeysStripBackticks() { + SchemaMetadataTransform transform = + new SchemaMetadataTransform( + "id, `order-id`, `customer``name`", "region, `order-id`", null, null); + + assertThat(transform.getPrimaryKeys()).containsExactly("id", "order-id", "customer`name"); + assertThat(transform.getPartitionKeys()).containsExactly("region", "order-id"); + } + @Test void testTableOptionsWithCommaDelimiter() { SchemaMetadataTransform transform =