@@ -49,6 +49,13 @@ struct VectorRecord {
   const int64_t timestamp_;  // Timestamp for the record
   VectorData data_;          // Shared pointer to the vector data
 
+  /// Bitmask of subtask indices this record was routed to (multicast routing).
+  /// Bit i set  ⇒  record lives in subtask i's WindowState.
+  /// Used by Owner-Computes dedup: only the lowest-indexed subtask in the
+  /// intersection of query_mask & candidate_mask emits a match.
+  /// Supports up to 64 subtasks (uint64_t). 0 = unset / legacy path.
+  uint64_t routing_mask_ = 0;
+
   // Constructor with move semantics for efficiency
   VectorRecord(const uint64_t &uid, const int64_t &timestamp, VectorData &&data);
 
 
@@ -105,6 +105,13 @@ namespace sageFlow {
    private:
     enum class InternalIndexKind { NONE, IVF, BRUTEFORCE, VAMANA, HDR_TREE };  // 可扩展
 
+    /// Deterministic hash combining for match pair (left_uid, right_uid).
+    /// Used in fallback dedup path when routing_mask is unavailable.
+    static uint64_t combinedMatchId(uint64_t left_uid, uint64_t right_uid) {
+        uint64_t x = left_uid ^ (right_uid + 0x9e3779b97f4a7c15ULL + (left_uid << 6) + (left_uid >> 2));
+        return x;
+    }
+
     bool createIndexPair(IndexType type, const std::string& prefix);
     bool createIndexPair(IndexType type, const std::string& prefix, const IndexParameters& params);
 
 
@@ -48,6 +48,7 @@ struct JoinMetrics {
   std::atomic<uint64_t> total_records_left{0};    ///< Total records processed on left side
   std::atomic<uint64_t> total_records_right{0};   ///< Total records processed on right side
   std::atomic<uint64_t> total_emits{0};           ///< Total results emitted
+  std::atomic<uint64_t> owner_dedup_count{0};     ///< Matches skipped by owner-computes rule
   std::atomic<uint64_t> window_records_left_completed{0};   ///< Records expired from left window
   std::atomic<uint64_t> window_records_right_completed{0};  ///< Records expired from right window
 
@@ -89,7 +90,7 @@ struct JoinMetrics {
       join_function_ns = emit_ns = lock_wait_ns = 0;
     window_insert_count = index_op_count = expire_count = candidate_fetch_count = similarity_count =
       join_function_count = emit_count = lock_wait_count = 0;
-    total_records_left = total_records_right = total_emits = 0;
+    total_records_left = total_records_right = total_emits = owner_dedup_count = 0;
     window_records_left_completed = window_records_right_completed = 0;
     apply_processing_ns = apply_processing_count = e2e_latency_ns = e2e_latency_count = 0;
     qiq_q1_ns = qiq_q1_count = qiq_insert_ns = qiq_insert_count = qiq_q2_ns = qiq_q2_count = 0;
@@ -115,7 +116,7 @@ struct JoinMetrics {
     EMIT(similarity_ns) EMIT(join_function_ns) EMIT(emit_ns) EMIT(lock_wait_ns)
     EMIT(window_insert_count) EMIT(index_op_count) EMIT(expire_count) EMIT(candidate_fetch_count)
     EMIT(similarity_count) EMIT(join_function_count) EMIT(emit_count) EMIT(lock_wait_count)
-    EMIT(total_records_left) EMIT(total_records_right) EMIT(total_emits)
+    EMIT(total_records_left) EMIT(total_records_right) EMIT(total_emits) EMIT(owner_dedup_count)
     EMIT(window_records_left_completed) EMIT(window_records_right_completed)
     EMIT(apply_processing_ns) EMIT(apply_processing_count)
     EMIT(e2e_latency_ns) EMIT(e2e_latency_count)
 
@@ -47,13 +47,21 @@ void ResultPartition::emit(Response&& data, int slot) const {
   // 注意：isBroadcast() 可能在 partition() 调用后改变状态（训练完成后变为 false）
   if (partitioner_->isBroadcast()) {
     // 广播模式：将数据发送到所有通道
+    // Build routing_mask: all channels are targets
+    uint64_t route_mask = 0;
+    for (size_t i = 0; i < output_channels_.size() && i < 64; ++i) {
+      route_mask |= (uint64_t{1} << i);
+    }
+
     for (size_t i = 0; i < output_channels_.size(); ++i) {
       if (i == output_channels_.size() - 1) {
         // 最后一个通道，移动数据
+        if (data.record_) data.record_->routing_mask_ = route_mask;
         pushWithRetry(output_channels_[i], {std::move(data), slot});
       } else {
         // 其他通道，复制数据
         Response data_copy{data.type_, data.record_ ? std::make_unique<VectorRecord>(*data.record_) : nullptr};
+        if (data_copy.record_) data_copy.record_->routing_mask_ = route_mask;
         pushWithRetry(output_channels_[i], {std::move(data_copy), slot});
       }
     }
@@ -62,23 +70,36 @@ void ResultPartition::emit(Response&& data, int slot) const {
     // 这用于边界向量复制（如 ClusteredJoin 的 multicast_k > 1）
     auto target_channels = partitioner_->partitionMulti(data, output_channels_.size());
 
+    // Build routing_mask from multicast targets
+    uint64_t route_mask = 0;
+    for (size_t ch : target_channels) {
+      if (ch < 64) route_mask |= (uint64_t{1} << ch);
+    }
+
     if (target_channels.size() == 1) {
       // 只有一个目标，直接移动
+      if (data.record_) data.record_->routing_mask_ = route_mask;
       pushWithRetry(output_channels_[target_channels[0]], {std::move(data), slot});
     } else {
       // 多个目标，复制数据到前 n-1 个，移动到最后一个
       for (size_t i = 0; i < target_channels.size(); ++i) {
         if (i == target_channels.size() - 1) {
+          if (data.record_) data.record_->routing_mask_ = route_mask;
           pushWithRetry(output_channels_[target_channels[i]], {std::move(data), slot});
         } else {
           Response data_copy{data.type_, data.record_ ? std::make_unique<VectorRecord>(*data.record_) : nullptr};
+          if (data_copy.record_) data_copy.record_->routing_mask_ = route_mask;
           pushWithRetry(output_channels_[target_channels[i]], {std::move(data_copy), slot});
         }
       }
     }
   } else {
     // 单播模式：发送到单个目标分区
     size_t channel_index = partitioner_->partition(data, output_channels_.size());
+    // Single target: set routing_mask to just that channel
+    if (data.record_ && channel_index < 64) {
+      data.record_->routing_mask_ = (uint64_t{1} << channel_index);
+    }
     pushWithRetry(output_channels_[channel_index], {std::move(data), slot});
   }
 }
 
@@ -975,6 +975,9 @@ void JoinOperator::executeJoinWithState(
             t_joinF.stop();
             t_similarity.resume();
             if (res.record_) {
+                // Propagate candidate's routing_mask into the join result so that
+                // the Owner-Computes dedup in the emit stage can use it.
+                res.record_->routing_mask_ = cand->routing_mask_;
                 local_return_pool.emplace_back(left_slot_id_, std::move(res.record_));
             }
         } catch (const std::exception& e) {
@@ -1243,6 +1246,15 @@ auto JoinOperator::apply(Response&& record, int slot, Collector& collector,
                 }
             }
 
+            // Build routing bitmask for Owner-Computes dedup.
+            // Bit i set  ⇒  record lives in subtask i's WindowState.
+            uint64_t route_mask = 0;
+            for (size_t t : target_subtasks) {
+                if (t < 64) route_mask |= (uint64_t{1} << t);
+            }
+            data_ptr->routing_mask_ = route_mask;
+            data_for_join->routing_mask_ = route_mask;
+
             for (size_t target_subtask : target_subtasks) {
                 auto data_for_insert = std::make_unique<VectorRecord>(*data_ptr);
                 updateSideWithState(current_state, index_id, std::move(data_for_insert), now_time_stamp, slot, target_subtask);
@@ -1349,7 +1361,44 @@ auto JoinOperator::apply(Response&& record, int slot, Collector& collector,
     // 发送 Join 结果
     {
         MetricsTimer t_emit(JoinMetrics::instance().emit_ns);
+        // Owner-Computes dedup: for each match (query, candidate), only the
+        // lowest-indexed subtask that holds BOTH records emits the result.
+        // routing_mask_ is a bitmask of subtasks each record was routed to.
+        // intersection = query_mask & cand_mask  ⇒  subtasks holding both.
+        // owner = __builtin_ctzll(intersection)  ⇒  lowest such subtask.
+        // If current_subtask == owner → emit.  Otherwise → skip.
+        //
+        // Fallback: if either routing_mask_ is 0 (unset / legacy path),
+        // fall back to intra-apply hash-set dedup (apply_seen) as before.
+        std::unordered_set<uint64_t> apply_seen;  // only used in fallback path
         for (auto& p : local_return_pool) {
+            if (p.second && parallelism_ > 1) {
+                const uint64_t left_uid = (p.first == left_slot_id_) ? p.second->uid_ : data_for_join->uid_;
+                const uint64_t right_uid = (p.first == left_slot_id_) ? data_for_join->uid_ : p.second->uid_;
+
+                const uint64_t query_mask = data_for_join->routing_mask_;
+                const uint64_t cand_mask  = p.second->routing_mask_;
+
+                if (query_mask != 0 && cand_mask != 0) {
+                    // Owner-Computes path (preferred)
+                    const uint64_t intersection = query_mask & cand_mask;
+                    if (intersection == 0) {
+                        // Should not happen: defensive emit anyway.
+                    } else {
+                        const size_t owner = static_cast<size_t>(__builtin_ctzll(intersection));
+                        if (owner != subtask_index) {
+                            metrics_increment(JoinMetrics::instance().owner_dedup_count);
+                            continue;  // another subtask is the owner
+                        }
+                    }
+                } else {
+                    // Fallback: routing_mask not set
+                    const uint64_t cid = combinedMatchId(left_uid, right_uid);
+                    if (!apply_seen.insert(cid).second) {
+                        continue;
+                    }
+                }
+            }
             Response out{ResponseType::Record, std::move(p.second)};
             collector.collect(std::make_unique<Response>(std::move(out)), p.first);
             metrics_increment(JoinMetrics::instance().total_emits);