fix issue pingcap#7810 (pingcap#7854) (pingcap#7871)

close pingcap#7810
ti-chi-bot · Aug 1, 2023 · 4df1054 · 4df1054
1 parent bae3aef
commit 4df1054
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 32 deletions.
diff --git a/dbms/src/Flash/executeQuery.cpp b/dbms/src/Flash/executeQuery.cpp
@@ -92,7 +92,6 @@ BlockIO executeDAG(IQuerySource & dag, Context & context, bool internal)
     /// Hold element of process list till end of query execution.
     res.process_list_entry = process_list_entry;
 
-    prepareForInputStream(context, QueryProcessingStage::Complete, res.in);
     if (likely(!internal))
         logQueryPipeline(logger, res.in);
 

diff --git a/dbms/src/Interpreters/executeQuery.cpp b/dbms/src/Interpreters/executeQuery.cpp
@@ -146,6 +146,32 @@ void onExceptionBeforeStart(const String & query, Context & context, time_t curr
     }
 }
 
+void prepareForInputStream(
+    Context & context,
+    QueryProcessingStage::Enum stage,
+    const BlockInputStreamPtr & in)
+{
+    assert(in);
+    if (auto * stream = dynamic_cast<IProfilingBlockInputStream *>(in.get()))
+    {
+        stream->setProgressCallback(context.getProgressCallback());
+        stream->setProcessListElement(context.getProcessListElement());
+
+        /// Limits on the result, the quota on the result, and also callback for progress.
+        /// Limits apply only to the final result.
+        if (stage == QueryProcessingStage::Complete)
+        {
+            IProfilingBlockInputStream::LocalLimits limits;
+            limits.mode = IProfilingBlockInputStream::LIMITS_CURRENT;
+            const auto & settings = context.getSettingsRef();
+            limits.size_limits = SizeLimits(settings.max_result_rows, settings.max_result_bytes, settings.result_overflow_mode);
+
+            stream->setLimits(limits);
+            stream->setQuota(context.getQuota());
+        }
+    }
+}
+
 std::tuple<ASTPtr, BlockIO> executeQueryImpl(
     IQuerySource & query_src,
     Context & context,
@@ -386,32 +412,6 @@ void logQuery(const String & query, const Context & context, const LoggerPtr & l
         joinLines(query));
 }
 
-void prepareForInputStream(
-    Context & context,
-    QueryProcessingStage::Enum stage,
-    const BlockInputStreamPtr & in)
-{
-    assert(in);
-    if (auto * stream = dynamic_cast<IProfilingBlockInputStream *>(in.get()))
-    {
-        stream->setProgressCallback(context.getProgressCallback());
-        stream->setProcessListElement(context.getProcessListElement());
-
-        /// Limits on the result, the quota on the result, and also callback for progress.
-        /// Limits apply only to the final result.
-        if (stage == QueryProcessingStage::Complete)
-        {
-            IProfilingBlockInputStream::LocalLimits limits;
-            limits.mode = IProfilingBlockInputStream::LIMITS_CURRENT;
-            const auto & settings = context.getSettingsRef();
-            limits.size_limits = SizeLimits(settings.max_result_rows, settings.max_result_bytes, settings.result_overflow_mode);
-
-            stream->setLimits(limits);
-            stream->setQuota(context.getQuota());
-        }
-    }
-}
-
 std::shared_ptr<ProcessListEntry> setProcessListElement(
     Context & context,
     const String & query,

diff --git a/dbms/src/Interpreters/executeQuery.h b/dbms/src/Interpreters/executeQuery.h
@@ -61,9 +61,4 @@ void logQueryPipeline(const LoggerPtr & logger, const BlockInputStreamPtr & in);
 
 void logQuery(const String & query, const Context & context, const LoggerPtr & logger);
 
-void prepareForInputStream(
-    Context & context,
-    QueryProcessingStage::Enum stage,
-    const BlockInputStreamPtr & in);
-
 } // namespace DB