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,16 @@
     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") \
@@ -358,7 +368,7 @@
       "due to overrun.") \
     M(QueryProfilerRuns, "Number of times QueryProfiler had been run.") \
     M(QueryProfilerErrors, "Invalid memory accesses during asynchronous stack unwinding.") \
-    \
+\
     M(CreatedLogEntryForMerge, "Successfully created log entry to merge parts in ReplicatedMergeTree.") \
     M(NotCreatedLogEntryForMerge, \
       "Log entry to merge parts in ReplicatedMergeTree is not created due to concurrent log update by another replica.") \
@@ -420,10 +430,10 @@
     M(DiskS3CompleteMultipartUpload, "Number of DiskS3 API CompleteMultipartUpload calls.") \
     M(DiskS3PutObject, "Number of DiskS3 API PutObject calls.") \
     M(DiskS3GetObject, "Number of DiskS3 API GetObject calls.") \
-    \
+\
     M(S3Clients, "Number of created S3 clients.") \
     M(TinyS3Clients, "Number of S3 clients copies which reuse an existing auth provider from another client.") \
-    \
+\
     M(ReadBufferFromS3Microseconds, "Time spent on reading from S3.") \
     M(ReadBufferFromS3InitMicroseconds, "Time spent initializing connection to S3.") \
     M(ReadBufferFromS3Bytes, "Bytes read from S3.") \
@@ -537,13 +547,14 @@
     M(IOUringCQEsFailed, "Total number of completed io_uring CQEs with failures") \
 \
     M(ParallelReplicasAvailableCount, "Number of replicas available to execute a query with task-based parallel replicas") \
-    M(ParallelReplicasUnavailableCount, "Number of replicas which was chosen, but found to be unavailable during query execution with task-based parallel replicas") \
-    \
+    M(ParallelReplicasUnavailableCount, \
+      "Number of replicas which was chosen, but found to be unavailable during query execution with task-based parallel replicas") \
+\
     M(S3CachedCredentialsProvidersReused, "Total number of reused credentials provider from the cache") \
     M(S3CachedCredentialsProvidersAdded, "Total number of newly added credentials providers to the cache") \
-    \
+\
     M(PythonGILAcquired, "Number of times the Python GIL was acquired") \
-    M(PythonGILWaitMicroseconds, "Total time spent waiting to acquire the Python GIL in microseconds") \
+    M(PythonGILWaitMicroseconds, "Total time spent waiting to acquire the Python GIL in microseconds")
 
 
 namespace ProfileEvents

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) \