Hi Xuyang, Thank you for your feedback. Regarding your questions and comments:
1. For a materialized table declared from the Table API both would return the same SQL, generated by QueryOperation#asSerializableString(). But for a materialized table declared in SQL, they would return two different strings. When inspecting a materialized table with TableEnvironment#fromMaterializedTable(...), which was created via SQL, getOriginalQuery() will return the user-written query, while getExpandedQuery() returns the validated query with every identifier fully qualified. Regarding queries that have no SQL serialization yet, like session windows, would be rejected at materializeInto(...) with a ValidationException before anything is written to the catalog. These query limitations will be explicitly documented. 2. I would propose to apply changes in call order, each to the result of the earlier ones, as the SQL sequence in the user journey shows. So option(k, v).resetOption(k) resets k. I think this is the simplest rule because it is what already happens when using SQL, where a user also runs alter queries in order. Also rejecting conflicts would need a rule for every pair of calls and would make the Table API stricter than SQL. I'll add this to the execute() Javadoc. 3. I agree with you, it can be done as an improvement in the future. 4. In that case can't resume() be the recovering option here? When we reject a redeploy due to the mentioned inconsistency we can issue a proper error message with guidance. Users should confirm if their job is stopped and can no longer write, once they are sure they can run resume() to start a new refresh job without state and keep the table and its data. What do you think about this? I'll update the FLIP shortly with these information. Best, Timo On Mon, Oct 12, 2026 at 11:31 AM Xuyang <[email protected]> wrote: > Hi, Timo. > Thanks for the proposal. Having Table API support for materialized tables > would be useful, especially for defining and managing pipelines directly in > Java. I have a few questions about the new API, along with some thoughts on > the redeploy and recovery behavior: > 1. For materialized tables defined through the Table API, what would > getOriginalQuery() and getExpandedQuery() return? Since there is no > user-written SQL, would they return generated SQL? If the implementation > uses QueryOperation#asSerializableString(), existing serialization does not > support queries such as session windows. Would these limitations be > explicitly documented? > 2. MaterializedTableEvolution combines multiple changes into a single > ALTER. How are ordering and conflicts handled? For example, would option(k, > v).resetOption(k) be applied in invocation order, with the last operation > taking precedence, or be rejected as a conflict? > 3. Regarding your point about restarting jobs on redeploy, I think the > current behavior of always restarting could be improved. However, some > changes may be difficult to detect, such as changes to UDF implementations, > upstream table configurations, or the underlying data. In any case, this > could be addressed as a separate, small follow-up improvement. > 4. Regarding inconsistencies between the metadata in the MT catalog and > the actual cluster state, I don’t think requiring users to recreate a > cluster with the same ID just to execute DROP is reasonable. Perhaps we > could provide an explicit recovery option: once the old job is confirmed to > be absent or stopped, users could redeploy the refresh job to the target > cluster while preserving the existing table and its data. If the old > cluster is unreachable, users would first need to ensure that the old job > can no longer write, to avoid two jobs writing to the same table > concurrently. > I look forward to hearing your thoughts. > > > > -- > > Best！ > Xuyang > > > > At 2026-10-09 23:40:44, "Timo Theusner" <[email protected]> wrote: > >Hi everyone, > > > >I'd like to start a discussion on FLIP-616: Materialized Tables support > for > >Table API [1]. > > > >Materialized tables (FLIP-435 [2]) have become a convenient way to define > >pipelines declaratively. So far, though, this only works via SQL against > >the SQL Gateway. Table API users can list materialized tables, but they > >cannot define or manage them, and running the DDL through executeSql() > >only gets them halfway, since the table is registered but no refresh job > is > >ever started. > > > >The FLIP closes this gap. A Table can be turned into a materialized table > >via Table#materializeInto(...), optionally with a descriptor for > >freshness, refresh strategy, schema details or start mode. The result is a > >MaterializedTable handle that lets users change, inspect, and control it. > >To avoid two diverging implementations, the logic currently living in the > >SQL Gateway moves into the planner, so the gateway and the Table > >API behave the same. From a regular Table API program, continuous > refreshes > >work against a session cluster. Full refresh mode still > >needs the gateway for now, since it depends on a workflow scheduler. As a > >side effect, executeSql() would then actually manage refresh jobs > >for materialized table statements outside the gateway. > > > >There are two points I'd especially like your input on: > >1. Should redeploying an unchanged definition restart the refresh job? > > Today, CREATE OR ALTER always stops the job and starts a new one > >without state, even if nothing changed. Since the list of changes is > already > > computed, skipping the restart when it is empty would be a small > check. > >On the other hand, always restarting keeps the semantics simple for now > > and matches what SQL does today. > >2. How should users get out of "stuck" tables? > > If the cluster no longer "knows" the refresh job of an ACTIVATED > table, > >a redeploy is rejected, even though it would start a fresh job without > state > > anyway. That leaves DROP as the only way out, and for catalogs that > own > >the storage, DROP also deletes the data. If the cluster is not reachable > > at all, the only workaround is to start a cluster under the recorded > >cluster ID so that DROP sees the job as gone. Is that acceptable, or > should > >a > > redeploy be able to recover such tables? > > > >Looking forward to your thoughts and feedback! > > > >Best, > >Timo > > > >[1] > > > https://cwiki.apache.org/confluence/spaces/FLINK/pages/451979479/FLIP-616+Materialized+Tables+support+for+Table+API > > > >[2] > > > https://cwiki.apache.org/confluence/spaces/FLINK/pages/294816797/FLIP-435+Introduce+a+New+Materialized+Table+for+Simplifying+Data+Pipelines >
