From 23217651287338b8e7ff0a75ed237a2708841122 Mon Sep 17 00:00:00 2001
From: Dylan Chen <zilin@singularity-data.com>
Date: Mon, 30 Oct 2023 14:08:53 +0800
Subject: [PATCH] use ColIndexMapping new instead of without_target_size

---
 .../rule/apply_join_transpose_rule.rs         | 27 ++++++++++++-------
 .../optimizer/rule/apply_offset_rewriter.rs   |  7 ++++-
 2 files changed, 24 insertions(+), 10 deletions(-)

diff --git a/src/frontend/src/optimizer/rule/apply_join_transpose_rule.rs b/src/frontend/src/optimizer/rule/apply_join_transpose_rule.rs
index 7ac121692c81d..98363f2aba213 100644
--- a/src/frontend/src/optimizer/rule/apply_join_transpose_rule.rs
+++ b/src/frontend/src/optimizer/rule/apply_join_transpose_rule.rs
@@ -356,7 +356,7 @@ impl ApplyJoinTransposeRule {
             correlated_indices,
             false,
         );
-        let output_indices: Vec<_> = {
+        let (output_indices, target_size) = {
             let (apply_left_len, join_right_len) = match apply_join_type {
                 JoinType::LeftSemi | JoinType::LeftAnti => (apply_left_len, 0),
                 JoinType::RightSemi | JoinType::RightAnti => (0, join.right().schema().len()),
@@ -368,14 +368,19 @@ impl ApplyJoinTransposeRule {
                 join_left_len + apply_left_len..join_left_len + apply_left_len + join_right_len,
             );
 
-            match join.join_type() {
+            let output_indices: Vec<_> = match join.join_type() {
                 JoinType::LeftSemi | JoinType::LeftAnti => left_iter.collect(),
                 JoinType::RightSemi | JoinType::RightAnti => right_iter.collect(),
                 _ => left_iter.chain(right_iter).collect(),
-            }
+            };
+
+            let target_size = join_left_len + apply_left_len + join_right_len;
+            (output_indices, target_size)
         };
-        let mut output_indices_mapping =
-            ColIndexMapping::without_target_size(output_indices.iter().map(|x| Some(*x)).collect());
+        let mut output_indices_mapping = ColIndexMapping::new(
+            output_indices.iter().map(|x| Some(*x)).collect(),
+            target_size,
+        );
         let new_join = LogicalJoin::new(
             join.left(),
             new_join_right,
@@ -518,7 +523,7 @@ impl ApplyJoinTransposeRule {
             false,
         );
 
-        let output_indices: Vec<_> = {
+        let (output_indices, target_size) = {
             let (apply_left_len, join_right_len) = match apply_join_type {
                 JoinType::LeftSemi | JoinType::LeftAnti => (apply_left_len, 0),
                 JoinType::RightSemi | JoinType::RightAnti => (0, join.right().schema().len()),
@@ -529,11 +534,14 @@ impl ApplyJoinTransposeRule {
             let right_iter = join_left_len + apply_left_len * 2
                 ..join_left_len + apply_left_len * 2 + join_right_len;
 
-            match join.join_type() {
+            let output_indices: Vec<_> = match join.join_type() {
                 JoinType::LeftSemi | JoinType::LeftAnti => left_iter.collect(),
                 JoinType::RightSemi | JoinType::RightAnti => right_iter.collect(),
                 _ => left_iter.chain(right_iter).collect(),
-            }
+            };
+
+            let target_size = join_left_len + apply_left_len * 2 + join_right_len;
+            (output_indices, target_size)
         };
         let new_join = LogicalJoin::new(
             new_join_left,
@@ -548,8 +556,9 @@ impl ApplyJoinTransposeRule {
                 new_join.into()
             }
             JoinType::Inner | JoinType::LeftOuter | JoinType::RightOuter | JoinType::FullOuter => {
-                let mut output_indices_mapping = ColIndexMapping::without_target_size(
+                let mut output_indices_mapping = ColIndexMapping::new(
                     output_indices.iter().map(|x| Some(*x)).collect(),
+                    target_size,
                 );
                 // Leave other condition for predicate push down to deal with
                 LogicalFilter::create(
diff --git a/src/frontend/src/optimizer/rule/apply_offset_rewriter.rs b/src/frontend/src/optimizer/rule/apply_offset_rewriter.rs
index cba7b174433dc..559ace9d5dbe0 100644
--- a/src/frontend/src/optimizer/rule/apply_offset_rewriter.rs
+++ b/src/frontend/src/optimizer/rule/apply_offset_rewriter.rs
@@ -74,8 +74,13 @@ pub struct ApplyCorrelatedIndicesConverter {}
 impl ApplyCorrelatedIndicesConverter {
     pub fn convert_to_index_mapping(correlated_indices: &[usize]) -> ColIndexMapping {
         // Inverse anyway.
-        let col_mapping = ColIndexMapping::without_target_size(
+        let target_size = match correlated_indices.iter().max_by_key(|&&x| x) {
+            Some(target_max) => target_max + 1,
+            None => 0,
+        };
+        let col_mapping = ColIndexMapping::new(
             correlated_indices.iter().copied().map(Some).collect_vec(),
+            target_size,
         );
         let mut map = vec![None; col_mapping.target_size()];
         for (src, dst) in col_mapping.mapping_pairs() {