liting liu created FLINK-41312:
----------------------------------
Summary: Preserve source bucket keys when auto-creating
Fluss-to-Fluss sink tables
Key: FLINK-41312
URL: https://issues.apache.org/jira/browse/FLINK-41312
Project: Flink
Issue Type: Improvement
Components: Flink CDC
Reporter: liting liu
h2. Motivation
For Fluss-to-Fluss replication, a missing sink table is created automatically.
However, the source table's bucket keys are not carried in the CDC schema.
Unless the user repeats them in the sink's bucket.key option, the new target
can have a different bucket layout.
This also affects ordinary non-partitioned tables; implicit partitioning is not
required.
h2. Minimal reproduction
# Create a non-partitioned Fluss source table cdc_repro.orders with the
following definition:
{code}
Columns: id INT NOT NULL, tenant INT NOT NULL, marker STRING
Primary key: (id, tenant)
Bucket keys: (tenant)
Bucket count: 2
{code}
# Leave cdc_repro.orders absent in a second Fluss cluster.
# Start a Fluss-to-Fluss CDC pipeline for that table, without a sink bucket.key
override.
# Inspect the bucket keys of the automatically created target table.
The current code derives target bucket keys (id, tenant), instead of preserving
the source bucket keys (tenant). The columns and primary key can therefore
match while the bucket layout differs.
h2. Current implementation
Reviewed Apache Flink CDC master at db45d06a1e35bc6be87f7b58913c768f8f299d3e:
* FlussMetadataAccessor.getTableSchema() passes only the Fluss schema and
partition keys to FlussConversions.toCdcSchema(); source bucket keys are not
propagated.
* FlussMetaDataApplier.applyCreateTable() gets bucket keys from the sink
configuration.
* FlussConversions.toFlussTable() defaults unspecified bucket keys to primary
keys minus partition keys.
Source references:
*
https://github.com/apache/flink-cdc/blob/db45d06a1e35bc6be87f7b58913c768f8f299d3e/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/FlussMetadataAccessor.java#L98-L104
*
https://github.com/apache/flink-cdc/blob/db45d06a1e35bc6be87f7b58913c768f8f299d3e/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java#L169-L186
h2. Expected behavior / proposal
When automatically creating a Fluss sink table from a Fluss source, preserve
the source bucket keys if no explicit sink bucket.key override is provided.
* Keep explicit sink overrides authoritative.
* Keep the current default for sources that do not provide Fluss bucket
metadata.
* Preserve the existing target table's layout when the target already exists.
The current documentation describes the existing default, so this issue
proposes improving the Fluss-to-Fluss auto-creation behavior rather than
claiming the implementation violates that documented default.
h2. Related work and workaround
FLINK-40850 / PR #4564 fixes compatibility checks for pre-created primary-key
tables with custom bucket keys. It explicitly leaves new-table creation
unchanged, so it does not address this case.
The current workaround is to repeat the source bucket keys in the sink
configuration:
{code:yaml}
sink:
type: fluss
bootstrap.servers: <target-bootstrap>
bucket.key: cdc_repro.orders:tenant
{code}
h2. Validation scope
The metadata propagation and defaulting behavior above were confirmed by
source-code inspection. A new upstream end-to-end reproduction has not been run
for this report.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
