xuyangzhong opened a new pull request, #3596: URL: https://github.com/apache/fluss/pull/3596
<!-- Generated-by: Cursor (Claude Opus 4.8) following the guidelines (https://github.com/apache/fluss/blob/main/AGENTS.md) --> ### Purpose Linked issue: close #xxx `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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
