[flink] Support ALTER MATERIALIZED TABLE ... AS <query> in FlinkCatalog#3596
Open
xuyangzhong wants to merge 1 commit into
Open
[flink] Support ALTER MATERIALIZED TABLE ... AS <query> in FlinkCatalog#3596xuyangzhong wants to merge 1 commit into
xuyangzhong wants to merge 1 commit into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: close #3595
FlinkCatalog#alterTablecannot alter a materialized table's definition query.Running
ALTER MATERIALIZED TABLE <tbl> AS <new_query>makes Flink emit aTableChange.ModifyDefinitionQuery, whichFlinkConversionsdoes not recognizeand rejects with
UnsupportedOperationException: Unsupported flink table change.ModifyDefinitionQuerywas introduced in Flink 2.0, so it cannot be referenceddirectly 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
TableChangeAdapterinfluss-flink-commonwhosedefault
convert(...)returnsOptional.empty()(nothing version-specific toconvert for the compile baseline).
TableChangeAdapterinfluss-flink-2.2to convertTableChange.ModifyDefinitionQueryinto a FlussSetOptiononmaterialized-table.definition-query. New version-specific changes only need anew branch here in the future.
TableChangeinFlinkConversions#toFlussTableChangesthrough
TableChangeAdapter.convert(...)before throwingUnsupportedOperationException, so no call-site changes are needed later.Tests
Flink22TableChangeAdapterTest:ModifyDefinitionQueryis converted to theexpected
SetOption; an unhandled change (ModifyRefreshStatus) returnsOptional.empty().Flink22FlinkConversionsTest:FlinkConversions.toFlussTableChangesmapsModifyDefinitionQueryto aSetOptiononmaterialized-table.definition-query.Flink22MaterializedTableITCase:testAlterMaterializedTableAsQueryInContinuousMode: alter the definitionquery (same output schema) and verify the query/refresh job are updated.
testAlterMaterializedTableAsQueryWithAppendColumn: append a nullable columnat the end (the only schema evolution Flink allows for materialized tables).
testAlterMaterializedTableAsQueryWithUnsupportedSchemaChange: dropping andreordering columns are rejected by the Flink planner.
API and Format
No. No public API or storage format changes; only the existing
materialized-table.definition-queryoption is written on alter.Documentation
Later.