[Refactor] Localize MERGE duplicate-match check above a shuffle join, keyed by slot id

Follow-up to the MERGE INTO planner/executor commits. Rework the
duplicate-match (cardinality) check so its correctness rests on the merge
join rather than on a bespoke distribution, mirroring how Spark/Trino keep
the check local to the join.

* Analyzer pins HINT_JOIN_SHUFFLE on the source-target merge join (same
  mechanism as IcebergEqualityDeleteRewriteRule), so the target side is
  never broadcast/replicated: each target row is owned by one instance and
  all source rows matching it meet on the same node, and the hash join's
  probe side is locally shuffled by join key so they meet on the same driver.

* The planner places EnforceUniqueRowLocatorNode directly above the merge
  join, in the join's fragment, before any write-distribution exchange. The
  required property is now only the Iceberg write-clustering shuffle
  (partition columns for partitioned targets); it no longer carries any
  duplicate-check responsibility. This decouples the cardinality check from
  the write distribution and lets non-partitioned inserts spread by join key
  instead of funneling onto one node.

* validateMergeJoinKeepsTargetUnreplicated rejects plans that would let a
  target row be matched by source rows on different drivers: a nestloop join
  is accepted only when its source is a constant relation of at most one row
  (no scan, exact cardinality <= 1); a scanned or multi-row-constant nestloop
  (no source-target equality in ON) is rejected with a clear error.

* The unique-key columns are identified by SLOT ID; the BE resolves them
  through the chunk's slot-id map, removing any FE-side prediction of the BE
  chunk column order.

* Remove the now-unused custom local column-hash exchange
  (ColumnHashPartitioner / ColumnHashPartitionExchanger /
  interpolate_local_column_hash_partition_exchange) and its NULL-key
  round-robin handling; the merge join's own distribution supersedes them.

* Rename EnforceUnique{Node,Operator} to EnforceUniqueRowLocator{Node,Operator}
  (and the TEnforceUniqueRowLocatorNode thrift) to reflect that it is an
  Iceberg row-locator-specific check, not a general uniqueness operator.

Tests: FE plan tests assert the merge join is a shuffle join with the check
in its fragment, the constant single-row nestloop is accepted, and the
non-equi / multi-row-constant nestloop is rejected; BE tests cover slot-id
resolution and missing-slot rejection.

Signed-off-by: GavinMar <yangguansuo@starrocks.com>

Author: GavinMar

be/src/exec/CMakeLists.txt

@@ -245,7 +245,7 @@ set(PIPELINE_OPERATORS_BATCH_SRCS
     pipeline/bucket_process_operator.cpp
     pipeline/table_function_operator.cpp
     pipeline/assert_num_rows_operator.cpp
-    pipeline/enforce_unique_operator.cpp
+    pipeline/enforce_unique_row_locator_operator.cpp
     pipeline/set/union_passthrough_operator.cpp
     pipeline/set/union_const_source_operator.cpp
     pipeline/set/raw_values_source_operator.cpp
@@ -355,7 +355,7 @@ set(EXEC_NON_PIPELINE_FILES
     parquet_builder.cpp
     file_scan_node.cpp
     assert_num_rows_node.cpp
-    enforce_unique_node.cpp
+    enforce_unique_row_locator_node.cpp
     intersect_hash_set.cpp
     intersect_node.cpp
     json_parser.cpp

be/src/exec/enforce_unique_node.cpp

@@ -0,0 +1,75 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "exec/enforce_unique_row_locator_node.h"
+
+#include "exec/pipeline/enforce_unique_row_locator_operator.h"
+#include "exec/pipeline/exec_node_pipeline_adapter.h"
+#include "exec/pipeline/limit_operator.h"
+#include "exec/pipeline/pipeline_builder.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "runtime/runtime_state.h"
+
+namespace starrocks {
+
+EnforceUniqueRowLocatorNode::EnforceUniqueRowLocatorNode(ObjectPool* pool, const TPlanNode& tnode,
+                                                         const DescriptorTbl& descs)
+        : PipelineNode(pool, tnode, descs) {
+    if (tnode.__isset.enforce_unique_row_locator_node &&
+        tnode.enforce_unique_row_locator_node.__isset.unique_key_slot_ids) {
+        const auto& slot_ids = tnode.enforce_unique_row_locator_node.unique_key_slot_ids;
+        _unique_key_slot_ids.assign(slot_ids.begin(), slot_ids.end());
+    }
+}
+
+Status EnforceUniqueRowLocatorNode::init(const TPlanNode& tnode, RuntimeState* state) {
+    RETURN_IF_ERROR(ExecNode::init(tnode, state));
+    return Status::OK();
+}
+
+void EnforceUniqueRowLocatorNode::close(RuntimeState* state) {
+    if (is_closed()) {
+        return;
+    }
+    ExecNode::close(state);
+}
+
+StatusOr<pipeline::OpFactories> EnforceUniqueRowLocatorNode::decompose_to_pipeline(
+        pipeline::PipelineBuilderContext* context) {
+    using namespace pipeline;
+
+    ASSIGN_OR_RETURN(auto ops, _children[0]->decompose_to_pipeline(context));
+    // No exchange is interpolated here: each driver keeps a local seen-set, and the
+    // co-location of all matches of one target row is guaranteed by the MERGE join
+    // below this node. The FE places this node directly above the merge join and
+    // forbids broadcasting/replicating the target side (shuffle join hint +
+    // MergeIntoPlanner::validateMergeJoinKeepsTargetUnreplicated), so cross-BE the
+    // join key owns each target row on one instance, and within a BE the hash join's
+    // probe side is locally shuffled by join key across drivers. NULL-key rows
+    // (NOT-MATCHED insert rows) are exempt from the check and may sit on any driver.
+    auto factory = std::make_shared<EnforceUniqueRowLocatorOperatorFactory>(context->next_operator_id(), id(),
+                                                                            _unique_key_slot_ids);
+    ops.emplace_back(std::move(factory));
+
+    // Initialize OperatorFactory's fields involving runtime filters.
+    auto&& rc_rf_probe_collector = std::make_shared<RcRfProbeCollector>(1, std::move(this->runtime_filter_collector()));
+    pipeline::init_runtime_filter_for_operator(*this, ops.back().get(), context, rc_rf_probe_collector);
+
+    if (limit() != -1) {
+        ops.emplace_back(std::make_shared<LimitOperatorFactory>(context->next_operator_id(), id(), limit()));
+    }
+    return ops;
+}
+
+} // namespace starrocks

be/src/exec/enforce_unique_row_locator_node.cpp

@@ -16,6 +16,7 @@
 
 #include <vector>
 
+#include "common/global_types.h"
 #include "common/statusor.h"
 #include "exec/pipeline_node.h"
 #include "gen_cpp/PlanNodes_types.h"
@@ -25,18 +26,22 @@ namespace starrocks {
 // Plan node that enforces uniqueness of (file_path, row_position) keys
 // across the data stream. Used by MERGE INTO to guarantee that each target
 // row is matched by at most one source row.
-class EnforceUniqueNode final : public PipelineNode {
+//
+// The key columns are identified by SLOT ID and resolved per chunk through the
+// chunk's slot-id map, so this node is insensitive to the physical column order
+// of the child's output chunk.
+class EnforceUniqueRowLocatorNode final : public PipelineNode {
 public:
-    EnforceUniqueNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs);
-    ~EnforceUniqueNode() override = default;
+    EnforceUniqueRowLocatorNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs);
+    ~EnforceUniqueRowLocatorNode() override = default;
 
     Status init(const TPlanNode& tnode, RuntimeState* state = nullptr) override;
     void close(RuntimeState* state) override;
 
     StatusOr<pipeline::OpFactories> decompose_to_pipeline(pipeline::PipelineBuilderContext* context) override;
 
 private:
-    std::vector<int32_t> _unique_key_col_indices;
+    std::vector<SlotId> _unique_key_slot_ids;
 };
 
 } // namespace starrocks

be/src/exec/enforce_unique_node.h …c/exec/enforce_unique_row_locator_node.hbe/src/exec/enforce_unique_node.h renamed to be/src/exec/enforce_unique_row_locator_node.h be/src/exec/enforce_unique_node.h renamed to be/src/exec/enforce_unique_row_locator_node.h

@@ -53,7 +53,7 @@
 #include "exec/cross_join_node.h"
 #include "exec/dict_decode_node.h"
 #include "exec/empty_set_node.h"
-#include "exec/enforce_unique_node.h"
+#include "exec/enforce_unique_row_locator_node.h"
 #include "exec/except_node.h"
 #include "exec/exchange_node.h"
 #include "exec/fetch_node.h"
@@ -368,8 +368,8 @@ Status ExecFactory::create_vectorized_node(RuntimeState* state, ObjectPool* pool
         CREATE_NODE(ConnectorScanNode, pool, new_node, descs);
         return Status::OK();
     }
-    case TPlanNodeType::ENFORCE_UNIQUE_NODE:
-        CREATE_NODE(EnforceUniqueNode, pool, tnode, descs);
+    case TPlanNodeType::ENFORCE_UNIQUE_ROW_LOCATOR_NODE:
+        CREATE_NODE(EnforceUniqueRowLocatorNode, pool, tnode, descs);
         return Status::OK();
     default:
         return Status::InternalError(strings::Substitute("Vectorized engine not support node: $0", tnode.node_type));

be/src/exec/exec_factory.cpp

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "exec/pipeline/enforce_unique_operator.h"
+#include "exec/pipeline/enforce_unique_row_locator_operator.h"
 
 #include "column/binary_column.h"
 #include "column/column_helper.h"
@@ -21,14 +21,14 @@
 
 namespace starrocks::pipeline {
 
-Status EnforceUniqueOperator::prepare(RuntimeState* state) {
+Status EnforceUniqueRowLocatorOperator::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(Operator::prepare(state));
     _seen_rows_counter = ADD_COUNTER(_unique_metrics, "SeenSetRows", TUnit::UNIT);
     _seen_memory_counter = ADD_COUNTER(_unique_metrics, "SeenSetMemoryUsage", TUnit::BYTES);
     return Status::OK();
 }
 
-int64_t EnforceUniqueOperator::_seen_memory_usage() const {
+int64_t EnforceUniqueRowLocatorOperator::_seen_memory_usage() const {
     // flat_hash_set/map store entries inline in the bucket array; capacity()
     // reflects allocated buckets. Path strings are heap-allocated separately.
     int64_t bytes = static_cast<int64_t>(_seen.capacity()) * sizeof(std::pair<int32_t, int64_t>);
@@ -39,7 +39,7 @@ int64_t EnforceUniqueOperator::_seen_memory_usage() const {
     return bytes;
 }
 
-void EnforceUniqueOperator::close(RuntimeState* state) {
+void EnforceUniqueRowLocatorOperator::close(RuntimeState* state) {
     if (_seen_rows_counter != nullptr) {
         COUNTER_SET(_seen_rows_counter, static_cast<int64_t>(_seen.size()));
         COUNTER_SET(_seen_memory_counter, _seen_memory_usage());
@@ -51,38 +51,47 @@ void EnforceUniqueOperator::close(RuntimeState* state) {
     Operator::close(state);
 }
 
-bool EnforceUniqueOperator::has_output() const {
+bool EnforceUniqueRowLocatorOperator::has_output() const {
     return _output_chunk != nullptr;
 }
 
-bool EnforceUniqueOperator::need_input() const {
+bool EnforceUniqueRowLocatorOperator::need_input() const {
     return !_input_finished && _output_chunk == nullptr;
 }
 
-bool EnforceUniqueOperator::is_finished() const {
+bool EnforceUniqueRowLocatorOperator::is_finished() const {
     return _input_finished && _output_chunk == nullptr;
 }
 
-Status EnforceUniqueOperator::set_finishing(RuntimeState* state) {
+Status EnforceUniqueRowLocatorOperator::set_finishing(RuntimeState* state) {
     _input_finished = true;
     return Status::OK();
 }
 
-StatusOr<ChunkPtr> EnforceUniqueOperator::pull_chunk(RuntimeState* state) {
+StatusOr<ChunkPtr> EnforceUniqueRowLocatorOperator::pull_chunk(RuntimeState* state) {
     return std::move(_output_chunk);
 }
 
-Status EnforceUniqueOperator::push_chunk(RuntimeState* state, const ChunkPtr& chunk) {
+Status EnforceUniqueRowLocatorOperator::push_chunk(RuntimeState* state, const ChunkPtr& chunk) {
     size_t num_rows = chunk->num_rows();
     if (num_rows == 0) {
         return Status::OK();
     }
 
-    // Key-count validation runs once in EnforceUniqueOperatorFactory::prepare.
-    DCHECK_EQ(_unique_key_col_indices.size(), 2);
+    // Key-count validation runs once in EnforceUniqueRowLocatorOperatorFactory::prepare.
+    DCHECK_EQ(_unique_key_slot_ids.size(), 2);
 
-    auto file_path_col = chunk->get_column_by_index(_unique_key_col_indices[0]);
-    auto row_pos_col = chunk->get_column_by_index(_unique_key_col_indices[1]);
+    // Resolve the key columns by slot id. Chunk::get_column_by_slot_id is only
+    // DCHECK-guarded, so verify existence explicitly: in RELEASE builds a missing
+    // slot would silently default-construct a map entry pointing at column 0.
+    for (SlotId slot_id : _unique_key_slot_ids) {
+        if (!chunk->is_slot_exist(slot_id)) {
+            return Status::InternalError("EnforceUniqueRowLocatorOperator: key slot " + std::to_string(slot_id) +
+                                         " does not exist in the input chunk");
+        }
+    }
+    auto file_path_col = chunk->get_column_by_slot_id(_unique_key_slot_ids[0]);
+    auto row_pos_col = chunk->get_column_by_slot_id(_unique_key_slot_ids[1]);
 
     // Determine if columns are nullable
     const NullableColumn* file_path_nullable = nullptr;
@@ -96,24 +105,23 @@ Status EnforceUniqueOperator::push_chunk(RuntimeState* state, const ChunkPtr& ch
 
     // Get the underlying data columns (unwrap nullable/const).
     //
-    // Defense-in-depth type check: the FE resolves unique_key_col_indices by
-    // slot ID now (MergeIntoPlanner::insertEnforceUniqueNode), so this shouldn't
-    // fire in practice. But down_cast is unchecked in RELEASE builds, so without
-    // these guards a bad index — i.e. the historical Bug #1 where indices [0,1]
-    // pointed at non-binary/non-int64 data columns — would dereference garbage
-    // and SIGSEGV instead of producing a diagnosable error. Verify the unwrapped
-    // types first and fail with a clear message if they mismatch.
+    // Defense-in-depth type check: the keys are resolved by slot id, so a
+    // mismatch means the FE bound the wrong slots. down_cast is unchecked in
+    // RELEASE builds — without these guards a wrong slot pointing at a
+    // non-binary/non-int64 column would dereference garbage and SIGSEGV
+    // instead of producing a diagnosable error. Verify the unwrapped types
+    // first and fail with a clear message if they mismatch.
     const Column* file_path_inner = ColumnHelper::get_data_column(file_path_col.get());
     if (!file_path_inner->is_binary()) {
-        return Status::InternalError("EnforceUniqueOperator: expected _file key column (index " +
-                                     std::to_string(_unique_key_col_indices[0]) + ") to be a binary column, got " +
+        return Status::InternalError("EnforceUniqueRowLocatorOperator: expected _file key column (slot " +
+                                     std::to_string(_unique_key_slot_ids[0]) + ") to be a binary column, got " +
                                      file_path_inner->get_name());
     }
     const Column* row_pos_data_col = ColumnHelper::get_data_column(row_pos_col.get());
     const auto* row_pos_typed = dynamic_cast<const FixedLengthColumn<int64_t>*>(row_pos_data_col);
     if (row_pos_typed == nullptr) {
-        return Status::InternalError("EnforceUniqueOperator: expected _pos key column (index " +
-                                     std::to_string(_unique_key_col_indices[1]) +
+        return Status::InternalError("EnforceUniqueRowLocatorOperator: expected _pos key column (slot " +
+                                     std::to_string(_unique_key_slot_ids[1]) +
                                      ") to be FixedLengthColumn<int64_t>, got " + row_pos_data_col->get_name());
     }
     const auto* file_path_data = down_cast<const BinaryColumn*>(file_path_inner);
@@ -154,16 +162,16 @@ Status EnforceUniqueOperator::push_chunk(RuntimeState* state, const ChunkPtr& ch
     return Status::OK();
 }
 
-Status EnforceUniqueOperatorFactory::prepare(RuntimeState* state) {
+Status EnforceUniqueRowLocatorOperatorFactory::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(OperatorFactory::prepare(state));
-    if (_unique_key_col_indices.size() != 2) {
-        return Status::InternalError("EnforceUniqueOperator expects exactly 2 key columns, got " +
-                                     std::to_string(_unique_key_col_indices.size()));
+    if (_unique_key_slot_ids.size() != 2) {
+        return Status::InternalError("EnforceUniqueRowLocatorOperator expects exactly 2 key columns, got " +
+                                     std::to_string(_unique_key_slot_ids.size()));
     }
     return Status::OK();
 }
 
-void EnforceUniqueOperatorFactory::close(RuntimeState* state) {
+void EnforceUniqueRowLocatorOperatorFactory::close(RuntimeState* state) {
     OperatorFactory::close(state);
 }