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