Hi Xuyang, Hi Gustavo, Thank you for reviewing this FLIP, and the positive feedback!
If there are no objections, I would start the VOTE for it tomorrow. Best, Timo On Wed, Oct 21, 2026 at 8:56 AM Gustavo de Morais <[email protected]> wrote: > Hi Timo, > > Materialized Tables are a key part of an easier interface for creating > Flink jobs and having it exposed in TableAPI makes a lot of sense. I like > the detailed examples and the FLIP looks good to me. > > Thanks for working on it +1. > > Kind regards, > Gustavo > > On Wed, 21 Oct 2026 at 07:39, Xuyang <[email protected]> wrote: > > > Hi Timo, > > That makes sense to me. +1 > > > > > > > > > > > > -- > > > > Best！ > > Xuyang > > > > > > > > At 2026-10-19 17:12:58, "Timo Theusner" <[email protected]> wrote: > > >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 > > >> >> > > >> > > >
