Merge #148306

148306: sql: lift lookup join parallelization into execbuilder r=yuzefovich a=yuzefovich

In preparation for expanding the heuristic for when it's safe to parallelize cross-range reads in the lookup join, this commit lifts the decision into the execbuilder. As a reminder, when Streamer API is not used, we currently choose to enable parallelism when an input row can result in at most one lookup row (which is the case for index joins as well as lookup joins when equality columns form a key), also when the session variable enables it for multi-lookup case. Previously, this was determined when creating the JoinReader, and now we make this determination in the execbuilder. We add a new field to the processor spec based on the execbuilder argument (for compatibility with 25.2 release we also keep the old logic for now).

Additionally, this commit enhances the EXPLAIN output to show `parallel` attribute in VERBOSE mode when applicable (when the streamer is used as well as when Parallelize is set to true by the execbuilder). This improvement was part of the reason to lifting the determination up there (so that `parallelize` boolean argument to ConstructLookupJoin is captured by the explain factory).

Informs: #134351.
Epic: None

Release note: None

Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: craig[bot]

pkg/ccl/logictestccl/testdata/logic_test/partitioning_hash_sharded_index_query_plan

@@ -72,6 +72,7 @@ vectorized: true
                 │ table: t_parent@t_parent_pkey
                 │ equality cols are key
                 │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_id_shard_16_eq = crdb_internal_id_shard_16)) AND (pid = id)
+                │ parallel
                 │
                 └── • render
                     │ columns: (crdb_internal_id_shard_16_eq, pid)
@@ -150,6 +151,7 @@ vectorized: true
                 │ table: t_parent@t_parent_pkey
                 │ equality cols are key
                 │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_id_shard_16_eq = crdb_internal_id_shard_16)) AND (pid = id)
+                │ parallel
                 │
                 └── • render
                     │ columns: (crdb_internal_id_shard_16_eq, pid)
@@ -244,6 +246,7 @@ vectorized: true
             │ table: t_gen_random_uuid@t_gen_random_uuid_pkey
             │ equality cols are key
             │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_user_id_shard_16_eq = crdb_internal_user_id_shard_16)) AND (user_id_default = user_id)
+            │ parallel
             │
             └── • render
                 │ columns: (crdb_internal_user_id_shard_16_eq, crdb_internal_user_id_shard_16_comp, column2, val_cast, user_id_default)
@@ -297,6 +300,7 @@ vectorized: true
                 │ table: t_gen_random_uuid@t_gen_random_uuid_pkey
                 │ equality cols are key
                 │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_user_id_shard_16_eq = crdb_internal_user_id_shard_16)) AND (user_id_default = user_id)
+                │ parallel
                 │
                 └── • render
                     │ columns: (crdb_internal_user_id_shard_16_eq, crdb_internal_user_id_shard_16_comp, column2, val_cast, user_id_default)
@@ -420,6 +424,7 @@ vectorized: true
             │ table: t_unique_hash_pk@t_unique_hash_pk_pkey
             │ equality cols are key
             │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_id_shard_16_eq = crdb_internal_id_shard_16)) AND (column1 = id)
+            │ parallel
             │
             └── • render
                 │ columns: (crdb_internal_id_shard_16_eq, crdb_internal_id_shard_16_comp, column1, column2)
@@ -505,6 +510,7 @@ vectorized: true
             │ table: t_unique_hash_pk@t_unique_hash_pk_pkey
             │ equality cols are key
             │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_id_shard_16_eq = crdb_internal_id_shard_16)) AND (column1 = id)
+            │ parallel
             │
             └── • render
                 │ columns: (crdb_internal_id_shard_16_eq, crdb_internal_id_shard_16_comp, column1, column2)
@@ -671,6 +677,7 @@ vectorized: true
 │                           │ equality cols are key
 │                           │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_id_shard_16_eq = crdb_internal_id_shard_16)) AND (column1 = id)
 │                           │ locking strength: for update
+│                           │ parallel
 │                           │
 │                           └── • render
 │                               │ columns: (crdb_internal_id_shard_16_eq, crdb_internal_id_shard_16_comp, column1, column2)
@@ -803,6 +810,7 @@ vectorized: true
                 │ equality cols are key
                 │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_id_shard_16_eq = crdb_internal_id_shard_16)) AND (column1 = id)
                 │ locking strength: for update
+                │ parallel
                 │
                 └── • render
                     │ columns: (crdb_internal_id_shard_16_eq, crdb_internal_id_shard_16_comp, column1, column2)
@@ -956,6 +964,7 @@ vectorized: true
             │ table: t_unique_hash_sec_key@idx_uniq_hash_email
             │ equality cols are key
             │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_email_shard_16_eq = crdb_internal_email_shard_16)) AND (column2 = email)
+            │ parallel
             │
             └── • render
                 │ columns: (crdb_internal_email_shard_16_eq, column1, column2, column3, crdb_internal_email_shard_16_comp)
@@ -1018,6 +1027,7 @@ vectorized: true
             │ table: t_unique_hash_sec_key@idx_uniq_hash_email
             │ equality cols are key
             │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_email_shard_16_eq = crdb_internal_email_shard_16)) AND (column2 = email)
+            │ parallel
             │
             └── • render
                 │ columns: (crdb_internal_email_shard_16_eq, column1, column2, column3, crdb_internal_email_shard_16_comp)
@@ -1033,6 +1043,7 @@ vectorized: true
                     │ table: t_unique_hash_sec_key@t_unique_hash_sec_key_pkey
                     │ equality cols are key
                     │ lookup condition: (part IN ('new york', 'seattle')) AND (column1 = id)
+                    │ parallel
                     │
                     └── • render
                         │ columns: (crdb_internal_email_shard_16_comp, column1, column2, column3)
@@ -1158,6 +1169,7 @@ vectorized: true
 │                   │ table: t_unique_hash_sec_key@idx_uniq_hash_email
 │                   │ equality cols are key
 │                   │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_email_shard_16_eq = crdb_internal_email_shard_16)) AND (column2 = email)
+│                   │ parallel
 │                   │
 │                   └── • render
 │                       │ columns: (crdb_internal_email_shard_16_eq, crdb_internal_email_shard_16_comp, column1, column2, column3)
@@ -1414,6 +1426,7 @@ vectorized: true
 │                                   │ equality cols are key
 │                                   │ lookup condition: ((part IN ('new york', 'seattle')) AND (crdb_internal_email_shard_16_eq = crdb_internal_email_shard_16)) AND (column2 = email)
 │                                   │ locking strength: for update
+│                                   │ parallel
 │                                   │
 │                                   └── • render
 │                                       │ columns: (crdb_internal_email_shard_16_eq, crdb_internal_email_shard_16_comp, column1, column2, column3)
@@ -1638,6 +1651,7 @@ vectorized: true
 │                       │ equality cols are key
 │                       │ lookup condition: (part IN ('new york', 'seattle')) AND (column1 = id)
 │                       │ locking strength: for update
+│                       │ parallel
 │                       │
 │                       └── • render
 │                           │ columns: (crdb_internal_email_shard_16_comp, column1, column2, column3)