Skip to content

[flink] Support ALTER MATERIALIZED TABLE ... AS <query> in FlinkCatalog#3596

Open
xuyangzhong wants to merge 1 commit into
apache:mainfrom
xuyangzhong:mt_change
Open

[flink] Support ALTER MATERIALIZED TABLE ... AS <query> in FlinkCatalog#3596
xuyangzhong wants to merge 1 commit into
apache:mainfrom
xuyangzhong:mt_change

Conversation

@xuyangzhong

@xuyangzhong xuyangzhong commented Jul 6, 2026

Copy link
Copy Markdown
Contributor

Purpose

Linked issue: close #3595

FlinkCatalog#alterTable cannot alter a materialized table's definition query.
Running ALTER MATERIALIZED TABLE <tbl> AS <new_query> makes Flink emit a
TableChange.ModifyDefinitionQuery, which FlinkConversions does not recognize
and rejects with UnsupportedOperationException: Unsupported flink table change.

ModifyDefinitionQuery was introduced in Flink 2.0, so it cannot be referenced
directly from fluss-flink-common (compiled against an older Flink baseline).
This PR adds support in a way that keeps the shared module compilable while
letting version-specific modules convert newer table changes.

Brief change log

  • Add a version-agnostic TableChangeAdapter in fluss-flink-common whose
    default convert(...) returns Optional.empty() (nothing version-specific to
    convert for the compile baseline).
  • Override TableChangeAdapter in fluss-flink-2.2 to convert
    TableChange.ModifyDefinitionQuery into a Fluss SetOption on
    materialized-table.definition-query. New version-specific changes only need a
    new branch here in the future.
  • Route any unrecognized TableChange in FlinkConversions#toFlussTableChanges
    through TableChangeAdapter.convert(...) before throwing
    UnsupportedOperationException, so no call-site changes are needed later.

Tests

  • Flink22TableChangeAdapterTest: ModifyDefinitionQuery is converted to the
    expected SetOption; an unhandled change (ModifyRefreshStatus) returns
    Optional.empty().
  • Flink22FlinkConversionsTest: FlinkConversions.toFlussTableChanges maps
    ModifyDefinitionQuery to a SetOption on materialized-table.definition-query.
  • Flink22MaterializedTableITCase:
    • testAlterMaterializedTableAsQueryInContinuousMode: alter the definition
      query (same output schema) and verify the query/refresh job are updated.
    • testAlterMaterializedTableAsQueryWithAppendColumn: append a nullable column
      at the end (the only schema evolution Flink allows for materialized tables).
    • testAlterMaterializedTableAsQueryWithUnsupportedSchemaChange: dropping and
      reordering columns are rejected by the Flink planner.

API and Format

No. No public API or storage format changes; only the existing
materialized-table.definition-query option is written on alter.

Documentation

Later.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Support ALTER MATERIALIZED TABLE ... AS <query> in FlinkCatalog

1 participant