BE optimize right join

Author: zclllyybb

be/src/pipeline/dependency.h

@@ -659,20 +659,18 @@ struct HashJoinSharedState : public JoinSharedState {
     // Cached build ASOF column (computed once by executing expression on build_block)
     vectorized::ColumnPtr asof_build_col;
 
-    // ASOF RIGHT JOIN: column-based cache for efficient accumulation across probe batches
-    // Per build row (1-indexed, 0 unused): whether a matching probe was found
+    // ASOF RIGHT JOIN: per-task local caches submitted during finish_probing, then merged
+    // Each probe task accumulates matches lock-free in its own AsofRightLocalCache.
+    // The last task to enter finish_probing merges all caches and populates merged_* for output.
+    std::vector<AsofRightLocalCache> asof_right_local_caches;
+    std::mutex asof_right_submit_mutex; // only for submitting local caches (one lock per task)
+    // Merged result (populated once by last task during finish_probing merge)
     std::vector<uint8_t> asof_right_has_match;
-    // Per build row: index into probe cache for the best matching probe row (-1 = none)
     std::vector<int32_t> asof_right_cache_idx;
-    // Append-only column of best probe ASOF values (for efficient compare_at, no Field)
-    vectorized::MutableColumnPtr asof_right_best_probe_values;
-    // Append-only cache of full probe row data (for output in finish_probing)
     std::unique_ptr<vectorized::MutableBlock> asof_right_probe_cache;
     size_t asof_right_output_idx = 0;
-    // Barrier: counts down from num_probe_tasks. Only the last task (counter→0) outputs.
+    // Barrier: counts down from num_probe_tasks. Only the last task (counter→0) merges+outputs.
     std::atomic<int32_t> asof_right_remaining_probers {0};
-    // Mutex protecting concurrent writes to probe_cache/best_probe_values/has_match/cache_idx
-    std::mutex asof_right_cache_mutex;
 };
 
 struct PartitionedHashJoinSharedState

be/src/pipeline/exec/hashjoin_build_sink.cpp

@@ -378,16 +378,10 @@ Status HashJoinBuildSinkLocalState::build_asof_index(vectorized::Block& block) {
         }
     }
 
-    // Pre-initialize ASOF RIGHT JOIN state during build phase so it's ready
-    // before any probe operator calls finish_probing (avoids race condition)
+    // Pre-initialize ASOF RIGHT JOIN barrier counter during build phase.
+    // Per-task local caches and merged result are created during probe/finish_probing.
     if (p._join_op == TJoinOp::ASOF_RIGHT_INNER_JOIN ||
         p._join_op == TJoinOp::ASOF_RIGHT_OUTER_JOIN) {
-        // Use build_block rows (includes sentinel at index 0) to match _build_block indexing.
-        // hash_table->size() returns next.size() which may be larger than build_block rows.
-        size_t build_block_size =
-                _shared_state->build_block ? _shared_state->build_block->rows() : 0;
-        _shared_state->asof_right_has_match.resize(build_block_size, 0);
-        _shared_state->asof_right_cache_idx.resize(build_block_size, -1);
         _shared_state->asof_right_remaining_probers.store(
                 static_cast<int32_t>(
                         std::max(_shared_state->hash_table_variant_vector.size(), (size_t)1)),

be/src/pipeline/exec/join/process_hash_table_probe.h

@@ -41,6 +41,16 @@ using MutableColumns = std::vector<vectorized::MutableColumnPtr>;
 using NullMap = vectorized::ColumnUInt8::Container;
 using ConstNullMapPtr = const NullMap*;
 
+// Per-task local cache for ASOF RIGHT JOIN lock-free probe accumulation.
+// Each probe task appends qualifying probe entries (one per probe row per bucket).
+// Caches are merged by the last task in finish_probing: entries are grouped by bucket,
+// sorted by ASOF value, and binary-searched per build row for O(log P) matching.
+struct AsofRightLocalCache {
+    std::vector<uint32_t> bucket_ids;   // bucket_id for each appended probe entry
+    MutableColumnPtr probe_asof_values; // probe ASOF column values (non-nullable) for sort/search
+    std::unique_ptr<vectorized::MutableBlock> probe_cache; // full probe row data for output
+};
+
 template <int JoinOpType>
 struct ProcessHashTableProbe {
     ProcessHashTableProbe(HashJoinProbeLocalState* parent, int batch_size);
@@ -146,6 +156,9 @@ struct ProcessHashTableProbe {
     // in the hash table.
     // -1 means null, 0 means false, 1 means true
     DorisVector<int8_t> mark_join_flags;
+
+    // ASOF RIGHT JOIN: per-task local cache for lock-free probe accumulation
+    std::unique_ptr<AsofRightLocalCache> _asof_right_local_cache;
 };
 
 } // namespace pipeline