Hi Xuyang, Thanks a lot for the confirmation, and for the very thorough back-and-forth throughout this discussion — it really helped sharpen the proposal, especially around the "arbitrary position" semantics, the planner-side contract, and the catalog persistence story. Much appreciated!
@all: with Xuyang's last round of feedback addressed, the FLIP has stabilized. If anyone — Timo, Sergey, Yunfeng, Jingsong, Ron, Shengkai, and others who have been following the thread — still has open concerns, please let me know in the next ~3 days. Otherwise I'll start the VOTE thread. Thanks again to everyone for the reviews! Best, FeatZhang On Mon, May 18, 2026 at 5:20 PM Xuyang <[email protected]> wrote: > Hi FeatZhang, > > Thanks for the update and the revised proposal! I don't have further > questions. Looking forward to seeing this move forward. > > > > -- > > Best！ > Xuyang > > > > 在 2026-05-16 13:44:20，"FeatZhang" <[email protected]> 写道： > >Hi Xuyang, > > > >Thanks a lot for the careful follow-ups — these are exactly the right > >questions to ask before we move to a vote. After re-checking the code > >paths you pointed at, I want to revise several statements I made in my > >previous reply, and tighten the FLIP accordingly. I'll go through them > >one by one. > > > >1. Semantics at any position in the pipeline > >============================================= > >Fully agree. To make this concrete, I'll add a new sub-section > >"Semantics at Different Pipeline Positions" to the FLIP, covering: > > > > - APPLY_WATERMARK on a base table (with or without an existing > > DDL watermark) > > - APPLY_WATERMARK on top of a non-materialized view / sub-query > > - APPLY_WATERMARK applied multiple times in the same query > > - Interaction with TUMBLE / HOP / SESSION / CUMULATE > > - Interaction with joins (regular / interval / temporal) > > > >The mental model will be explicitly aligned with the DataStream API: > >each APPLY_WATERMARK in SQL corresponds to one > >`assignTimestampsAndWatermarks(...)` call in the DataStream pipeline, > >applied in the order they appear. > > > >2. Monotonicity validation > >========================== > >You're right, and I want to correct my earlier reply. Today's > >`CREATE TABLE ... WATERMARK FOR ... AS ...` does NOT enforce > >monotonicity at the planner level — the planner only checks that: > > > > - the rowtime column exists and is of TIMESTAMP / TIMESTAMP_LTZ, > > - the watermark expression is a valid scalar expression over the > > table's schema and resolves to a comparable type. > > > >Monotonicity is a runtime contract: the WatermarkAssignerOperator > >emits watermarks that are guaranteed to be non-decreasing. > > > >APPLY_WATERMARK will follow exactly the same contract — no stricter > >planner-level monotonicity check. I'll update the FLIP's "Planner > >Changes → Validation" section to reflect this. > > > >3. Override timing — clarification > >================================== > >I think this was a wording issue on my side rather than a real > >design disagreement. Let me restate it: > > > > - APPLY_WATERMARK introduces a dedicated WatermarkAssigner node in > > the plan. Whether the input already carries a watermark or not, > > the plan ends up with the new assigner positioned downstream of > > the existing one. > > - At runtime there is no "merge" or "reconciliation": each > > WatermarkAssigner operator independently emits its own watermark > > stream; downstream operators simply observe the watermark from > > the most recent upstream assigner. > > > >This is the same model as calling `assignTimestampsAndWatermarks()` > >twice in DataStream — the second call wins because it sits later in > >the operator chain, not because of any planner-level magic. > > > >So "planner-level override" was a poor choice of words. The correct > >description is: **the planner decides the operator topology; the > >runtime emits watermarks according to that topology**, exactly like > >DataStream. I'll rephrase the FLIP accordingly and drop the > >"override" terminology. > > > >4. "Watermark expression evaluation: needs to support arbitrary > > expressions" > >==================================================================== > >Apologies, this statement was inaccurate. After re-checking, > >StreamExecWatermarkAssigner already evaluates the watermark > >expression through the standard `ExprCodeGenerator`, which supports > >the same scalar expressions as DDL today (arithmetic on TIMESTAMP / > >TIMESTAMP_LTZ, INTERVAL arithmetic, scalar UDFs, etc.). > > > >What APPLY_WATERMARK actually needs from the ExecNode is: > > > > - resolving the rowtime column index from the DESCRIPTOR, since the > > input may be a non-base-table (view / sub-query / projected > > relation), > > - wiring the watermark expression's input row to the upstream > > operator's output row instead of a TableScan output. > > > >No new expression capability is required. I'll fix this in the FLIP. > > > >5. State management > >=================== > >You're right, this should be removed. In the scope of this FLIP the > >APPLY_WATERMARK ExecNode is **stateless**, identical to the existing > >`WatermarkAssignerOperator`. It does not buffer rows and does not > >evict late data; late-data handling remains the responsibility of > >the downstream window / join operators, exactly as it works today. > > > >The "state management" bullet in my previous reply was speculation > >about future watermark strategies (idle source detection, etc.) and > >does not belong in this FLIP. I'll drop it. > > > >------------------------------------------------------------ > >Updated summary of the design after this round > >------------------------------------------------------------ > > > > - Scope: base tables, non-materialized views, sub-queries; any > > relation position in the query. > > - Semantics: aligned with DataStream API > > `assignTimestampsAndWatermarks()`; multiple applications in the > > same query are positional, not "overriding". > > - Validation: same contract as today's DDL — scalar expression on > > a TIMESTAMP / TIMESTAMP_LTZ rowtime column; no planner-level > > monotonicity check. > > - ExecNode: stateless, reuses StreamExecWatermarkAssigner with > > minor wiring changes (rowtime column resolution + non-TableScan > > input handling). > > - Out of scope: WatermarkFunction interface, idle-source state, > > runtime-level merge of multiple watermark strategies. > > > >I'll update the FLIP document and PR #27984 to match the points > >above, and post a diff summary here once it's done. > > > >Thanks again for pushing on these — the FLIP is much cleaner after > >this round. > > > >Best regards, > >FeatZhang > > > >On Mon, May 11, 2026 at 10:45 AM Xuyang <[email protected]> wrote: > > > >> Hi FeatZhang. Thanks for the detailed responses. I have a few follow-up > >> comments and questions: > >> > >> > >> 1. Support for APPLY_WATERMARK at any node/position > >> I generally agree with the direction that APPLY_WATERMARK should be > >> applicable at any node and any position — one or more times — similar to > >> how the DataStream API allows watermark assignment. However, I think we > >> need to clearly articulate the precise behavior/semantics in each > scenario > >> to reduce user confusion. Aligning the mental model with the DataStream > API > >> (where users can call assignTimestampsAndWatermarks() at arbitrary > points > >> in the pipeline) would also help lower the learning curve. > >> 2. Monotonicity validation of watermark expressions > >> Why do we need to enforce monotonicity guarantees on the watermark > >> expression at the planner level? As far as I know, Flink SQL currently > does > >> NOT perform such validation at the DDL level for CREATE TABLE ... > WATERMARK > >> FOR ... AS .... What it actually does is ensure at runtime that emitted > >> watermarks are non-decreasing. If the existing DDL path does not > validate > >> monotonicity at planning time, why should APPLY_WATERMARK introduce a > >> stricter contract? > >> 3. Planner-level watermark override > >> Could you elaborate more on why the watermark override must happen at > the > >> planner level? If I understand correctly, in the DataStream API, users > can > >> define different watermark strategies at different nodes in the same > >> pipeline, and the runtime handles watermark propagation naturally. > >> 4. "Watermark expression evaluation: Needs to support arbitrary > >> expressions" > >> You mentioned that StreamExecWatermarkAssigner currently has limitations > >> in watermark expression evaluation and "needs to support arbitrary > >> expressions." Could you clarify what the current limitations are > exactly? > >> Today's CREATE TABLE ... WATERMARK FOR ts AS ts - INTERVAL '5' SECOND > >> already supports scalar expressions, what additional expression types > does > >> APPLY_WATERMARK require that are not already supported? > >> 5. State management > >> You mentioned "State management". Are we proposing to introduce a > stateful > >> watermark assigner node that evicts late data? This sounds like a > >> significant change that goes well beyond the scope of this FLIP. The > >> current WatermarkAssigner is stateless, it simply computes and emits > >> watermarks without buffering data. > >> Looking forward to your clarification! > >> > >> > >> > >> > >> > >> -- > >> > >> Best！ > >> Xuyang > >> > >> > >> > >> At 2026-05-09 19:08:32, "熊饶饶" <[email protected]> wrote: > >> >Hi Feat Zhang, > >> > > >> >+1 (non-binding) > >> > > >> >This is a well-motivated proposal that addresses a real pain point in > >> Flink SQL. The inability to define watermarks in views currently forces > >> users to either: > >> > > >> >Reference underlying table watermarks directly, breaking encapsulation > >> >Create separate physical tables for each watermark strategy, leading to > >> data duplication > >> >The proposed syntax CREATE VIEW ... WATERMARK FOR col AS expr is > >> intuitive and aligns naturally with existing DDL watermark semantics. > The > >> backward-compatible design (storing watermark metadata in catalog > options) > >> is also a smart choice — it avoids breaking existing views while > enabling > >> the new capability. > >> > > >> >The Data Lakehouse / medallion architecture use case is particularly > >> compelling. Being able to define watermark strategies at the Silver/Gold > >> layer while keeping Bronze as raw data would significantly simplify > >> pipeline design for many teams. > >> > > >> >Looking forward to seeing this move to a formal FLIP! > >> > > >> >Best regards, > >> >Raorao Xiong > >> > > >> >> 2026年5月7日 20:56，FeatZhang <[email protected]> 写道： > >> >> > >> >> Hi Xuyang, > >> >> > >> >> Thank you for the thorough review and thoughtful questions. > >> >> > >> >> *Problem this FLIP aims to solve*: The core goal of this FLIP is to > >> allow > >> >> watermarks to be defined on *computed columns (and, more generally, > on > >> any > >> >> column produced inside a SQL query) directly in SQL statements*. > Today, > >> >> watermarks in Flink SQL can only be declared at the CREATE TABLE > level > >> >> via WATERMARK > >> >> FOR ... AS ..., which means the time attribute must be a column > visible > >> at > >> >> the base table DDL. This makes it impossible to attach a watermark > to a > >> >> timestamp derived inside a query — for example, one computed by > string > >> >> parsing, JSON extraction, or any expression inside a view or > subquery — > >> >> without pushing that computation back down into the source table > DDL. By > >> >> introducing APPLY_WATERMARK as an explicit relational operator that > can > >> be > >> >> applied to *base tables, non-materialized views, and subqueries*, > users > >> can > >> >> assign watermark semantics to any (computed) column produced by a > query, > >> >> which also addresses the broader motivations listed in the FLIP: > broken > >> >> layered-pipeline abstractions, lack of per-query / multi-tenant > >> watermark > >> >> strategies, and the current gap between SQL and the DataStream API. > >> >> > >> >> Now let me address each of your points: > >> >> Support for Non-materialized Views > >> >> > >> >> You raised an excellent point about the scope of APPLY_WATERMARK in > >> layered > >> >> architectures. > >> >> > >> >> *My position*: APPLY_WATERMARK should support base tables, > >> non-materialized > >> >> views, *and subqueries* (this matches the Goals section of the FLIP). > >> >> Here's why: > >> >> > >> >> - Non-materialized views dissolve into the surrounding plan during > >> >> optimization (inline expansion) > >> >> - There's no physical "view" node in the execution plan—just a > logical > >> >> alias > >> >> - The watermark becomes a relational transformation applied on top > of > >> >> the view's / subquery's output > >> >> > >> >> The key design principle is: *watermark definition is an explicit > >> >> relational operator, not attached metadata*. > >> >> > >> >> I'd also like to clarify the positions in the prior thread to avoid > >> >> confusion: > >> >> > >> >> - *Lincoln* originally proposed APPLY_WATERMARK(table, > >> DESCRIPTOR(col), > >> >> expr) scoped to base tables only. > >> >> - *Timo* raised the concern about blurring the view abstraction > ("*A > >> >> view usually dissolves into the plan … would a watermark definition > >> >> suddenly introduce an optimization barrier? If this is an > optimization > >> >> barrier, is this still a view or a new concept?*"). This is exactly > >> why > >> >> this FLIP does *not* attach watermarks to views via CREATE VIEW / > >> ALTER > >> >> VIEW, and keeps views as pure logical aliases. > >> >> - *Gyula* emphasized that watermark assignment should be available > on > >> >> views and subqueries too, consistent with the DataStream API. > >> >> > >> >> To address Timo's concern concretely: > >> >> > >> >> - Watermark semantics are applied at query planning time via an > >> explicit > >> >> relational operator (APPLY_WATERMARK), not hidden in view/catalog > >> metadata. > >> >> - No watermark information is persisted into the catalog for views > — > >> the > >> >> catalog stays unchanged (see FLIP "Catalog Changes": *No catalog > >> changes > >> >> are required*). > >> >> - Views continue to dissolve transparently into the plan; the > >> >> optimization barrier only appears where APPLY_WATERMARK is > explicitly > >> used. > >> >> > >> >> Monotonicity Validation > >> >> > >> >> Great question! Monotonicity guarantees are essential for watermark > >> >> correctness: > >> >> > >> >> - Watermarks define the boundary of "late" data > >> >> - If the watermark expression is not monotonically non-decreasing, > the > >> >> watermark could move backward > >> >> - This would cause data that was previously considered "on-time" > to be > >> >> treated as late (or vice-versa), breaking event-time semantics > >> >> > >> >> *Validation requirement*: In line with the FLIP's Planner Changes > >> section, > >> >> the planner validates that watermark_expression is a valid *scalar* > >> expression > >> >> over columns of the input schema, and (as with today's CREATE TABLE > ... > >> >> WATERMARK FOR ... AS ...) the expression must produce a monotonically > >> >> non-decreasing value relative to the designated rowtime column. > >> >> > >> >> Typical valid forms are the same as what's allowed in DDL today, for > >> >> example: > >> >> > >> >> -- Bounded out-of-orderness (most common) > >> >> APPLY_WATERMARK(t, DESCRIPTOR(ts), ts - INTERVAL '5' SECOND) > >> >> -- Strictly ascending > >> >> APPLY_WATERMARK(t, DESCRIPTOR(ts), ts) > >> >> > >> >> Note: watermark_expression is a *scalar expression* per the FLIP > (not an > >> >> aggregate / window function). Richer forms such as user-defined > >> watermark > >> >> strategies are explicitly out of scope for this FLIP and are tracked > as > >> a > >> >> future WatermarkFunction interface, which also depends on the Calcite > >> >> lambda upgrade mentioned by Timo. > >> >> Override Timing (Planner vs Runtime) > >> >> > >> >> You raised a valid concern. Let me clarify the design, which aligns > with > >> >> the FLIP's "Planner Changes → Interaction with Existing Table > >> Watermarks" > >> >> section: > >> >> > >> >> *Current proposal*: Planner-level override > >> >> > >> >> - During query compilation, when the input to APPLY_WATERMARK > already > >> >> carries a watermark (e.g., from CREATE TABLE ... WATERMARK), the > >> >> LogicalWatermarkAssigner node produced by APPLY_WATERMARK > *overrides* > >> the > >> >> upstream watermark strategy. > >> >> - When the input has no watermark (e.g., a view or a subquery), > >> >> APPLY_WATERMARK introduces a new one. > >> >> - This makes override behavior explicit in the plan and keeps room > for > >> >> standard optimizations. > >> >> > >> >> *Why not runtime-level override*: > >> >> > >> >> 1. Planner-level override keeps watermark semantics a first-class, > >> >> visible part of the plan (consistent with how VECTOR_SEARCH / > >> ML_PREDICT > >> >> are modeled as specialized ExecNodes in the FLIP). > >> >> 2. The override point is deterministic and inspectable via EXPLAIN. > >> >> 3. Simpler execution model — no dual-watermark reconciliation at > >> runtime. > >> >> > >> >> If concrete use cases for a runtime-level override surface later, we > >> could > >> >> revisit this via a hint (e.g., /*+ RUNTIME_OVERRIDE */), but it's not > >> part > >> >> of this FLIP. > >> >> Relationship with BuiltInProcessTableFunction > >> >> > >> >> Good observation. APPLY_WATERMARK is declared as a built-in PTF at > the > >> SQL > >> >> surface, similar in spirit to TO_CHANGELOG / FROM_CHANGELOG, but the > >> FLIP > >> >> intentionally maps it to a *specialized ExecNode* rather than a > generic > >> PTF > >> >> runtime — the same pattern used by VECTOR_SEARCH and ML_PREDICT. > >> >> > >> >> *Option A: Reuse the generic BuiltInProcessTableFunction runtime* > >> >> > >> >> - Pros: Consistent with other built-in PTFs at the runtime layer. > >> >> - Cons: Watermark assignment is not a row-transforming PTF — it > >> changes > >> >> stream metadata (time attribute + watermark strategy). Forcing it > >> through > >> >> the generic PTF runtime would require extending the PTF contract > with > >> >> watermark semantics. > >> >> > >> >> *Option B: Dedicated LogicalWatermarkAssigner + specialized ExecNode > >> (the > >> >> FLIP's choice)* > >> >> > >> >> - Pros: Keeps watermark semantics a first-class citizen in the > >> planner; > >> >> cleanly integrates with watermark propagation rules; no need to > >> overload > >> >> the PTF contract; same pattern as VECTOR_SEARCH / ML_PREDICT > already > >> >> established in Flink. > >> >> - Cons: A new dedicated node, though that cost is small compared to > >> the > >> >> semantic clarity. > >> >> > >> >> *Current decision*: Option B, as stated in the FLIP > ("*APPLY_WATERMARK > >> >> compiles to a specialized ExecNode --- similar to how VECTOR_SEARCH > and > >> >> ML_PREDICT are handled*"). Open to revisiting based on community > >> feedback. > >> >> StreamExecWatermarkAssigner Sufficiency > >> >> > >> >> For the physical implementation: > >> >> > >> >> *Yes, StreamExecWatermarkAssigner should be sufficient*, with some > >> >> modifications: > >> >> > >> >> 1. *Input handling*: Currently assumes direct table scan; needs to > >> >> handle APPLY_WATERMARK's column mapping > >> >> 2. *Watermark expression evaluation*: Needs to support arbitrary > >> >> expressions (currently limited) > >> >> 3. *State management*: May need additional state for handling > >> >> out-of-order events > >> >> > >> >> The key insight is that APPLY_WATERMARK conceptually translates to: > >> >> > >> >> TableScan -> Calc (expression evaluation) -> WatermarkAssigner > >> >> > >> >> StreamExecWatermarkAssigner handles the last step; the Calc step > handles > >> >> the expression. > >> >> ------------------------------ > >> >> Summary of Proposed Responses > >> >> > >> >> - *Scope of input*: Support base tables, non-materialized views > *and > >> >> subqueries* (per FLIP Goals); views/catalog semantics stay > unchanged. > >> >> - *Monotonicity*: Validate watermark_expression as a scalar > >> expression; > >> >> same monotonicity contract as today's DDL watermarks. > >> >> - *Override timing*: Planner-level override at > >> LogicalWatermarkAssigner; > >> >> potential /*+ RUNTIME_OVERRIDE */ hint as future work. > >> >> - *PTF reuse*: Dedicated LogicalWatermarkAssigner + specialized > >> ExecNode > >> >> (same pattern as VECTOR_SEARCH / ML_PREDICT). > >> >> - *ExecNode sufficiency*: StreamExecWatermarkAssigner is sufficient > >> with > >> >> minor modifications (input handling + expression evaluation). > >> >> > >> >> ------------------------------ > >> >> > >> >> Looking forward to further discussion! > >> >> > >> >> Best regards, > >> >> FeatZhang > >> >> > >> >> > >> >> On Wed, May 6, 2026 at 4:39 PM Xuyang <[email protected]> wrote: > >> >> > >> >>> Hi, FeatZhang. Thanks for driving this discussion. I've read through > >> the > >> >>> full FLIP and the mailing list context, and I have a few questions: > >> >>> 1. If I understand correctly, in a Layered Data Architecture, > >> >>> silver_events would typically be a table, a materialized view, or a > >> >>> materialized table. From the mailing list discussion, it seems like > no > >> >>> consensus was reached on this point. I think we still need to > consider > >> >>> whether APPLY_WATERMARK should be allowed on (non-materialized) > views. > >> >>> 2. In the Planner Changes section under Logical Plan, could you > >> elaborate > >> >>> on why monotonicity guarantees need to be ensured for the watermark > >> >>> expression validation? > >> >>> 3. (nit) In the Watermark Override part under Planner Changes, > >> shouldn't > >> >>> the override of the upstream watermark happen at runtime rather than > >> at the > >> >>> planner level? > >> >>> 4. I feel that APPLY_WATERMARK is quite similar to TO_CHANGELOG and > >> >>> FROM_CHANGELOG. Is what we actually need just a > >> >>> BuiltInProcessTableFunction? That way, we would only need to further > >> extend > >> >>> ProcessTableFunction to support this. > >> >>> 5. If we choose to translate APPLY_WATERMARK into a specialized > >> ExecNode > >> >>> (similar to VECTOR_SEARCH and ML_PREDICT), would the existing > >> >>> StreamExecWatermarkAssigner be sufficient for this purpose? > >> >>> > >> >>> > >> >>> > >> >>> > >> >>> > >> >>> -- > >> >>> > >> >>> Best！ > >> >>> Xuyang > >> >>> > >> >>> > >> >>> > >> >>> At 2026-04-21 22:22:33, "FeatZhang" <[email protected]> wrote: > >> >>>> Hi everyone, > >> >>>> > >> >>>> Thank you for the feedback and discussions on the initial proposal. > >> I've > >> >>>> revised the FLIP based on the community's input and would like to > >> share > >> >>>> the updated version. > >> >>>> > >> >>>> > >> >>>> FLIP-XXX: Support Flexible Watermark Assignment via Built-in > Function > >> >>>> < > >> >>> > >> > https://drive.google.com/open?id=17PXYAi6Pb91OqFhVVK7tRULiaHAb6wiX79jjC0NaaDA > >> >>>> > >> >>>> > >> >>>> > >> >>>> KEY UPDATES > >> >>>> =========== > >> >>>> > >> >>>> The proposal has evolved from "Support Watermark Definition in SQL > >> Views" > >> >>>> to a more flexible and powerful approach: FLIP-XXX: Support > Flexible > >> >>>> Watermark Assignment via APPLY_WATERMARK Function. > >> >>>> > >> >>>> What's Changed: > >> >>>> > >> >>>> 1. Broader Scope: Instead of limiting watermark definitions to SQL > >> views > >> >>>> only, the new proposal introduces a built-in table function > >> >>>> APPLY_WATERMARK that works with: > >> >>>> - Base tables > >> >>>> - Views (both regular and materialized) > >> >>>> - Subqueries > >> >>>> - Any table-valued expressions > >> >>>> > >> >>>> 2. More Flexible Design: The function-based approach provides: > >> >>>> - Dynamic watermark assignment at query time without modifying > >> catalog > >> >>>> metadata > >> >>>> - Override capability for existing watermark strategies > >> >>>> - Composability with other SQL operations > >> >>>> - No need for DDL changes or catalog write permissions > >> >>>> > >> >>>> 3. Better SQL Semantics: Using a table function follows SQL > standard > >> >>>> patterns and integrates naturally with Flink's existing function > >> >>>> ecosystem. > >> >>>> > >> >>>> UPDATED FLIP DOCUMENT > >> >>>> ===================== > >> >>>> > >> >>>> The revised FLIP is now available at: > >> >>>> https://iwiki.woa.com/p/4019879693 > >> >>>> > >> >>>> Key sections include: > >> >>>> - Motivation and use cases > >> >>>> - Public interfaces and SQL syntax > >> >>>> - Implementation plan > >> >>>> - Compatibility analysis > >> >>>> - Test plan > >> >>>> > >> >>>> EXAMPLE USAGE > >> >>>> ============= > >> >>>> > >> >>>> -- Apply watermark to a view > >> >>>> SELECT *FROM APPLY_WATERMARK(my_view, DESCRIPTOR(event_time), > >> >>>> event_time - INTERVAL '5' SECOND); > >> >>>> -- Override existing watermark strategy > >> >>>> SELECT *FROM APPLY_WATERMARK(my_table_with_watermark, > DESCRIPTOR(ts), > >> >>>> ts - INTERVAL '10' SECOND -- Different from DDL watermark > >> >>>> ); > >> >>>> -- Use in complex queries > >> >>>> SELECT window_start, > >> >>>> window_end, > >> >>>> COUNT(*)FROM TABLE(TUMBLE(TABLE APPLY_WATERMARK(orders, > >> >>>> DESCRIPTOR(order_time), order_time - INTERVAL '5' SECOND), > >> >>>> DESCRIPTOR(order_time), INTERVAL '1' HOUR))GROUP BY window_start, > >> >>>> window_end; > >> >>>> > >> >>>> > >> >>>> IMPLEMENTATION PROGRESS > >> >>>> ======================= > >> >>>> > >> >>>> I've also opened a draft PR #27984 with the initial implementation: > >> >>>> - Core built-in function definition > >> >>>> - SQL-to-RelNode conversion rules > >> >>>> - Physical plan integration > >> >>>> - Unit tests and documentation (English + Chinese) > >> >>>> > >> >>>> The PR is available at: > >> >>>> https://github.com/apache/flink/pull/27984 > >> >>>> > >> >>>> REQUEST FOR FEEDBACK > >> >>>> ==================== > >> >>>> > >> >>>> I would appreciate your thoughts on: > >> >>>> > >> >>>> 1. Function naming: Is APPLY_WATERMARK clear and intuitive? > >> >>>> (Alternative considered: WITH_WATERMARK, SET_WATERMARK) > >> >>>> > >> >>>> 2. DESCRIPTOR syntax: Using DESCRIPTOR(column_name) to specify the > >> >>>> rowtime column—does this align well with Flink's existing > patterns? > >> >>>> > >> >>>> 3. Override behavior: Should APPLY_WATERMARK always override > existing > >> >>>> watermarks, or should we provide a mode parameter > >> >>>> (e.g., OVERRIDE, MERGE)? > >> >>>> > >> >>>> 4. Performance considerations: Any concerns about the > function-based > >> >>>> approach vs. catalog-level watermark definitions? > >> >>>> > >> >>>> > >> >>>> > >> >>>> > >> >>>> > >> >>>> Looking forward to your valuable feedback! > >> >>>> > >> >>>> Best regards, > >> >>>> FeatZhang > >> >>>> > >> >>>> On Thu, Feb 12, 2026 at 6:24 PM Timo Walther <[email protected]> > >> wrote: > >> >>>> > >> >>>>> Hi everyone, > >> >>>>> > >> >>>>> I think we all agree that we clearly want this functionality, just > >> the > >> >>>>> "how" needs to be discussed. I also like Lincoln’s suggestion of > >> >>>>> introducing a built-in PTF for this, I had similar ideas in mind. > >> >>>>> > >> >>>>> There are two issues with a APPLY_WATERMARK function, but both on > the > >> >>>>> short-term roadmap: > >> >>>>> > >> >>>>> 1) This function would need to be a function that takes an > >> expression. > >> >>>>> Ideally as a lambda function. Newer Calcite versions have already > >> lambda > >> >>>>> expression support. At Confluent we were planning to work on a > >> Calcite > >> >>>>> upgrade this quarter especially to get lambda support in and > improve > >> >>>>> built-in functions that work on collections. > >> >>>>> > >> >>>>> 2) User-defined PTFs are currently not able to emit watermarks. We > >> could > >> >>>>> introduce a new interface WatermarkFunction (similar to > >> >>>>> ChangelogFunction) that would offer this to everyone. > Alternatively, > >> we > >> >>>>> could only use the PTF signature, but translate to a specialized > >> >>>>> ExecNode similar how we do it for VECTOR_SEARCH and ML_PREDICT. > >> >>>>> > >> >>>>> In any case, even if we go with the function approach, we > definitely > >> >>>>> need a full FLIP on this. > >> >>>>> > >> >>>>> Thanks, > >> >>>>> Timo > >> >>>>> > >> >>>>> On 12.02.26 08:25, Gyula Fóra wrote: > >> >>>>>> Hi All! > >> >>>>>> I would like to chime in here quickly from a slightly different > >> angle. > >> >>>>>> While I am the first to admit that I cannot grasp all the > planning / > >> >>>>>> conceptual implications, I also feel the need for more flexible > >> >>> watermark > >> >>>>>> handling as suggested by Feat. > >> >>>>>> > >> >>>>>> Anything that can only be applied to base/catalog tables is very > >> >>> limiting > >> >>>>>> from a usability perspective. Watermarks feel like they should > be a > >> >>>>> simple > >> >>>>>> function that you can apply on a column/table as part of a > >> query/view. > >> >>>>> For > >> >>>>>> example extract timestamp from a string convert to TS -> apply > >> >>> watermark > >> >>>>>> etc. > >> >>>>>> > >> >>>>>> Users often receive the tables/catalogs as given and can only > write > >> >>>>>> queries. > >> >>>>>> > >> >>>>>> Fixing this would eliminate a long standing disconnect between > the > >> >>>>>> datastream api flexible watermark handling compared to the > currently > >> >>> very > >> >>>>>> restrictive SQL approach. > >> >>>>>> > >> >>>>>> Cheers > >> >>>>>> Gyula > >> >>>>>> > >> >>>>>> On Thu, Feb 12, 2026 at 7:54 AM FeatZhang <[email protected]> > >> >>> wrote: > >> >>>>>> > >> >>>>>>> Hi Timo, Lincoln, > >> >>>>>>> > >> >>>>>>> Thank you both for the detailed feedback. > >> >>>>>>> > >> >>>>>>> I agree with the concern that non-materialized SQL views should > >> >>> remain a > >> >>>>>>> pure logical abstraction. Introducing watermark definitions > >> directly > >> >>>>>>> into CREATE > >> >>>>>>> VIEW or ALTER VIEW could blur the boundary between logical > aliasing > >> >>> and > >> >>>>>>> physical planning semantics, especially considering optimization > >> >>>>> barriers > >> >>>>>>> and watermark propagation behavior. > >> >>>>>>> > >> >>>>>>> Lincoln’s suggestion of introducing a built-in function such as: > >> >>>>>>> > >> >>>>>>> APPLY_WATERMARK(table, DESCRIPTOR(rowtime_column), > >> >>> watermark_expression) > >> >>>>>>> > >> >>>>>>> is a cleaner direction. It keeps watermark definition as an > >> explicit > >> >>>>>>> relational transformation rather than attaching additional > >> semantics > >> >>> to > >> >>>>>>> views. > >> >>>>>>> > >> >>>>>>> However, to fully address the original use cases (especially > >> logical > >> >>>>> reuse > >> >>>>>>> and layered lakehouse architectures), I propose that the table > >> >>> parameter > >> >>>>>>> should support: > >> >>>>>>> > >> >>>>>>> - Base tables > >> >>>>>>> - Non-materialized views > >> >>>>>>> > >> >>>>>>> If APPLY_WATERMARK can accept both, we can: > >> >>>>>>> > >> >>>>>>> - Preserve the conceptual purity of SQL views > >> >>>>>>> - Avoid redefining view semantics > >> >>>>>>> - Still enable logical reuse via views > >> >>>>>>> - Allow different watermark strategies over the same logical > >> >>>>> relation > >> >>>>>>> > >> >>>>>>> In other words, watermark definition becomes an explicit > relational > >> >>>>>>> operator applied on top of any logical relation, instead of > being > >> >>>>> embedded > >> >>>>>>> into the view definition itself. > >> >>>>>>> > >> >>>>>>> From a planner perspective, this keeps the model consistent: > >> >>>>>>> > >> >>>>>>> - The function expands into a relational node > >> >>>>>>> - No optimization barrier is introduced by views > >> >>>>>>> - Watermark handling remains part of the logical plan > >> >>> transformation > >> >>>>>>> > >> >>>>>>> I will prepare a PR to prototype APPLY_WATERMARK with support > for > >> >>> both > >> >>>>> base > >> >>>>>>> tables and non-materialized views, and share it for further > >> >>> discussion. > >> >>>>>>> > >> >>>>>>> Looking forward to your thoughts. > >> >>>>>>> > >> >>>>>>> Best, > >> >>>>>>> Feat > >> >>>>>>> > >> >>>>>>> > >> >>>>>>> Lincoln Lee <[email protected]> 于2026年2月12日周四 12:19写道： > >> >>>>>>> > >> >>>>>>>> Agree with Timo’s point regarding the conceptual semantics. We > >> >>> should > >> >>>>> not > >> >>>>>>>> directly extend non-materialized views with additional > watermark > >> >>>>>>>> definitions. > >> >>>>>>>> > >> >>>>>>>> Regarding the use case mentioned by Feat, defining different > >> >>> watermark > >> >>>>>>>> strategies > >> >>>>>>>> for the same data source, especially in the case of catalog > >> tables, > >> >>> we > >> >>>>>>> are > >> >>>>>>>> exploring a possible solution introducing a built-in function: > >> >>>>>>>> ```sql > >> >>>>>>>> APPLY_WATERMARK(table, DESCRIPTOR(rowtime_column), > >> >>>>> watermark_expression) > >> >>>>>>>> ``` > >> >>>>>>>> This function only support base tables as input and not support > >> >>> views, > >> >>>>>>>> subqueries > >> >>>>>>>> or derived relations. > >> >>>>>>>> > >> >>>>>>>> This would address a meaningful subset of the identified use > cases > >> >>>>>>> without > >> >>>>>>>> redefining the role of SQL views. > >> >>>>>>>> > >> >>>>>>>> > >> >>>>>>>> Best, > >> >>>>>>>> Lincoln Lee > >> >>>>>>>> > >> >>>>>>>> > >> >>>>>>>> Timo Walther <[email protected]> 于2026年2月11日周三 23:29写道： > >> >>>>>>>> > >> >>>>>>>>> Hi Feat, > >> >>>>>>>>> > >> >>>>>>>>> thanks for proposing this FLIP. We had similar discussions in > the > >> >>>>> past, > >> >>>>>>>>> but so far could never reach consensus. > >> >>>>>>>>> > >> >>>>>>>>> SQL views are actually a very simple concept, they just give > SQL > >> >>> text > >> >>>>>>> an > >> >>>>>>>>> alias. A view has no other properties except for the view > >> >>> definition. > >> >>>>>>>>> Everything else is dynamically computed when the SQL text is > >> >>> inserted > >> >>>>>>>>> into the larger plan. A view is never evaluated without the > >> >>>>> surrounding > >> >>>>>>>>> plan. > >> >>>>>>>>> > >> >>>>>>>>> Watermarks in the middle of a pipeline raise a couple of > tricky > >> >>>>> issues: > >> >>>>>>>>> > >> >>>>>>>>> - What if the upstream table is updating, how would you deal > with > >> >>>>>>>>> watermarks in the downstream view? > >> >>>>>>>>> - What if the upstream table emits already watermarks? Would > the > >> >>> view > >> >>>>>>>>> catch them and discard this information? > >> >>>>>>>>> - A view usually dissolves into the plan (e.g. via projection > or > >> >>>>> filter > >> >>>>>>>>> pushdown). Would a watermark definition suddenly introduce an > >> >>>>>>>>> optimization barrier? If this is an optimization barrier, is > this > >> >>>>> still > >> >>>>>>>>> a view or a new concept? E.g. a "materialized view" or > >> "pre-planned > >> >>>>>>>> view"? > >> >>>>>>>>> > >> >>>>>>>>> Cheers, > >> >>>>>>>>> Timo > >> >>>>>>>>> > >> >>>>>>>>> > >> >>>>>>>>> > >> >>>>>>>>> On 11.02.26 03:51, FeatZhang wrote: > >> >>>>>>>>>> Hi Flink Community, > >> >>>>>>>>>> > >> >>>>>>>>>> I'd like to propose adding watermark support for SQL Views in > >> >>> Flink > >> >>>>>>> to > >> >>>>>>>>>> better support event-time processing scenarios. > >> >>>>>>>>>> Problem Statement > >> >>>>>>>>>> > >> >>>>>>>>>> Currently, Flink SQL views cannot define watermarks. This > >> >>> limitation > >> >>>>>>>>>> creates several challenges: > >> >>>>>>>>>> > >> >>>>>>>>>> - *Broken Abstraction*: Users must reference underlying > >> table > >> >>>>>>>>> watermarks > >> >>>>>>>>>> directly, exposing implementation details > >> >>>>>>>>>> - *No Flexibility*: Cannot define different watermark > >> >>> strategies > >> >>>>>>>> for > >> >>>>>>>>>> different use cases on the same data source > >> >>>>>>>>>> - *Limited Architecture Support*: Incompatible with > modern > >> >>>>>>> layered > >> >>>>>>>>> data > >> >>>>>>>>>> architectures (Bronze/Silver/Gold medallion pattern) > >> >>>>>>>>>> > >> >>>>>>>>>> For example, in multi-tenant scenarios, different tenants may > >> >>> require > >> >>>>>>>>>> different lateness tolerance, but currently we cannot create > >> views > >> >>>>>>> with > >> >>>>>>>>>> different watermark strategies on the same source table. > >> >>>>>>>>>> Proposed Solution > >> >>>>>>>>>> > >> >>>>>>>>>> I propose adding two SQL syntax options to support watermark > >> >>>>>>>> definitions > >> >>>>>>>>> in > >> >>>>>>>>>> views: > >> >>>>>>>>>> > >> >>>>>>>>>> *Option 1: CREATE VIEW with WATERMARK* > >> >>>>>>>>>> > >> >>>>>>>>>> CREATE VIEW user_activity > >> >>>>>>>>>> WATERMARK FOR event_time AS event_time - INTERVAL '5' > SECONDAS > >> >>> SELECT > >> >>>>>>>>>> user_id, event_time, action FROM raw_events; > >> >>>>>>>>>> > >> >>>>>>>>>> *Option 2: ALTER VIEW SET WATERMARK* > >> >>>>>>>>>> > >> >>>>>>>>>> ALTER VIEW user_activity SET WATERMARK FOR event_time AS > >> >>> event_time - > >> >>>>>>>>>> INTERVAL '5' SECOND; > >> >>>>>>>>>> > >> >>>>>>>>>> Key Design Aspects > >> >>>>>>>>>> > >> >>>>>>>>>> - *Backward Compatibility*: Watermark stored as optional > >> >>>>> metadata > >> >>>>>>>> in > >> >>>>>>>>>> view options; existing views continue to work unchanged > >> >>>>>>>>>> - *Validation*: Watermark column must exist in view > schema > >> >>> and > >> >>>>> be > >> >>>>>>>> of > >> >>>>>>>>>> TIMESTAMP/TIMESTAMP_LTZ type > >> >>>>>>>>>> - *Storage*: Watermark metadata persists in catalog > options > >> >>> map > >> >>>>>>>>> (works > >> >>>>>>>>>> with all catalog implementations) > >> >>>>>>>>>> - *Propagation*: Follows existing Flink watermark > >> propagation > >> >>>>>>> rules > >> >>>>>>>>> in > >> >>>>>>>>>> joins and nested views > >> >>>>>>>>>> > >> >>>>>>>>>> Use Case Example: Data Lakehouse Architecture > >> >>>>>>>>>> > >> >>>>>>>>>> -- Bronze: Raw data (no watermark)CREATE TABLE bronze_events > >> >>>>>>> (raw_data > >> >>>>>>>>>> STRING, ingestion_time TIMESTAMP(3)) WITH (...); > >> >>>>>>>>>> -- Silver: Cleaned data with watermarkCREATE VIEW > silver_events > >> >>>>>>>>>> WATERMARK FOR event_time AS event_time - INTERVAL '10' > SECONDAS > >> >>>>>>> SELECT > >> >>>>>>>>>> CAST(JSON_VALUE(raw_data, '$.event_time') AS > TIMESTAMP(3)) > >> >>> AS > >> >>>>>>>>> event_time, > >> >>>>>>>>>> JSON_VALUE(raw_data, '$.user_id') AS user_idFROM > >> >>>>>>>>>> bronze_eventsWHERE JSON_VALUE(raw_data, '$.event_time') IS > NOT > >> >>> NULL; > >> >>>>>>>>>> -- Gold: AggregationsSELECT TUMBLE_START(event_time, INTERVAL > >> '1' > >> >>>>>>>>>> HOUR), COUNT(*)FROM silver_eventsGROUP BY TUMBLE(event_time, > >> >>> INTERVAL > >> >>>>>>>>>> '1' HOUR); > >> >>>>>>>>>> > >> >>>>>>>>>> Reference Materials > >> >>>>>>>>>> > >> >>>>>>>>>> - FLIP Document: FLIP-XXX: Support Watermark in Flink SQL > >> >>> View > >> >>>>>>>>>> < > >> >>>>>>>>> > >> >>>>>>>> > >> >>>>>>> > >> >>>>> > >> >>> > >> > https://docs.google.com/document/d/1OBGTi3Xb-Kpcf_nHeKA30XiRPEKV4PU7FpieXQhWn7Y/edit?usp=sharing > >> >>>>>>>>>> > >> >>>>>>>>>> - JIRA Issue: > >> >>> https://issues.apache.org/jira/browse/FLINK-39062 > >> >>>>>>>>>> - Implementation POC: > >> >>>>>>>>>> - [FLINK-39062][table] Support WATERMARK clause in CREATE > >> >>> VIEW > >> >>>>>>>>> statement > >> >>>>>>>>>> <https://github.com/apache/flink/pull/27571> > >> >>>>>>>>>> - [FLINK-39062][table] Support ALTER VIEW SET WATERMARK > >> >>> syntax > >> >>>>>>>>>> <https://github.com/apache/flink/pull/27570> > >> >>>>>>>>>> > >> >>>>>>>>>> Implementation Timeline > >> >>>>>>>>>> > >> >>>>>>>>>> Estimated 6-8 weeks covering parser layer, planner layer, > >> catalog > >> >>>>>>>>>> integration, and comprehensive testing. > >> >>>>>>>>>> Request for Feedback > >> >>>>>>>>>> > >> >>>>>>>>>> This enhancement would significantly improve Flink's support > for > >> >>>>>>>> layered > >> >>>>>>>>>> data architectures and flexible event-time processing. I'm > happy > >> >>> to > >> >>>>>>>>> provide > >> >>>>>>>>>> more details or start a formal FLIP process if the community > >> sees > >> >>>>>>> value > >> >>>>>>>>> in > >> >>>>>>>>>> this proposal. > >> >>>>>>>>>> > >> >>>>>>>>>> Looking forward to the community's feedback! > >> >>>>>>>>>> > >> >>>>>>>>>> Best regards, > >> >>>>>>>>>> > >> >>>>>>>>>> Feat Zhang > >> >>>>>>>>>> > >> >>>>>>>>>> FLIP-XXX: Support Watermark in Flink SQL View > >> >>>>>>>>>> < > >> >>>>>>>>> > >> >>>>>>>> > >> >>>>>>> > >> >>>>> > >> >>> > >> > https://drive.google.com/open?id=1OBGTi3Xb-Kpcf_nHeKA30XiRPEKV4PU7FpieXQhWn7Y > >> >>>>>>>>>> > >> >>>>>>>>>> > >> >>>>>>>>> > >> >>>>>>>>> > >> >>>>>>>> > >> >>>>>>> > >> >>>>>> > >> >>>>> > >> >>>>> > >> >>> > >> >
