Skip to content

[FLINK-41214][table] Add Table#materializeInto for materialized tables - #29838

Open
Timm0 wants to merge 2 commits into
apache:masterfrom
Timm0:FLINK-41214
Open

Timm0 wants to merge 2 commits into
apache:masterfrom
Timm0:FLINK-41214

Conversation

@Timm0

@Timm0 Timm0 commented Nov 9, 2026

Copy link
Copy Markdown
Contributor

What is the purpose of the change

Table API programs can't declare a materialized table, they have to fall back to SQL CREATE MATERIALIZED TABLE. This adds Table#materializeInto, whose MaterializedPipeline#execute() has the semantics of CREATE OR ALTER MATERIALIZED TABLE ... AS <query>. The declaration logic moves from the planner converters into a shared api-java factory, so SQL and the Table API produce the same operations.

Brief change log

  • New Table#materializeInto(String) and materializeInto(String, Duration), MaterializedPipeline and MaterializedTable (@PublicEvolving).
  • New MaterializedTableOperationFactory with the create/alter/convert logic of AbstractCreateMaterializedTableConverter and SqlCreateOrAlterMaterializedTableConverter. The SQL converter now feeds it through SqlMaterializedTableMergeContext, the Table API through QueryMaterializedTableMergeContext.
  • Schema helpers moved from the planner MaterializedTableUtils to MaterializedTableSchemaUtils.
  • The Table API rejects a regular table or a temporary table or view at the path, an invalid freshness, and queries not expressible in SQL.
  • Tests: MaterializedPipelineImplTest for validation rules, ITCase CREATEs switched to materializeInto, ValidateAndExtractColumnChangesTest moved to api-java.
  • Docs: new docs/content/docs/sql/materialized-table/table-api.md, with an zh copy.

Verifying this change

This change added tests and can be verified as follows:

  • Added MaterializedPipelineImplTest, which rejects a regular or temporary table at the path, an invalid freshness , and queries that can't be stored as SQL
  • Switched the MaterializedTableTableApiITCase CREATE statements to materializeInto wherever the API can express them
  • Moved ValidateAndExtractColumnChangesTest from the planner to api-java
  • The existing SQL tests pass unchanged, so SQL behavior is unchanged. The planner materialized table converter tests, SqlNodeToOperationConvertTableToMaterializedTableTest, and the gateway MaterializedTableStatementITCase, MaterializedTableConversionITCase, SqlGatewayRestEndpointMaterializedTableITCase and DeployScriptITCase

Does this pull request potentially affect one of the following parts:

  • Dependencies (does it add or upgrade a dependency): no
  • The public API, i.e., is any changed class annotated with @Public(Evolving): yes
  • The serializers: no
  • The runtime per-record code paths (performance sensitive): no
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
  • The S3 file system connector: no

Documentation

  • Does this pull request introduce a new feature? yes
  • If yes, how is the feature documented? docs and JavaDocs

Was generative AI tooling used to co-author this PR?
  • Yes (please specify the tool below)

Generated-by: Claude Code Opus 5.5

@Timm0
Timm0 marked this pull request as ready for review November 9, 2026 16:41
@flinkbot

flinkbot commented Nov 9, 2026 •

Copy link
Copy Markdown
Collaborator

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

@AHeise AHeise left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice to see SQL and the Table API converge on one factory. I diffed the moved create/alter/convert logic against master and it matches, apart from the validation order (inline). Clean split into refactor + API commits.

Main ask: there's no test for redeploying over an existing materialized table. Every materializeInto(...).execute() in MaterializedTableTableApiITCase is a first create, and every ALTER is SQL. I'd add at least:

  • SQL-created MT (with WITH/DISTRIBUTED BY/PARTITIONED BY) → materializeInto redeploy, asserting options, distribution, partition keys and freshness afterwards
  • a redeploy with a changed but compatible query (exercises hasSchemaDefinition=false against a stored schema)
  • a rejected column change leaving the table SUSPENDED, as documented

Also, a couple of the swapped ITCases were the only plain-TableEnvironment SQL CREATE coverage (e.g. without FRESHNESS); I'd keep one SQL variant next to the new Table API ones.

@AHeise AHeise self-assigned this Nov 11, 2026
@Timm0
Timm0 force-pushed the FLINK-41214 branch 2 times, most recently from fbd043a to dfa6629 Compare November 12, 2026 11:19
- Move the create/alter/convert logic of `AbstractCreateMaterializedTableConverter` and `SqlCreateOrAlterMaterializedTableConverter` into `MaterializedTableOperationFactory`
- Add `MaterializedTableMergeContext` as the factory input, implemented for SQL by `SqlMaterializedTableMergeContext`
- Move the schema helpers of the planner `MaterializedTableUtils` to `MaterializedTableSchemaUtils`, and `ValidateAndExtractColumnChangesTest` with them
- Add `Table#materializeInto(String)` and `materializeInto(String, Duration)` returning a `MaterializedPipeline`
- `MaterializedPipeline#execute()` returns a `MaterializedTable` handle
- Build the operation through `MaterializedTableOperationFactory` with `QueryMaterializedTableMergeContext`
- Reject a regular or temporary table at the path, an invalid freshness, and queries not expressible in SQL
- Add `MaterializedPipelineImplTest`
- Switch ITCase CREATEs to `materializeInto`
- Add the Table API page of the materialized table docs
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.

3 participants