This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new f24e80811 [FLINK-40850][pipeline-connector/fluss] Accept custom bucket
keys in pre-created primary-key tables (#4564)
f24e80811 is described below
commit f24e80811a51a26561b8e4c09c809ad1d4b58f95
Author: Xiaobing Fang <[email protected]>
AuthorDate: Sat Oct 10 14:38:08 2026 +0800
[FLINK-40850][pipeline-connector/fluss] Accept custom bucket keys in
pre-created primary-key tables (#4564)
---
.../fluss/sink/FlussMetaDataApplier.java | 16 ++++--
.../fluss/sink/FlussMetadataApplierTest.java | 57 ++++++++++++++++++++++
2 files changed, 70 insertions(+), 3 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
index 92cd2ef70..3622d4992 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
@@ -169,7 +169,10 @@ public class FlussMetaDataApplier implements
MetadataApplier, ExistingTableSchem
} else {
TableInfo currentTableInfo =
admin.getTableInfo(tablePath).get();
// sanity check to prevent unexpected table schema evolution.
- sanityCheck(inferredFlussTable, currentTableInfo);
+ sanityCheck(
+ inferredFlussTable,
+ currentTableInfo,
+ bucketKeysMap.containsKey(tableIdentifier));
}
} catch (Exception e) {
LOG.error("Failed to apply schema change {}", event, e);
@@ -222,7 +225,10 @@ public class FlussMetaDataApplier implements
MetadataApplier, ExistingTableSchem
}
}
- private void sanityCheck(TableDescriptor inferredFlussTable, TableInfo
currentTableInfo) {
+ private void sanityCheck(
+ TableDescriptor inferredFlussTable,
+ TableInfo currentTableInfo,
+ boolean hasExplicitBucketKeys) {
List<String> inferredPrimaryKeyColumnNames =
inferredFlussTable.getSchema().getPrimaryKeyColumnNames().stream()
.sorted()
@@ -242,7 +248,11 @@ public class FlussMetaDataApplier implements
MetadataApplier, ExistingTableSchem
List<String> inferredBucketKeys = inferredFlussTable.getBucketKeys();
List<String> currentBucketKeys = currentTableInfo.getBucketKeys();
- if (!inferredBucketKeys.equals(currentBucketKeys)) {
+ boolean bucketKeysMatch =
+ !hasExplicitBucketKeys &&
!inferredPrimaryKeyColumnNames.isEmpty()
+ ? inferredBucketKeys.containsAll(currentBucketKeys)
+ : inferredBucketKeys.equals(currentBucketKeys);
+ if (!bucketKeysMatch) {
throw new ValidationException(
"The table schema inferred by Flink CDC is not matched
with current Fluss table schema. "
+ "\n New Fluss table's bucket keys : "
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
index 5a2f82dad..dface1798 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
@@ -28,11 +28,13 @@ import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.IntType;
import
org.apache.flink.cdc.runtime.operators.schema.common.ExistingTableSchemaExpander;
+import org.apache.flink.table.api.ValidationException;
import org.apache.fluss.client.Connection;
import org.apache.fluss.client.ConnectionFactory;
import org.apache.fluss.client.admin.Admin;
import org.apache.fluss.exception.InvalidConfigException;
+import org.apache.fluss.metadata.DatabaseDescriptor;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
@@ -46,12 +48,14 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
+import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlussSchema;
import static org.apache.fluss.config.ConfigOptions.TABLE_REPLICATION_FACTOR;
import static org.apache.fluss.types.DataTypeChecks.equalsWithFieldId;
import static org.assertj.core.api.Assertions.assertThat;
@@ -672,6 +676,59 @@ public class FlussMetadataApplierTest {
}
}
+ @ParameterizedTest
+ @ValueSource(strings = {"tenant,id", "tenant"})
+ void testExistingPrimaryKeyTableWithCustomBucketKeys(String bucketKeys)
throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("tenant", DataTypes.INT().notNull())
+ .physicalColumn("marker", DataTypes.STRING())
+ .primaryKey("id", "tenant")
+ .build();
+ TablePath tablePath = new TablePath(DATABASE_NAME, "table1");
+ CreateTableEvent event =
+ new CreateTableEvent(TableId.tableId(DATABASE_NAME, "table1"),
schema);
+ List<String> currentBucketKeys = Arrays.asList(bucketKeys.split(","));
+ admin.createDatabase(DATABASE_NAME, DatabaseDescriptor.EMPTY,
true).get();
+ admin.createTable(
+ tablePath,
+ TableDescriptor.builder()
+ .schema(toFlussSchema(schema))
+ .distributedBy(2, currentBucketKeys)
+ .build(),
+ false)
+ .get();
+ TableInfo originalTableInfo = admin.getTableInfo(tablePath).get();
+
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(event);
+ TableInfo tableInfo = admin.getTableInfo(tablePath).get();
+
assertThat(tableInfo.getTableId()).isEqualTo(originalTableInfo.getTableId());
+
assertThat(tableInfo.getBucketKeys()).containsExactlyElementsOf(currentBucketKeys);
+ assertThat(tableInfo.getNumBuckets()).isEqualTo(2);
+ }
+
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.singletonMap(
+ DATABASE_NAME + ".table1", Arrays.asList("id",
"tenant")),
+ Collections.emptyMap())) {
+ assertThatThrownBy(() -> applier.applySchemaChange(event))
+ .rootCause()
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("New Fluss table's bucket keys :
[id, tenant]")
+ .hasMessageContaining("Current Fluss's bucket keys: " +
currentBucketKeys);
+ }
+ }
+
@Test
void testCreateTableWithTableOptions() throws Exception {
Schema schema =
