support more emit modes in window (#517)

Author: yl-lisen

src/Core/Streaming/Watermark.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <base/types.h>
+
 namespace DB
 {
 namespace Streaming
@@ -9,5 +11,25 @@ namespace Streaming
 /// The 1970-01-01 00:00:00 is Int64 value 0, earlier times are negative, so we need to use signed Int64 as watermark
 constexpr Int64 INVALID_WATERMARK = std::numeric_limits<Int64>::min();
 constexpr Int64 TIMEOUT_WATERMARK = std::numeric_limits<Int64>::max();
+
+/// TODO: Separate EmitMode into two parts:
+/// 1) EmitStrategy         - how to do finalization (what to emit) for all aggregates processing
+/// 2) WatermarkStragegy    - how / when to stamp watermark
+enum class EmitMode : uint8_t
+{
+    None = 0,
+    Tail,
+
+    Periodic,  /// Emit a processing time watermark at periodic interval
+
+    Watermark, /// Allow time skew in same window, emit the watermark when a window closed
+    PeriodicWatermark, /// Same as WATERMARK, but emit the watermark at periodic interval
+
+    /* emit only keyed and changed states for aggregating */
+    OnUpdate = 1u << 7, /// Emit a processing time watermark per batch of events
+    PeriodicOnUpdate, /// Same as Periodic
+    WatermarkOnUpdate, /// Same as Watermark, but emit the watermark per batch of events
+    PeriodicWatermarkOnUpdate, /// Same as PeriodicWatermark
+};
 }
 }

src/Interpreters/ExpressionAnalyzer.cpp

@@ -1997,7 +1997,7 @@ ExpressionAnalysisResult::ExpressionAnalysisResult(
                 if (const auto * proxy = storage->as<Streaming::ProxyStream>())
                 {
                     if (auto window_desc = proxy->getStreamingWindowFunctionDescription();
-                        window_desc && window_desc->type == Streaming::WindowType::SESSION)
+                        window_desc && window_desc->type == Streaming::WindowType::Session)
                     {
                         auto & step = chain.getLastStep();
                         step.addRequiredOutput(window_desc->argument_names[0]);

src/Interpreters/InterpreterSelectQuery.cpp

@@ -111,6 +111,7 @@
 #include <Processors/QueryPlan/Streaming/WatermarkStep.h>
 #include <Processors/QueryPlan/Streaming/WatermarkStepWithSubstream.h>
 #include <Processors/QueryPlan/Streaming/WindowStep.h>
+#include <Processors/Transforms/Streaming/AggregatingHelper.h>
 #include <Processors/Transforms/Streaming/WatermarkStamper.h>
 #include <Storages/Streaming/ProxyStream.h>
 #include <Storages/Streaming/StorageStream.h>
@@ -3219,7 +3220,7 @@ void InterpreterSelectQuery::executeStreamingAggregation(
                 ErrorCodes::NOT_IMPLEMENTED,
                 "User defined aggregation function with emit strategy shouldn't be used together with other aggregation function");
 
-        if (windowType() != Streaming::WindowType::NONE)
+        if (windowType() != Streaming::WindowType::None)
             throw Exception(
                 ErrorCodes::NOT_IMPLEMENTED,
                 "User defined aggregation function with emit strategy shouldn't be used together with streaming window");
@@ -3240,6 +3241,9 @@ void InterpreterSelectQuery::executeStreamingAggregation(
     auto tracking_updates_type = Streaming::TrackingUpdatesType::None;
     if (data_stream_semantic_pair.isChangelogOutput())
         tracking_updates_type = Streaming::TrackingUpdatesType::UpdatesWithRetract;
+    /// TODO: A optimization for `emit on update`, we don't need to track updates and just directly convert each input (fast in and fast out)
+    else if (Streaming::AggregatingHelper::onlyEmitUpdates(emit_mode))
+        tracking_updates_type = Streaming::TrackingUpdatesType::Updates;
 
     Streaming::Aggregator::Params params(
         header_before_aggregation,
@@ -3278,10 +3282,10 @@ void InterpreterSelectQuery::executeStreamingAggregation(
     /// 2) `shuffle by`: calculating light substream without substream ID (The data have been shuffled by `LightShufflingTransform`)
     if (query_info.hasPartitionByKeys() || light_shuffled)
         query_plan.addStep(std::make_unique<Streaming::AggregatingStepWithSubstream>(
-            query_plan.getCurrentDataStream(), std::move(params), final, emit_version, data_stream_semantic_pair.isChangelogOutput()));
+            query_plan.getCurrentDataStream(), std::move(params), final, emit_version, data_stream_semantic_pair.isChangelogOutput(), emit_mode));
     else
         query_plan.addStep(std::make_unique<Streaming::AggregatingStep>(
-            query_plan.getCurrentDataStream(), std::move(params), final, merge_threads, temporary_data_merge_threads, emit_version, data_stream_semantic_pair.isChangelogOutput()));
+            query_plan.getCurrentDataStream(), std::move(params), final, merge_threads, temporary_data_merge_threads, emit_version, data_stream_semantic_pair.isChangelogOutput(), emit_mode));
 }
 
 /// Resolve input / output data stream semantic.
@@ -3370,7 +3374,7 @@ Streaming::WindowType InterpreterSelectQuery::windowType() const
         }
     }
 
-    return Streaming::WindowType::NONE;
+    return Streaming::WindowType::None;
 }
 
 bool InterpreterSelectQuery::hasStreamingGlobalAggregation() const
@@ -3498,12 +3502,14 @@ void InterpreterSelectQuery::buildShufflingQueryPlan(QueryPlan & query_plan)
         query_plan.getCurrentDataStream(), std::move(substream_key_positions), shuffle_output_streams));
 }
 
-void InterpreterSelectQuery::buildWatermarkQueryPlan(QueryPlan & query_plan) const
+void InterpreterSelectQuery::buildWatermarkQueryPlan(QueryPlan & query_plan)
 {
     assert(isStreamingQuery());
     auto params = std::make_shared<Streaming::WatermarkStamperParams>(
         query_info.query, query_info.syntax_analyzer_result, query_info.streaming_window_params);
 
+    emit_mode = params->mode; /// saved it to be used for streaming aggregating step
+
     bool skip_stamping_for_backfill_data = !context->getSettingsRef().emit_during_backfill.value;
 
     if (query_info.hasPartitionByKeys())

src/Interpreters/InterpreterSelectQuery.h

@@ -198,7 +198,7 @@ class InterpreterSelectQuery : public IInterpreterUnionOrSelectQuery
     void finalCheckAndOptimizeForStreamingQuery();
     bool shouldKeepAggregateState() const;
     void buildShufflingQueryPlan(QueryPlan & query_plan);
-    void buildWatermarkQueryPlan(QueryPlan & query_plan) const;
+    void buildWatermarkQueryPlan(QueryPlan & query_plan);
     void buildStreamingProcessingQueryPlanBeforeJoin(QueryPlan & query_plan);
     void buildStreamingProcessingQueryPlanAfterJoin(QueryPlan & query_plan);
     void checkEmitVersion();
@@ -257,6 +257,9 @@ class InterpreterSelectQuery : public IInterpreterUnionOrSelectQuery
     mutable std::optional<bool> is_streaming_query;
     bool shuffled_before_join = false;
     bool light_shuffled = false;
+
+    Streaming::EmitMode emit_mode = Streaming::EmitMode::None;
+
     /// Overall data stream semantic defines the output semantic of the current layer of SELECT
     Streaming::DataStreamSemanticPair data_stream_semantic_pair;
     /// proton: ends