support more emit modes in window #517

yl-lisen · 2024-01-23T07:03:21Z

PR checklist:

Did you run ClangFormat ?
Did you separate headers to a different section in existing community code base ?
Did you surround proton: starts/ends for new code in existing community code base ?

Please write user-readable short description of the changes:

Support emit intermediate windows (if specified emit periodic or emit on update)
Support only emit update groups (if specified emit on update or emit periodic on update
Tweak emit clause syntax:

EMIT [STREAM|CHANGELOG]
    [AFTER WATERMARK WITH DELAY <interval>]
    [PERIODIC <interval>]
    [ON UPDATE]
    [[ AND ]TIMEOUT <interval>]
    [[ AND ]LAST <last-x> [ON PROCTIME]]

(MODIFED)
[AFTER WATERMARK WITH DELAY <interval>] : Allow time skew in a certain range `[<max_event_ts - delay_interval>, <max_event_ts>]

(NEW) [ON UPDATE] : emit per update and only emit update per group by key
(NEW) [PERIODIC <interval> ON UPDATE] : periodic emit and only emit update per group by key

src/Parsers/Streaming/ParserEmitQuery.cpp

tests/stream/test_stream_smoke/0034_emit_mode.yaml

chenziliang · 2024-02-11T19:23:51Z

src/Core/Streaming/Watermark.h

+    /// TODO: more strategies ...
+};
+
+enum class WatermarkEmitMode


Why we need this since the WatermarkStrategy already says this purpose ?

This shall be just EmitMode instead of WatermarkEmitMode or better EmitStrategy ?

WatermarkStrategy tells us how / when to stamp watermark.
EmitStrategy tells us how to do finalization (what to emit) when watermark has been observed.

So for EmitStrategy, we only have 4 combinations :

emit all states in a tumble / hop window or global window when a watermark is observed.

emit only changed states per group key in a timed window or global window when a watermark is observed. This emit strategy works for session window and substream by default from day one

emit changelog, combined with 1) and 2), we have 2 here. But emit changelog for all statesprobably doesn't make sense at all,emit changelog for updates` make lots of sense though

So the valid emit strategy for now could be

1. emit all states 2. emit only keyed and changed states 3. emit changelog for keyed and changed states

Observing the timestamp, stamp watermark, assign window, emit states are all hand in hand things, we may need think about them carefully.

Previously we basically did everything (except emit states) in the stamper with session window as an exception.

I like the idea to separate the Observing the timestamp, stamp watermark and assigning window and emit states.

If we separate watermark stamping and window assignment, watermark itself could be pretty simple and having these use cases

Ascending - basically proc time processing (EMIT ON PROCTIME) or row number processing (row number as a timestamp, per row emit in future). For every event in a chunk, stamp a timestamp for each of them. This seems a stateless watermark strategy (probably we still have state incase wall clock rewind / adjusted on that machine ?)

BoundedOutOfOrderness - (EMIT ... WITH DELAY), observing the timestamp column for each event, maintain the max timestamp observed so far, if the out of order for an event is within delay, still assign a watermark to it, update the max observed timestamp as we go.

OutOfOrdernessInWindowAndBatch - do we need that ? Since for tumble, hopping window, we know the delay automatically ? it seems a special case of 2 ?

For session window, do we need a watermark ?

Yes, there are more and more actions. Separating these can better maintain them.
In fact, EmitStrategy should only be used within AggregatingTransform.

chenziliang · 2024-02-11T19:44:48Z

src/Interpreters/Streaming/WindowCommon.h

@@ -273,6 +273,9 @@ struct SessionWindowParams : WindowParams
    bool start_with_inclusion;
    bool end_with_inclusion;

+    /// TODO: So far, always assign session window in aggr
+    bool assign_window_pushdown = true;


/// So far, only for session window, we evaluate the watermark and window for the events in Aggregate Transform /// For other windows, we assigned the watermark in Watermark stamper step.

Previously we are kind mixing window assignment and watermark assignment for tumble / hopping / global windows since this is probably more performant.

Right now, we are separating these 2 steps, which are good for maintainability moving forward.

For session window, fundamentally, it doesn't have a fixed watermark because it highly depends on inputs and the evaluation of start / end expression.

So, it seems window assignment for now all happens at Aggregate transform step ? The Watermark evaluation happens in WatermarkStamper step and for session window, we don't have watermark

pushdown_window_assignment

chenziliang · 2024-02-12T05:34:52Z

src/Processors/QueryPlan/Streaming/AggregatingStep.h

@@ -22,7 +24,8 @@ class AggregatingStep : public ITransformingStep
        size_t merge_threads_,
        size_t temporary_data_merge_threads_,
        bool emit_version_,
-        bool emit_changelog_);
+        bool emit_changelog_,


bool emit_changelog_ is redundant moving forward since emit strategy shall cover that

do it in a separate PR

chenziliang · 2024-02-12T05:43:16Z

src/Processors/Transforms/Streaming/WindowAggregatingTransformWithSubstream.h

    virtual void removeBucketsImpl(Int64 watermark, const SubstreamContextPtr & substream_ctx) = 0;
    virtual bool needReassignWindow() const = 0;

    std::optional<size_t> window_start_col_pos;
    std::optional<size_t> window_end_col_pos;
+
+    bool only_emit_finalized_windows = true;


these bools are basically a derive from the emit strategy enums ? It appears making the whole code path complex

yes, there are different finalizing path in AggregatingTransform for different emit strategy

src/Processors/Transforms/Streaming/AggregatingHelper.h

chenziliang · 2024-02-12T05:47:45Z

src/Processors/Transforms/Streaming/AggregatingHelper.cpp

+    return DB::convertToChunk(params.aggregator.spliceAndConvertUpdatesToBlock(data, buckets));
+}
+
+Chunk mergeAndSpliceAndConvertUpdatesToChunk(


We probably don't need Splice in the name

mergeAndConvertUpdatesToChunk is existed, it's used in global aggr.
mergeAndSpliceAndConvertUpdatesToChunk is used in window aggr.

src/Processors/Transforms/Streaming/TumbleHelper.cpp

src/Processors/Transforms/Streaming/WatermarkStamper.cpp

chenziliang · 2024-02-12T05:57:58Z

src/Processors/Transforms/Streaming/WatermarkStamper.cpp

+    switch (params.type)
+    {
+        case WindowType::TUMBLE:
+            return TumbleHelper::validateWatermarkStrategyAndEmitMode(strategy, mode, assert_cast<TumbleWindowParams &>(params));


What's the benefit to use free functions compared with inheritance as we did switch and all type casts etc here

src/Processors/Transforms/Streaming/WatermarkStamper.cpp

chenziliang · 2024-02-12T06:06:33Z

src/Processors/Transforms/Streaming/WatermarkStamper.h

-
-    void processTimeout(Chunk & chunk);
+    /// \brief Emit a watermark according to the EmitMode
+    /// \p old_watermark_ts - saved watermark before current processing


Let's consistently use either @brief, \brief, \param etc, \param has builtin support by CLION

do it in a separate PR

chenziliang · 2024-02-12T06:07:39Z

src/Processors/Transforms/Streaming/WatermarkStamper.h

-    Int64 next_timeout_emit_ts = 0;
-    Int64 timeout_interval = 0;
+    /// For periodic or timeout timer
+    InternalTimer periodic_timer;


If we leverage Stopwatch we probably can do similar thing ?

src/Processors/Transforms/Streaming/WatermarkTransform.cpp

src/Processors/Transforms/Streaming/SessionHelper.cpp

chenziliang · 2024-02-12T06:28:44Z

General comments

We have so many different enums class / types regarding emit strategy, watermark strategy, emit on updates, emit finalized windows, bools etc and I think they can be simplified. They are so complex right now, i worried quite a lot it is way more complex than any other proton team member can maintain.
Comments, we will need write lots of comments to explain what important functions do etc, how they work etc
Free function vs Inheritance class / struct. It appears free function is easier to reuse, but too many free functions and switch case / if else may also cause maintenance issue.
I generally agree we shall refactor watermark, and window assignment into different class etc, but this refactor is expected to happen in a separate PR.
Emit behavior, it appears we still didn't emit per group by key for EMIT ON UPDATE but for that window, this is probably NOT what want in general.

For example, if we do SELECT k, count() FROM test GROUP BY k EMIT ON UPDATE, it only emits for keys with updates, for keys without any updates, don't emit anything.

For another example, if we do SELECT window_start, device, count() FROM tumble(device_metrics, 5s) GROUP BY window_start, device EMIT ON UPDATE, it is expected to emit only on specific device with updates in the current tumble window. For devices without updates in the current window, don't emit.

src/Core/Streaming/Watermark.h

chenziliang · 2024-02-12T06:35:22Z

src/Core/Streaming/Watermark.h

+    /// No watermark
+    None,
+
+    /// Watermark is generated based on each event, this is used for `EMIT AFTER WATERMARK WITHOUT DELAY`


For every event ? What's the use case ? Probably proc time processing ?

For session window, we need events timestamp is in ascending, which is required by session start predicates/end predicates

chenziliang · 2024-02-12T06:38:25Z

src/Core/Streaming/Watermark.h

+
+    /// (Built-in): Watermark is generated based on a batch events:
+    /// - Allow time skew in one batch events
+    OutOfOrdernessInBatch,


For every batch, we generate a watermark ?

unused, remove it

src/Core/Streaming/Watermark.h

chenziliang · 2024-02-12T16:52:36Z

src/Processors/Transforms/Streaming/AggregatingHelper.h

+
+[[maybe_unused]] ALWAYS_INLINE static bool onlyEmitUpdates(EmitMode mode) noexcept
+{
+    return static_cast<uint8_t>(mode) & EMIT_UPDATES_MASK;


Let's revisit this, it appears something wrong here since we are stuffing bit into an enum ?

yl-lisen self-assigned this Jan 23, 2024

yl-lisen force-pushed the feature/issue-440-support-more-emit-modes-in-window branch from 5776db7 to a5e38d0 Compare January 23, 2024 08:17

yl-lisen marked this pull request as ready for review January 23, 2024 09:45

yl-lisen requested a review from chenziliang January 23, 2024 16:06

yl-lisen force-pushed the feature/issue-440-support-more-emit-modes-in-window branch from 6caa777 to 11062b1 Compare January 24, 2024 09:24

yl-lisen force-pushed the feature/issue-440-support-more-emit-modes-in-window branch from ce95d80 to 42cb940 Compare February 5, 2024 22:57

chenziliang reviewed Feb 8, 2024

View reviewed changes

src/Parsers/Streaming/ParserEmitQuery.cpp Outdated Show resolved Hide resolved

chenziliang reviewed Feb 11, 2024

View reviewed changes

tests/stream/test_stream_smoke/0034_emit_mode.yaml Outdated Show resolved Hide resolved

chenziliang reviewed Feb 11, 2024

View reviewed changes

tests/stream/test_stream_smoke/0034_emit_mode.yaml Show resolved Hide resolved

chenziliang reviewed Feb 11, 2024

View reviewed changes

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/AggregatingHelper.h Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/AggregatingHelper.h Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/AggregatingHelper.h Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/TumbleHelper.cpp Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/WatermarkStamper.cpp Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/WatermarkStamper.cpp Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/WatermarkTransform.cpp Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Processors/Transforms/Streaming/SessionHelper.cpp Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Core/Streaming/Watermark.h Outdated Show resolved Hide resolved

chenziliang reviewed Feb 12, 2024

View reviewed changes

src/Core/Streaming/Watermark.h Outdated Show resolved Hide resolved

yl-lisen added 8 commits February 12, 2024 15:25

support more emit modes in window

cff2662

do emit per group by key for EMIT ON UPDATE

da3249d

remove unused WatermarkStrategy 'OutOfOrdernessInBatch'

422c328

rename 'assign_window_pushdown' to 'pushdown_window_assignment'

8e805c3

move out refactor code for emit mode

f508aac

rename xxxHelper to xxxWindowHelper

4e9c91d

rename WindowType::TUMBLE to WindowType::Tumble etc.

c5e77ed

tweak code/comments style

ddd1a73

yl-lisen force-pushed the feature/issue-440-support-more-emit-modes-in-window branch from 42cb940 to ddd1a73 Compare February 12, 2024 16:07

yl-lisen requested a review from chenziliang February 12, 2024 16:09

revert smoke test for hop window with substream and emit with delay

f0804ca

chenziliang reviewed Feb 12, 2024

View reviewed changes

Refine emit mode enum and inline helpers

bb4c3d1

chenziliang merged commit 2cea82f into develop Feb 12, 2024
6 checks passed

yl-lisen mentioned this pull request Mar 11, 2024

In window emit #440

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

support more emit modes in window #517

support more emit modes in window #517

yl-lisen commented Jan 23, 2024 •

edited by chenziliang

Loading

chenziliang Feb 11, 2024

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 12, 2024

chenziliang Feb 12, 2024

chenziliang Feb 12, 2024

yl-lisen Feb 12, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 11, 2024

chenziliang Feb 11, 2024

chenziliang Feb 12, 2024

chenziliang Feb 12, 2024 •

edited

Loading

yl-lisen Feb 12, 2024

chenziliang Feb 12, 2024

yl-lisen Feb 12, 2024

chenziliang Feb 12, 2024

yl-lisen Feb 12, 2024

chenziliang Feb 12, 2024

chenziliang Feb 12, 2024

yl-lisen Feb 12, 2024

chenziliang Feb 12, 2024

chenziliang commented Feb 12, 2024 •

edited

Loading

chenziliang Feb 12, 2024 •

edited

Loading

yl-lisen Feb 12, 2024 •

edited

Loading

chenziliang Feb 12, 2024

yl-lisen Feb 12, 2024

chenziliang Feb 12, 2024 •

edited

Loading

support more emit modes in window #517

support more emit modes in window #517

Conversation

yl-lisen commented Jan 23, 2024 • edited by chenziliang Loading

Choose a reason for hiding this comment

chenziliang Feb 11, 2024 • edited Loading

Choose a reason for hiding this comment

chenziliang Feb 11, 2024 • edited Loading

Choose a reason for hiding this comment

chenziliang Feb 11, 2024 • edited Loading

Choose a reason for hiding this comment

chenziliang Feb 11, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yl-lisen Feb 12, 2024 • edited Loading

Choose a reason for hiding this comment

chenziliang Feb 11, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chenziliang Feb 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chenziliang commented Feb 12, 2024 • edited Loading

chenziliang Feb 12, 2024 • edited Loading

Choose a reason for hiding this comment

yl-lisen Feb 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chenziliang Feb 12, 2024 • edited Loading

Choose a reason for hiding this comment

yl-lisen commented Jan 23, 2024 •

edited by chenziliang

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

yl-lisen Feb 12, 2024 •

edited

Loading

chenziliang Feb 11, 2024 •

edited

Loading

chenziliang Feb 12, 2024 •

edited

Loading

chenziliang commented Feb 12, 2024 •

edited

Loading

chenziliang Feb 12, 2024 •

edited

Loading

yl-lisen Feb 12, 2024 •

edited

Loading

chenziliang Feb 12, 2024 •

edited

Loading