Hi Xuyang, Thanks, good to hear resume() as the fallback works for you. Regarding the flag, I'd rather not like to add one at this point. RESUME on an ACTIVATED table is already a recovery-only call. While the job runs it is rejected, and for a job that has ended it already starts a new one, without deleting the existing table. A flag would add new syntax to SQL and the Table API for a rare case. I would say resume() should log a warning that names the cluster it checked, and the docs should state that a table should only be resumed once its old refresh job can no longer write. Let me know if that makes sense to you or if I overlooked a specific case here.
Best, Timo On Fri, Oct 16, 2026 at 5:19 AM Xuyang <[email protected]> wrote: > Hi, Timo. I don’t have any further concerns about the other points. > Regarding point 4, I think using resume() as a fallback to recover an > ACTIVATED table whose refresh job has disappeared is reasonable. Perhaps we > could provide an optional parameter similar to IGNORE NOT EXISTS or FORCE > (the exact name can be discussed and disabled by default), allowing users > to restart the job without deleting the existing table data. What do you > think? > > > > -- > > Best！ > Xuyang > > > > At 2026-10-15 23:27:38, "Timo Theusner" <[email protected]> wrote: > >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 > >> >
