fix(optimizer): fix distributed batch lookup join shuffle columns (ri…

…singwavelabs#7713)

- As title.

Approved-By: BugenZhao

src/frontend/planner_test/tests/testdata/batch_index_join.yaml

@@ -66,3 +66,13 @@
           └─BatchLookupJoin { type: Inner, predicate: t.a = t2.c, output: [t2.c, t2.d, t.a] }
             └─BatchExchange { order: [], dist: UpstreamHashShard(t.a) }
               └─BatchScan { table: t, columns: [t.a], distribution: SomeShard }
+- sql: |
+    create table t1 (a int, b int);
+    create table t2 (c int, d int);
+    create index idx on t2(c) include (d);
+    select * from t1 join idx on t1.a = idx.c and t1.b = idx.d;
+  batch_plan: |
+    BatchExchange { order: [], dist: Single }
+    └─BatchLookupJoin { type: Inner, predicate: t1.a = idx.c AND t1.b = idx.d, output: all }
+      └─BatchExchange { order: [], dist: UpstreamHashShard(t1.a) }
+        └─BatchScan { table: t1, columns: [t1.a, t1.b], distribution: SomeShard }

src/frontend/planner_test/tests/testdata/expr.yaml

@@ -536,7 +536,6 @@
   logical_plan: |
     LogicalProject { exprs: [Array(1:Int32, null:Int32)] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
-
 - name: regression (#7641) -fuzzing test failed at Bind error,types Boolean and Varchar cannot be matched
   sql: |
     select false >= 'LN1O0QP1yi' NOT IN (md5('4SeUPZhUbH'))

src/frontend/src/optimizer/plan_node/batch_lookup_join.rs

@@ -178,7 +178,11 @@ impl ToDistributedBatch for BatchLookupJoin {
         let input = self.input().to_distributed_with_required(
             &Order::any(),
             &RequiredDist::PhysicalDist(Distribution::UpstreamHashShard(
-                self.eq_join_predicate.left_eq_indexes(),
+                self.eq_join_predicate
+                    .left_eq_indexes()
+                    .into_iter()
+                    .take(self.lookup_prefix_len)
+                    .collect(),
                 self.right_table_desc.table_id,
             )),
         )?;