[flink] Tables read back from the catalog carry bucket and connection options, breaking CREATE OR ALTER MATERIALIZED TABLE and LIKE #4667

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

When Flink reads a table from the Fluss catalog, it gets more than the user declared:

• The buckets come back as bucket.num and bucket.key options, never as DISTRIBUTED BY. See FlinkConversions#toFlinkTable.
• FlinkCatalog#getTable adds bootstrap.servers, the client security configs and the storage configs such as table.replication.factor.

Two Flink features compare that table with a new statement and trip over the extra options. Reported by @fresh-borzoni in #4635.

1. Re-running CREATE OR ALTER MATERIALIZED TABLE fails on Flink 2.3

CREATE OR ALTER MATERIALIZED TABLE mt
  DISTRIBUTED BY HASH(a) INTO 3 BUCKETS
  FRESHNESS = INTERVAL '30' SECOND
  AS SELECT a, b FROM src;
-- running the same statement again fails

Flink turns every stored option that the statement doesn't set into a reset:

FlinkCatalog#alterTable rejects these resets with The option '<key>' is not supported to alter yet. The statement fails without DISTRIBUTED BY too.

2. CREATE TABLE ... DISTRIBUTED BY ... LIKE fails

CREATE TABLE t1 (a INT, b STRING) DISTRIBUTED BY HASH(a) INTO 3 BUCKETS;
CREATE TABLE t2 DISTRIBUTED BY HASH(a) INTO 4 BUCKETS LIKE t1;
-- Cannot use DISTRIBUTED BY together with the 'bucket.key' or 'bucket.num' option.

LIKE copies the options of t1 by default, so t2 gets them next to its own clause. This appears once #4635 is merged. LIKE t1 (EXCLUDING OPTIONS) works around it.

Solution

1. On Flink 1.20+, return the buckets as a TableDistribution instead of options. Flink 1.18 and 1.19 keep the options.
2. Implement SupportsBucketing in FlinkTableSink. Without it, DynamicSinkUtils#validateBucketing rejects every insert into a table that has a distribution.
3. Stop CREATE OR ALTER MATERIALIZED TABLE from resetting options that the catalog adds itself, such as bootstrap.servers and the security configs.

SHOW CREATE TABLE would then print DISTRIBUTED BY instead of the bucket options. That is a user-visible change.

Are you willing to submit a PR?

• I'm willing to submit a PR!