[
https://issues.apache.org/jira/browse/FLINK-41312?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
liting liu updated FLINK-41312:
-------------------------------
Description:
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.
The example below uses an ordinary non-partitioned table.
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.
was:
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.
please assign to me
> 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
> Priority: Major
>
> 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.
> The example below uses an ordinary non-partitioned table.
> 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)
