From d1cb567e6af48837d8413cd8724eb0c3cd936d28 Mon Sep 17 00:00:00 2001 From: lgqss Date: Wed, 10 Jun 2026 14:15:46 +0800 Subject: [PATCH] fix vec union join --- include/column/sort_merge.h | 4 ++-- include/common/schema_factory.h | 2 +- include/meta_server/meta_util.h | 4 ++-- src/exec/dml_node.cpp | 5 ++++- src/exec/dual_scan_node.cpp | 12 +++++++++++- src/exec/fetcher_store.cpp | 9 ++++++--- src/exec/join_node.cpp | 6 ++++++ src/exec/rocksdb_scan_node.cpp | 10 ++++++++-- src/runtime/arrow_io_excutor.cpp | 2 +- src/vector_index/vector_index.cpp | 4 ++-- test/test_dms.cpp | 8 ++++++-- test/test_fetcher_store.cpp | 17 +++++++++++++++++ 12 files changed, 66 insertions(+), 17 deletions(-) diff --git a/include/column/sort_merge.h b/include/column/sort_merge.h index 011e2b4a..86dd4a50 100644 --- a/include/column/sort_merge.h +++ b/include/column/sort_merge.h @@ -101,7 +101,7 @@ struct UnorderRow { } } - return true; + return false; // equal is not less-than } }; @@ -306,7 +306,7 @@ struct SingleRow : public SlotIndex { } } - return true; + return false; // equal is not less-than } private: CacheRow* _row = nullptr; diff --git a/include/common/schema_factory.h b/include/common/schema_factory.h index 650ee0e7..a290a17e 100644 --- a/include/common/schema_factory.h +++ b/include/common/schema_factory.h @@ -517,7 +517,7 @@ class RangePartition : public Partition { bool operator() (const Range& range1, const Range& range2) { int64_t res = range1.right_value.compare(range2.right_value); if (res == 0) { - return range1.partition_type <= range2.partition_type; + return range1.partition_type < range2.partition_type; } else { return res < 0; } diff --git a/include/meta_server/meta_util.h b/include/meta_server/meta_util.h index aed0de86..d50818de 100644 --- a/include/meta_server/meta_util.h +++ b/include/meta_server/meta_util.h @@ -142,7 +142,7 @@ struct RangeComparator { } const pb::Expr& pre_right_value = left.range().right_value(); const pb::Expr& next_right_value = right.range().right_value(); - return compare(pre_right_value, next_right_value) <= 0; + return compare(pre_right_value, next_right_value) < 0; } }; @@ -156,7 +156,7 @@ struct PointerRangeComparator { const pb::Expr& next_right_value = right->range().right_value(); int64_t res = compare(pre_right_value, next_right_value); if (res == 0) { - return left->type() <= right->type(); + return left->type() < right->type(); } else { return res < 0; } diff --git a/src/exec/dml_node.cpp b/src/exec/dml_node.cpp index 3ee9e49c..103d79bb 100644 --- a/src/exec/dml_node.cpp +++ b/src/exec/dml_node.cpp @@ -493,7 +493,10 @@ int DMLNode::insert_row(RuntimeState* state, SmartRecord record, bool is_update) } ret = vector_index_map[info.id]->insert_vector(_txn, vectors, pk_str, record); if (ret < 0) { - DB_WARNING_STATE(state, "vector_index fail insert, index_id: %ld", info.id); + if (ret != -2) { + // -2是向量维度对不上, 减少报警 + DB_WARNING_STATE(state, "vector_index fail insert, index_id: %ld", info.id); + } return ret; } continue; diff --git a/src/exec/dual_scan_node.cpp b/src/exec/dual_scan_node.cpp index 0e58c737..b6b3c267 100644 --- a/src/exec/dual_scan_node.cpp +++ b/src/exec/dual_scan_node.cpp @@ -116,7 +116,17 @@ int DualScanNode::open(RuntimeState* state) { ExecNode* join = _sub_query_node->get_node(pb::JOIN_NODE); _sub_query_runtime_state->is_simple_select = (join == nullptr); _sub_query_node->set_delay_fetcher_store(_delay_fetcher_store); - + // 当 delay_fetcher_store=true（UNION 向量化并发执行）时，子查询内的 JoinNode 也需要感知 delay， + // 以便 index join 路径在打开 outer node 前正确设置 delay，使其进入 EXEC_ARROW_ACERO + if (_delay_fetcher_store) { + std::vector join_nodes; + _sub_query_node->get_node(pb::JOIN_NODE, join_nodes); + for (ExecNode* node : join_nodes) { + if (node != nullptr) { + node->set_delay_fetcher_store(true); + } + } + } ret = _sub_query_node->open(_sub_query_runtime_state); if (ret < 0) { DB_WARNING("Fail to open _sub_query_node"); diff --git a/src/exec/fetcher_store.cpp b/src/exec/fetcher_store.cpp index 1fb22733..48628e8b 100755 --- a/src/exec/fetcher_store.cpp +++ b/src/exec/fetcher_store.cpp @@ -316,7 +316,6 @@ void OnRPCDone::select_resource_insulate_read_addr(pb::RegionInfo& info, bool& select_without_leader, bool& resource_insulate_read) { // offline强制隔离 - // TODO:全局索引update也会有SELECT，查询不在事务会不会有问题 _state->txn_id = 0; std::vector valid_addrs; if (info.learners_size() > 0) { @@ -389,8 +388,12 @@ void OnRPCDone::select_addr(pb::RegionInfo& info, && !_state->client_conn()->user_info->resource_tag.empty()) { insulate_resource_tag = _state->client_conn()->user_info->resource_tag; } - if (!insulate_resource_tag.empty() && _op_type == pb::OP_SELECT) { - return select_resource_insulate_read_addr(info, addr, + if (!insulate_resource_tag.empty() && _op_type == pb::OP_SELECT + && _state->client_conn() != nullptr + && _state->client_conn()->query_ctx != nullptr + && (_state->client_conn()->query_ctx->stmt_type == parser::NT_SELECT + || _state->client_conn()->query_ctx->stmt_type == parser::NT_UNION)) { + return select_resource_insulate_read_addr(info, addr, insulate_resource_tag, select_without_leader, resource_insulate_read); } diff --git a/src/exec/join_node.cpp b/src/exec/join_node.cpp index 1f9f155d..c5017f55 100644 --- a/src/exec/join_node.cpp +++ b/src/exec/join_node.cpp @@ -938,6 +938,9 @@ int JoinNode::hash_join(RuntimeState* state) { && (!_use_index_join || state->sign_exec_type == SignExecType::SIGN_EXEC_ARROW_FORCE_NO_INDEX_JOIN)) { return no_index_hash_join(state); } + if (is_delay_fetcher_store()) { + _outer_node->set_delay_fetcher_store(true); + } int ret = _outer_node->open(state); if (ret < 0) { DB_WARNING("ExecNode::outer table open fail"); @@ -1051,6 +1054,9 @@ int JoinNode::nested_loop_join(RuntimeState* state) { && (!_use_index_join || state->sign_exec_type == SignExecType::SIGN_EXEC_ARROW_FORCE_NO_INDEX_JOIN)) { return no_index_hash_join(state); } + if (is_delay_fetcher_store()) { + _outer_node->set_delay_fetcher_store(true); + } int ret = _outer_node->open(state); if (ret < 0) { DB_WARNING("ExecNode:: left table open fail"); diff --git a/src/exec/rocksdb_scan_node.cpp b/src/exec/rocksdb_scan_node.cpp index 2be14f8e..6defebd8 100644 --- a/src/exec/rocksdb_scan_node.cpp +++ b/src/exec/rocksdb_scan_node.cpp @@ -665,7 +665,10 @@ int RocksdbScanNode::open(RuntimeState* state) { _scan_conjuncts, _pre_filter_conjuncts); if (ret < 0) { - DB_FATAL("vector_index fail search, index:%ld, table:%ld", _index_info->id, _table_info->id); + if (ret != -2) { + // -2是向量维度对不上, 减少报警 + DB_FATAL("vector_index fail search, index:%ld, table:%ld", _index_info->id, _table_info->id); + } return -1; } if (!_vector_filter_conjuncts.empty()) { @@ -1270,7 +1273,10 @@ int RocksdbScanNode::index_ddl_work(RuntimeState* state, MemRow* row) { } ret = vector_index_map[index_id]->insert_vector(txn, word, pk_key.data(), record); if (ret < 0) { - DB_WARNING_STATE(state, "vector_index fail insert, index_id: %ld", index_id); + if (ret != -2) { + // -2是向量维度对不上, 减少报警 + DB_WARNING_STATE(state, "vector_index fail insert, index_id: %ld", index_id); + } return ret; } return 0; diff --git a/src/runtime/arrow_io_excutor.cpp b/src/runtime/arrow_io_excutor.cpp index d68f30e1..77002ee7 100644 --- a/src/runtime/arrow_io_excutor.cpp +++ b/src/runtime/arrow_io_excutor.cpp @@ -284,7 +284,7 @@ void GlobalArrowExecutor::execute(RuntimeState* state, arrow::Result lock(mu); - if (!done) { + while (!done) { cond.wait(lock); } } diff --git a/src/vector_index/vector_index.cpp b/src/vector_index/vector_index.cpp index 5745908c..5cd6428d 100755 --- a/src/vector_index/vector_index.cpp +++ b/src/vector_index/vector_index.cpp @@ -739,7 +739,7 @@ int VectorIndex::add_to_faiss(SmartFaissIndex faiss_index, std::vector cache_idxs; if ((int)cache_vectors.size() != _dimension) { DB_FATAL("cache_vectors.size:%lu != _dimension:%d", cache_vectors.size(), _dimension); - return -1; + return -2; } cache_idxs.emplace_back(cache_idx++); std::unordered_set need_not_cache_fields; @@ -925,7 +925,7 @@ int VectorIndex::search( from_chars_to_float_vec(search_data, search_vector); if ((int)search_vector.size() != _dimension) { DB_FATAL("search_vector.size:%lu != _dimension:%d", search_vector.size(), _dimension); - return -1; + return -2; } int64_t k = topk; std::vector flat_idxs; diff --git a/test/test_dms.cpp b/test/test_dms.cpp index 4c339aaa..a74e82ce 100644 --- a/test/test_dms.cpp +++ b/test/test_dms.cpp @@ -35,7 +35,11 @@ TEST(test_dms_utils, case_sql_rewrite) { std::vector select_fields {"k1", "k2"}; EXPECT_EQ(SqlRewrite::rewrite_to_select(stmt, select_fields, sql, rewrite_sql), 0); - EXPECT_EQ(rewrite_sql, "SELECT k1,k2 FROM tbl WHERE k3 = v3 and k4 = v4"); + EXPECT_EQ(rewrite_sql, "SELECT `k1`,`k2` FROM tbl WHERE k3 = v3 and k4 = v4"); + + select_fields = {"k1", "1"}; + EXPECT_EQ(SqlRewrite::rewrite_to_select(stmt, select_fields, sql, rewrite_sql), 0); + EXPECT_EQ(rewrite_sql, "SELECT `k1`,1 FROM tbl WHERE k3 = v3 and k4 = v4"); } { parser::SqlParser parser; @@ -49,7 +53,7 @@ TEST(test_dms_utils, case_sql_rewrite) { std::vector select_fields {"k1", "k2"}; EXPECT_EQ(SqlRewrite::rewrite_to_select(stmt, select_fields, sql, rewrite_sql), 0); - EXPECT_EQ(rewrite_sql, "SELECT k1,k2 FROM tbl WHERE k3 = v3 and k4 = v4"); + EXPECT_EQ(rewrite_sql, "SELECT `k1`,`k2` FROM tbl WHERE k3 = v3 and k4 = v4"); } { parser::SqlParser parser; diff --git a/test/test_fetcher_store.cpp b/test/test_fetcher_store.cpp index 1f5608dd..7d521aa3 100644 --- a/test/test_fetcher_store.cpp +++ b/test/test_fetcher_store.cpp @@ -15,6 +15,7 @@ #include "gtest/gtest.h" #include "schema_factory.h" #include "fetcher_store.h" +#include "query_context.h" #include "proto/meta.interface.pb.h" #include "proto/store.interface.pb.h" #include @@ -85,8 +86,17 @@ class FetcherStoreTest : public testing::Test { } } } + void set_select_client_conn(RuntimeState& state) { + auto ctx = std::make_shared(); + ctx->stmt_type = parser::NT_SELECT; + auto conn = std::make_shared(); + conn->query_ctx = ctx; + _select_conns.push_back(conn); + state.set_client_conn(conn.get()); + } std::map _instance_info; std::map _instance_logical_map; + std::vector> _select_conns; }; // 非select一定选leader @@ -245,6 +255,7 @@ TEST_F(FetcherStoreTest, test_resource_insulate_read_without_learner) { // 选watt peer FetcherStore fetcher_store; RuntimeState state; + set_select_client_conn(state); ExecNode store_request; OnSingleRPCDone done(&fetcher_store, &state, &store_request, ®ion, 1, 1, 0, 0, pb::OP_SELECT, false); for (int i = 0; i < 5; ++i) { @@ -257,6 +268,7 @@ TEST_F(FetcherStoreTest, test_resource_insulate_read_without_learner) { // watt peer返回not_leader读失败，选其他peer FetcherStore fetcher_store; RuntimeState state; + set_select_client_conn(state); ExecNode store_request; OnSingleRPCDone done(&fetcher_store, &state, &store_request, ®ion, 1, 1, 0, 0, pb::OP_SELECT, false); done.select_addr(); @@ -279,6 +291,7 @@ TEST_F(FetcherStoreTest, test_resource_insulate_read_without_learner) { // watt peer can not access，选其他peer FetcherStore fetcher_store; RuntimeState state; + set_select_client_conn(state); ExecNode store_request; OnSingleRPCDone done(&fetcher_store, &state, &store_request, ®ion, 1, 1, 0, 0, pb::OP_SELECT, false); done.select_addr(); @@ -321,6 +334,7 @@ TEST_F(FetcherStoreTest, test_resource_insulate_read_with_learner) { FLAGS_fetcher_learner_read = false; FetcherStore fetcher_store; RuntimeState state; + set_select_client_conn(state); ExecNode store_request; OnSingleRPCDone done(&fetcher_store, &state, &store_request, ®ion, 1, 1, 0, 0, pb::OP_SELECT, false); for (int i = 0; i < 5; ++i) { @@ -335,6 +349,7 @@ TEST_F(FetcherStoreTest, test_resource_insulate_read_with_learner) { // learner方式失败选peer FetcherStore fetcher_store; RuntimeState state; + set_select_client_conn(state); ExecNode store_request; OnSingleRPCDone done(&fetcher_store, &state, &store_request, ®ion, 1, 1, 0, 0, pb::OP_SELECT, false); done.select_addr(); @@ -358,6 +373,7 @@ TEST_F(FetcherStoreTest, test_resource_insulate_read_with_learner) { // learner not access选peer, peer失败选其他peer FetcherStore fetcher_store; RuntimeState state; + set_select_client_conn(state); ExecNode store_request; OnSingleRPCDone done(&fetcher_store, &state, &store_request, ®ion, 1, 1, 0, 0, pb::OP_SELECT, false); done.select_addr(); @@ -534,6 +550,7 @@ TEST_F(FetcherStoreTest, test_retry_later_choose_other_peer_read2) { FLAGS_fetcher_resource_tag = ""; FetcherStore fetcher_store; RuntimeState state; + set_select_client_conn(state); ExecNode store_request; OnSingleRPCDone done(&fetcher_store, &state, &store_request, ®ion, 1, 1, 0, 0, pb::OP_SELECT, false); done.select_addr();