Merge pull request #1182 from timeplus-io/backport/storageview-perf-and-clone

StorageView isStreamingQuery cache, view-chain race fix, and join changelog fix for nested aggregation

Author: yokofly

src/Interpreters/Context.cpp

@@ -4512,10 +4512,10 @@ cluster::TimerService & Context::getTimerService() const
     return *shared->global_system_timer;
 }
 
-Context::DataStreamSemanticCache & Context::getDataStreamSemanticCache() const
+Context::QueryAnalysisCache & Context::getQueryAnalysisCache() const
 {
-    assert(hasQueryContext());
-    return getQueryContext()->data_stream_semantic_cache;
+    chassert(hasQueryContext());
+    return getQueryContext()->query_analysis_cache;
 }
 
 /// proton: ends.

src/Interpreters/Context.h

@@ -515,8 +515,17 @@ class ContextData
     /// proton: starts. Parameters for time predicates of main table
     TimeParam time_param;
 
-    using DataStreamSemanticCache = std::unordered_map<std::string, Streaming::DataStreamSemanticEx>;
-    mutable DataStreamSemanticCache data_stream_semantic_cache;
+    /// Query-scoped cache for repeated analysis results (data-stream semantic, is-streaming, …).
+    /// Avoids O(N^2) analyzer recursion on N-deep view chains (issue #11203) and duplicate
+    /// InterpreterSelectWithUnionQuery builds during getSampleBlock. Per-query lifetime, no
+    /// invalidation needed. Keyed by query-string or storage-identity depending on caller.
+    struct QueryAnalysisCacheEntry
+    {
+        std::optional<bool> is_streaming;
+        std::optional<Streaming::DataStreamSemanticEx> data_stream_semantic;
+    };
+    using QueryAnalysisCache = std::unordered_map<std::string, QueryAnalysisCacheEntry>;
+    mutable QueryAnalysisCache query_analysis_cache;
     /// proton: end.
 
     /// Use copy constructor or createGlobal() instead
@@ -799,8 +808,8 @@ class Context: public ContextData, public std::enable_shared_from_this<Context>
 
     const std::set<RequiredColumnTuple> & requiredColumns() const { return required_columns; }
     void addRequiredColumns(RequiredColumnTuple && column_tuple) { required_columns.insert(std::move(column_tuple)); }
-    /// Get data stream semantic for subquery
-    DataStreamSemanticCache & getDataStreamSemanticCache() const;
+    /// Query-scoped analysis cache (data-stream semantic, is-streaming, …). Issue #11203.
+    QueryAnalysisCache & getQueryAnalysisCache() const;
     StoragePtr getTableFunctionResults(const String & key) const;
     void setTableFunctionResults(const String & key, const StoragePtr & table);
     /// proton: ends
@@ -1134,6 +1143,10 @@ class Context: public ContextData, public std::enable_shared_from_this<Context>
 
     bool isQueryFromMaterializedView() const { return is_query_from_materialized_view; }
     void setQueryFromMaterializedView(bool is_query_from_materialized_view_) { is_query_from_materialized_view = is_query_from_materialized_view_; }
+
+    /// True when `query_mode == "table"` — the query is a bounded/historical read.
+    /// `ProxyStream` pushes this setting when resolving `table(...)` table functions.
+    bool isHistoricalQueryMode() const { return getSettingsRef().query_mode.value == "table"; }
     /// proton: ends.
 
     ActionLocksManagerPtr getActionLocksManager();

src/Interpreters/InterpreterSelectWithUnionQuery.cpp

@@ -274,12 +274,12 @@ Block InterpreterSelectWithUnionQuery::getSampleBlock(const ASTPtr & query_ptr_,
     /// Using query string because query_ptr changes for every internal SELECT
     auto key = queryToString(query_ptr_);
 
-    auto & data_stream_semantic_cache = context_->getDataStreamSemanticCache();
+    auto & analysis_cache = context_->getQueryAnalysisCache();
     if (output_data_stream_semantic)
     {
-        auto semantic_iter = data_stream_semantic_cache.find(key);
-        if (semantic_iter != data_stream_semantic_cache.end())
-            *output_data_stream_semantic = semantic_iter->second;
+        auto it = analysis_cache.find(key);
+        if (it != analysis_cache.end() && it->second.data_stream_semantic.has_value())
+            *output_data_stream_semantic = *it->second.data_stream_semantic;
     }
 
     auto & cache = context_->getSampleBlockCache();
@@ -301,7 +301,7 @@ Block InterpreterSelectWithUnionQuery::getSampleBlock(const ASTPtr & query_ptr_,
     InterpreterSelectWithUnionQuery interpreter(query_ptr_, context_, std::move(select_options));
 
     auto data_stream_semantic = interpreter.getDataStreamSemantic();
-    data_stream_semantic_cache[key] = data_stream_semantic;
+    analysis_cache[key].data_stream_semantic = data_stream_semantic;
     if (output_data_stream_semantic)
         *output_data_stream_semantic = data_stream_semantic;
 

src/Interpreters/Streaming/CalculateDataStreamSemantic.cpp

@@ -9,20 +9,27 @@ namespace Streaming
 {
 namespace
 {
+enum class JoinChangelogMode
+{
+    None,
+    Try,
+    Force,
+};
+
 DataStreamSemanticEx calculateDataStreamSemanticForJoin(
     DataStreamSemanticEx left_input_data_stream_semantic,
     DataStreamSemanticEx right_input_data_stream_semantic,
     std::pair<JoinKind, JoinStrictness> kind_and_strictness,
     SelectQueryInfo & query_info,
-    bool allow_emit_changelog_join_result)
+    JoinChangelogMode join_changelog_mode)
 {
     assert(left_input_data_stream_semantic.streaming);
 
-    /// Speical handling
+    /// Special handling
     /// 1) for <stream> join <table>, the right inputs don't support changelog semantic
     if (!right_input_data_stream_semantic.streaming)
     {
-        /// Left stream semantic (Keep the original semantic)
+        /// Left stream semantic
         query_info.left_input_tracking_changes = isChangelogDataStream(left_input_data_stream_semantic);
 
         /// Right stream semantic (Append or VersionedKV or ChangelogKV)
@@ -36,14 +43,16 @@ DataStreamSemanticEx calculateDataStreamSemanticForJoin(
                 throw Exception(ErrorCodes::NOT_IMPLEMENTED, "The filled join data doesn't support changelog processing");
         }
 
-        if (query_info.force_emit_changelog && !isChangelogDataStream(left_input_data_stream_semantic))
+        /// Try/Force emitting changelog
+        if (join_changelog_mode != JoinChangelogMode::None && !isChangelogDataStream(left_input_data_stream_semantic))
         {
             if (canTrackChangesFromInput(left_input_data_stream_semantic))
             {
                 query_info.left_input_tracking_changes = true;
                 return DataStreamSemantic::Changelog;
             }
-            else
+
+            if (join_changelog_mode == JoinChangelogMode::Force)
                 throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Not implemented for emit changelog from append stream join table result");
         }
 
@@ -72,7 +81,20 @@ DataStreamSemanticEx calculateDataStreamSemanticForJoin(
                 kind_and_strictness.first,
                 kind_and_strictness.second);
 
-        return DataStreamSemantic::Append;
+        /// Try/Force emitting changelog
+        if (join_changelog_mode != JoinChangelogMode::None && !isChangelogDataStream(left_input_data_stream_semantic))
+        {
+            if (canTrackChangesFromInput(left_input_data_stream_semantic))
+            {
+                query_info.left_input_tracking_changes = true;
+                return DataStreamSemantic::Changelog;
+            }
+
+            if (join_changelog_mode == JoinChangelogMode::Force)
+                throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Not implemented for emit changelog from asof/any/cross join results");
+        }
+
+        return left_input_data_stream_semantic;
     }
 
     /// Left stream semantic
@@ -99,19 +121,21 @@ DataStreamSemanticEx calculateDataStreamSemanticForJoin(
         right_input_data_stream_semantic = DataStreamSemantic::Append;
     }
 
-    if (isJoinResultChangelog(left_input_data_stream_semantic, right_input_data_stream_semantic))
+    bool join_result_is_changelog = isJoinResultChangelog(left_input_data_stream_semantic, right_input_data_stream_semantic);
+
+    /// Try/Force emitting changelog
+    if (join_changelog_mode != JoinChangelogMode::None)
     {
-        if (allow_emit_changelog_join_result)
+        if (join_result_is_changelog)
             return DataStreamSemantic::Changelog;
-        else
-            /// NOTE: If the current layer doesn't need joined result emit changelog,
-            /// we shall emit append-only stream with mutable semantic(MutableStream).
-            /// In this way, the outer layer can still track changes if needed.
-            /// For example: select count() from (select * from kv join kv2 on kv.key = kv2.key)
-            return DataStreamSemanticEx{StorageSemantic::NativeKV};
+
+        if (join_changelog_mode == JoinChangelogMode::Force)
+            throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Not implemented for emit changelog from non-changelog join results");
     }
-    else
-        return DataStreamSemantic::Append;
+
+    /// When not emitting changelog, preserve NativeKV semantic so the outer layer can still track changes if needed.
+    /// e.g. select count() from (select * from kv join kv2 on kv.key = kv2.key)
+    return join_result_is_changelog ? DataStreamSemanticEx{StorageSemantic::NativeKV} : DataStreamSemantic::Append;
 }
 }
 
@@ -146,7 +170,12 @@ DataStreamSemanticPair calculateDataStreamSemantic(
     DataStreamSemanticPair semantic_pair;
     /// By default, the joined result stream semantic always is append-only unless the current layer has aggregates
     /// or the query forces to emit changelog
-    bool allow_emit_changelog_join_result = current_select_has_aggregates || query_info.force_emit_changelog;
+    JoinChangelogMode join_changelog_mode = JoinChangelogMode::None;
+    if (current_select_has_aggregates)
+        join_changelog_mode = JoinChangelogMode::Try;
+    else if (query_info.force_emit_changelog)
+        join_changelog_mode = JoinChangelogMode::Force;
+
     /// First, look at what the current layer does
 
     /// When the current layer has join or aggregates, we calculate the output data semantic locally and its inputs data stream semantic.
@@ -161,7 +190,7 @@ DataStreamSemanticPair calculateDataStreamSemantic(
                 *right_input_data_stream_semantic,
                 *kind_and_strictness,
                 query_info,
-                allow_emit_changelog_join_result);
+                join_changelog_mode);
         }
         else
         {
@@ -189,10 +218,7 @@ DataStreamSemanticPair calculateDataStreamSemantic(
             *right_input_data_stream_semantic,
             *kind_and_strictness,
             query_info,
-            allow_emit_changelog_join_result);
-
-        if (query_info.force_emit_changelog && !Streaming::isChangelogDataStream(semantic_pair.effective_input_data_stream_semantic))
-            throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Not implemented for emit changelog from non-changelog join results");
+            join_changelog_mode);
 
         semantic_pair.output_data_stream_semantic = semantic_pair.effective_input_data_stream_semantic;
     }

src/Storages/StorageView.cpp

@@ -322,19 +322,74 @@ bool StorageView::isReady() const
 
 bool StorageView::isStreamingQuery(ContextPtr query_context) const
 {
-    auto select = getInMemoryMetadataPtr()->getSelectQuery().inner_query;
+    /// proton: starts. Issue #11203 — query-scoped memoization so N-deep view chains don't
+    /// force quadratic analyzer recursion during a single top-level CREATE VIEW / getSampleBlock.
+
+    /// Historical `table(view)` path: InterpreterSelectQuery::isStreamingQuery() force-returns
+    /// false when query_mode == "table" regardless of storage. Short-circuit here so we skip
+    /// both the interpreter rebuild and the cache probe on that path.
+    if (query_context->isHistoricalQueryMode())
+        return false;
+
+    const auto metadata_ptr = getInMemoryMetadataPtr();
+
+    /// Parameterized views: same storage object can analyze to different queries across
+    /// invocations, so do not cache and keep the original behavior.
+    if (is_parameterized_view)
+    {
+        auto local_ctx = Context::createCopy(query_context);
+        local_ctx->setCollectRequiredColumns(false);
+        auto select = metadata_ptr->getSelectQuery().inner_query->clone();
+        return InterpreterSelectWithUnionQuery(
+                   select, local_ctx, SelectQueryOptions().createParameterizedView().noModify().analyze())
+            .isStreamingQuery();
+    }
+
+    const bool have_query_ctx = query_context->hasQueryContext();
+    std::string cache_key;
+    if (have_query_ctx)
+    {
+        const auto storage_id = getStorageID();
+        /// Key = view identity + metadata revision proxy + effective query mode.
+        /// StorageID carries database/table as a fallback when uuid is nil; the metadata
+        /// pointer acts as a cheap revision token (within one query it is stable, and across
+        /// metadata updates a fresh pointer forces a recompute even if we were long-lived —
+        /// though this cache's lifetime is per-query, so that is belt-and-suspenders).
+        cache_key = fmt::format(
+            "{}.{}|{}|{}|{}",
+            storage_id.database_name,
+            storage_id.table_name,
+            toString(storage_id.uuid),
+            fmt::ptr(metadata_ptr.get()),
+            query_context->getSettingsRef().query_mode.value);
+
+        auto & cache = query_context->getQueryAnalysisCache();
+        if (auto it = cache.find(cache_key); it != cache.end() && it->second.is_streaming.has_value())
+            return *it->second.is_streaming;
+    }
+
     auto local_ctx = Context::createCopy(query_context);
     local_ctx->setCollectRequiredColumns(false);
 
-    if (is_parameterized_view)
-        return InterpreterSelectWithUnionQuery(select, local_ctx, SelectQueryOptions().createParameterizedView().noModify().analyze()).isStreamingQuery();
-    else
-        return InterpreterSelectWithUnionQuery(select, local_ctx, SelectQueryOptions().noModify().analyze()).isStreamingQuery();
+    /// Analyzer mutates query_ptr in place, so concurrent analyses on the view's shared
+    /// inner_query race and tear the AST; clone before handing it off.
+    auto select = metadata_ptr->getSelectQuery().inner_query->clone();
+    const bool result = InterpreterSelectWithUnionQuery(
+                            select, local_ctx, SelectQueryOptions().noModify().analyze())
+                            .isStreamingQuery();
+
+    if (have_query_ctx)
+    {
+        auto & cache = query_context->getQueryAnalysisCache();
+        cache[std::move(cache_key)].is_streaming = result;
+    }
+    return result;
+    /// proton: ends.
 }
 
 bool StorageView::hasStreamingGlobalAggregation() const
 {
-    auto select = getInMemoryMetadataPtr()->getSelectQuery().inner_query;
+    auto select = getInMemoryMetadataPtr()->getSelectQuery().inner_query->clone();
     auto ctx = Context::createCopy(local_context);
     ctx->setCollectRequiredColumns(false);
 
@@ -346,20 +401,23 @@ bool StorageView::hasStreamingGlobalAggregation() const
 
 Streaming::DataStreamSemanticEx StorageView::dataStreamSemantic() const
 {
+    std::lock_guard lock(data_stream_semantic_mutex);
     if (data_stream_semantic_resolved)
         return data_stream_semantic;
 
-    auto select = getInMemoryMetadataPtr()->getSelectQuery().inner_query;
+    auto select = getInMemoryMetadataPtr()->getSelectQuery().inner_query->clone();
     auto ctx = Context::createCopy(local_context);
     ctx->setCollectRequiredColumns(false);
 
     if (is_parameterized_view)
-        data_stream_semantic = InterpreterSelectWithUnionQuery(select, ctx, SelectQueryOptions().createParameterizedView().noModify().analyze()).getDataStreamSemantic();
+        data_stream_semantic
+            = InterpreterSelectWithUnionQuery(select, ctx, SelectQueryOptions().createParameterizedView().noModify().analyze())
+                  .getDataStreamSemantic();
     else
-        data_stream_semantic = InterpreterSelectWithUnionQuery(select, ctx, SelectQueryOptions().noModify().analyze()).getDataStreamSemantic();
+        data_stream_semantic
+            = InterpreterSelectWithUnionQuery(select, ctx, SelectQueryOptions().noModify().analyze()).getDataStreamSemantic();
 
     data_stream_semantic_resolved = true;
-
     return data_stream_semantic;
 }
 

src/Storages/StorageView.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <mutex>
+
 #include <Parsers/ASTSelectQuery.h>
 #include <Parsers/IAST_fwd.h>
 #include <Storages/IStorage.h>
@@ -60,6 +62,7 @@ class StorageView final : public shared_ptr_helper<StorageView>, public IStorage
 
 private:
     ContextMutablePtr local_context;
+    mutable std::mutex data_stream_semantic_mutex;
     mutable bool data_stream_semantic_resolved = false;
     std::atomic_flag started;
     /// proton: ends.

tests/queries_ported/0_stateless/99053_global_aggr_over_global_aggr_bugs.reference

@@ -22,3 +22,6 @@
 2
 3
 4
+1	1
+1	-1
+2	1