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 > >> >> > >> >
