Streaming join key-domain pushdown: prune left backfill by a snapshot of right-side join keys

cmake/autogenerated_versions.txt

@@ -2,7 +2,7 @@
 
 # NOTE: has nothing common with DBMS_TCP_PROTOCOL_VERSION,
 # only DBMS_TCP_PROTOCOL_VERSION should be incremented on protocol changes.
-SET(VERSION_REVISION 158)
+SET(VERSION_REVISION 159)
 SET(VERSION_MAJOR 3)
 SET(VERSION_MINOR 0)
 SET(VERSION_PATCH 23)

src/Common/ProfileEvents.cpp

@@ -201,6 +201,15 @@
     M(SelectedMarks, "Number of marks (index granules) selected to read from a MergeTree table.") \
     M(SelectedRows, "Number of rows SELECTed from all tables.") \
     M(SelectedBytes, "Number of bytes (uncompressed; for columns as they stored in memory) SELECTed from all tables.") \
+    /* proton: starts. */ \
+    M(StreamingJoinKeyDomainBuilt, "Number of streaming join right-side key domains built for historical backfill pushdown.") \
+    M(StreamingJoinKeyDomainStorageFilterPushed, "Number of streaming join key-domain filters pushed into historical storage reads.") \
+    M(StreamingJoinKeyDomainPostFilterApplied, \
+      "Number of defensive post-read streaming join key-domain filters added to historical backfill plans.") \
+    M(StreamingJoinKeyDomainShardPruned, "Number of streaming join historical backfill reads whose shard set was pruned by a key domain.") \
+    M(StreamingJoinKeyDomainPrimaryKeyPruned, \
+      "Number of streaming join historical backfill reads whose primary-key mark ranges were pruned by a key domain.") \
+    /* proton: ends. */ \
 \
     M(WaitMarksLoadMicroseconds, "Time spent loading marks") \
     M(BackgroundLoadingMarksTasks, "Number of background tasks for loading marks") \

src/Core/Settings.h

@@ -1142,6 +1142,11 @@ Possible values: non-negative numbers. Note that if the value is too small or to
     M(UInt64, join_max_buffered_bytes, 524288000, "Max buffered bytes for stream to stream join", 0) \
     M(UInt64, join_buffered_data_block_size, 0, "For streaming join, when buffered data in memory, the data block size directs to merge small data blocks to form bigger ones to improve memory efficiency. 0 means disable merging small data blocks.", 0) \
     M(Int64, join_quiesce_threshold_ms, 0, "For streaming join, when left or right stream is in quiesce, the maximum time to wait before the join.", 0) \
+    M(Bool, enable_streaming_join_key_domain_pushdown, false, "Enable key-domain filtering for backfilled INNER LATEST streaming joins in normal execution mode. The optimization requires a historical seek_to, snapshots bounded right-side join keys, and filters only the left historical backfill; live streaming reads are not pruned.", 0) \
+    M(Bool, enable_streaming_join_key_domain_shard_pruning, false, "Enable historical-only shard pruning for streaming join key-domain pushdown when the left stream has a deterministic sharding expression compatible with the captured key domain. Live streaming reads remain on all query shards.", 0) \
+    M(UInt64, streaming_join_key_domain_pushdown_max_rows, 10000, "Maximum right-side rows and distinct keys to snapshot for streaming join key-domain pushdown. If this cap or a stricter user read cap is exceeded, the optimization is disabled for the query.", 0) \
+    M(UInt64, streaming_join_key_domain_pushdown_max_bytes, 16777216, "Maximum right-side bytes to snapshot for streaming join key-domain pushdown. If this cap or a stricter user read cap is exceeded, the optimization is disabled for the query. 0 means no feature byte limit.", 0) \
+    M(UInt64, streaming_join_key_domain_pushdown_disjunctive_filter_max_rows, 256, "Maximum captured keys for rendering the streaming join key-domain historical filter as an OR of key equalities. This form is more index-friendly for small key domains. 0 disables it.", 0) \
     M(Int64, max_join_range, 300, "Max join range", 0) \
     M(UInt64, keep_range_join_max_buckets, 0, "Max number of range buckets to keep per stream in a range join. When the number of buckets exceeds this limit, the oldest buckets are reclaimed even if the watermark has not advanced. 0 means no limit.", 0) \
     M(Bool, compact_kv_stream, true, "Control if compact a changelog kv or versioned kv stream during query", 0) \