Merge 'cherry-pick-mr-25202-1' into 'cnch-dev'

fix(clickhousech@m-5341556137): [cp] Store lockHolders in pipeline when there is no res.in to avoid core in HTTP INSERT OVERWRITE VALUES

See merge request: !25207

src/Interpreters/InterpreterInsertQuery.cpp

@@ -675,7 +675,15 @@ BlockIO InterpreterInsertQuery::execute()
     if (insert_query.is_overwrite && !lock_holders.empty())
     {
         /// Make sure lock is release after txn commit
-        res.in = std::make_shared<LockHoldBlockInputStream>(res.in, std::move(lock_holders));
+        if (res.in)
+        {
+            /// Only use LockHoldBlockInputStream if there is a input stream to avoid misjudgement.
+            res.in = std::make_shared<LockHoldBlockInputStream>(res.in, std::move(lock_holders));
+        }
+        else
+        {
+            res.pipeline.addLockHolders(std::move(lock_holders));
+        }
     }
     return res;
 }

src/Processors/Pipe.h

@@ -21,6 +21,7 @@
 
 #pragma once
 
+#include <iterator>
 #include <Processors/IProcessor.h>
 #include <QueryPlan/QueryIdHolder.h>
 #include <QueryPlan/QueryPlan.h>
@@ -28,6 +29,7 @@
 #include <DataStreams/SizeLimits.h>
 #include <Storages/TableLockHolder.h>
 #include <Interpreters/Cache/QueryCache.h> /// nested classes such as QC::Writer can't be fwd declared
+#include <Transaction/CnchLock.h>
 
 namespace DB
 {
@@ -153,6 +155,10 @@ public:
     /// For queries with nested interpreters (i.e. StorageDistributed)
     void addQueryPlan(std::unique_ptr<QueryPlan> plan) { holder.query_plans.emplace_back(std::move(plan)); }
 
+    void addLockHolders(CnchLockHolderPtrs && lock_holders_) { holder.lock_holders.insert(holder.lock_holders.end(),
+                                                                                          std::make_move_iterator(lock_holders_.begin()),
+                                                                                          std::make_move_iterator(lock_holders_.end())); }
+
 private:
     /// Destruction order: processors, header, locks, temporary storages, local contexts
 
@@ -172,6 +178,7 @@ private:
         std::vector<std::unique_ptr<QueryPlan>> query_plans;
         std::shared_ptr<QueryIdHolder> query_id_holder;
         CacheHolderPtr cache_holder;
+        CnchLockHolderPtrs lock_holders;
     };
 
     Holder holder;

src/Processors/QueryPipeline.h

@@ -156,6 +156,7 @@ public:
     void addStorageHolder(StoragePtr storage) { pipe.addStorageHolder(std::move(storage)); }
     void addCacheHolder(CacheHolderPtr cache_holder) { pipe.addCacheHolder(std::move(cache_holder)); }
     void addQueryPlan(std::unique_ptr<QueryPlan> plan) { pipe.addQueryPlan(std::move(plan)); }
+    void addLockHolders(CnchLockHolderPtrs && lock_holders) { pipe.addLockHolders(std::move(lock_holders)); }
     void setLimits(const StreamLocalLimits & limits) { pipe.setLimits(limits); }
     void setLeafLimits(const SizeLimits & limits) { pipe.setLeafLimits(limits); }
     void setQuota(const std::shared_ptr<const EnabledQuota> & quota_)

tests/queries/4_cnch_stateless/03032_insert_overwrite_http.reference

@@ -0,0 +1,14 @@
+0
+---- INSERT VALUES enable_optimizer = 0
+1
+---- INSERT VALUES enable_optimizer = 1
+2
+---- INSERT SELECT enable_optimizer = 0
+0
+1
+2
+---- INSERT SELECT enable_optimizer = 1
+0
+1
+2
+3

tests/queries/4_cnch_stateless/03032_insert_overwrite_http.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "DROP TABLE IF EXISTS ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "CREATE TABLE ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http (id UInt64) ENGINE = CnchMergeTree ORDER BY id"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "INSERT INTO ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http VALUES (0)"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "SELECT * FROM ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http ORDER BY id"
+
+echo "---- INSERT VALUES enable_optimizer = 0"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}&enable_optimizer=0" -d "INSERT OVERWRITE ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http VALUES (1)"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "SELECT * FROM ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http ORDER BY id"
+
+echo "---- INSERT VALUES enable_optimizer = 1"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}&enable_optimizer=1" -d "INSERT OVERWRITE ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http VALUES (2)"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "SELECT * FROM ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http ORDER BY id"
+
+echo "---- INSERT SELECT enable_optimizer = 0"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}&enable_optimizer=0" -d "INSERT OVERWRITE ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http SELECT number as id FROM numbers(3)"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "SELECT * FROM ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http ORDER BY id"
+
+echo "---- INSERT SELECT enable_optimizer = 1"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}&enable_optimizer=1" -d "INSERT OVERWRITE ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http SELECT number as id FROM numbers(4)"
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "SELECT * FROM ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http ORDER BY id"
+
+${CLICKHOUSE_CURL} -s "${CLICKHOUSE_URL}" -d "DROP TABLE IF EXISTS ${CLICKHOUSE_DATABASE}.test_insert_overwrite_http"