feat: subquery to join on not in

src/query/service/tests/it/pipelines/filter/random_filter_expr.rs

@@ -147,6 +147,7 @@ fn convert_predicate_tree_to_scalar_expr(node: PredicateNode, data_type: &DataTy
                 table_name: None,
                 column_position: None,
                 table_index: None,
+                source_table_index: None,
                 column_name: "".to_string(),
                 index: 0,
                 data_type: Box::new(data_type.clone()),

src/query/sql/src/planner/binder/bind_table_reference/bind_table.rs

@@ -164,7 +164,7 @@ impl Binder {
         {
             let change_type = get_change_type(&table_name_alias);
             if change_type.is_some() {
-                let table_index = self.metadata.write().add_table(
+                let (table_index, source_table_index) = self.metadata.write().add_table(
                     catalog,
                     database.clone(),
                     table_meta,
@@ -178,6 +178,7 @@ impl Binder {
                     bind_context,
                     database.as_str(),
                     table_index,
+                    source_table_index,
                     change_type,
                     sample,
                 )?;
@@ -281,7 +282,7 @@ impl Binder {
                 }
             }
             _ => {
-                let table_index = self.metadata.write().add_table(
+                let (table_index, source_table_index) = self.metadata.write().add_table(
                     catalog.clone(),
                     database.clone(),
                     table_meta.clone(),
@@ -296,6 +297,7 @@ impl Binder {
                     bind_context,
                     database.as_str(),
                     table_index,
+                    source_table_index,
                     None,
                     sample,
                 )?;

src/query/sql/src/planner/binder/bind_table_reference/bind_table_function.rs

@@ -143,7 +143,7 @@ impl Binder {
             } else {
                 None
             };
-            let table_index = self.metadata.write().add_table(
+            let (table_index, source_table_index) = self.metadata.write().add_table(
                 CATALOG_DEFAULT.to_string(),
                 "system".to_string(),
                 table.clone(),
@@ -154,8 +154,14 @@ impl Binder {
                 None,
             );
 
-            let (s_expr, mut bind_context) =
-                self.bind_base_table(bind_context, "system", table_index, None, sample)?;
+            let (s_expr, mut bind_context) = self.bind_base_table(
+                bind_context,
+                "system",
+                table_index,
+                source_table_index,
+                None,
+                sample,
+            )?;
             if let Some(alias) = alias {
                 bind_context.apply_table_alias(alias, &self.name_resolution_ctx)?;
             }
@@ -205,7 +211,7 @@ impl Binder {
                 None
             };
 
-            let table_index = self.metadata.write().add_table(
+            let (table_index, source_table_index) = self.metadata.write().add_table(
                 CATALOG_DEFAULT.to_string(),
                 "system".to_string(),
                 table.clone(),
@@ -216,8 +222,14 @@ impl Binder {
                 None,
             );
 
-            let (s_expr, mut bind_context) =
-                self.bind_base_table(bind_context, "system", table_index, None, &None)?;
+            let (s_expr, mut bind_context) = self.bind_base_table(
+                bind_context,
+                "system",
+                table_index,
+                source_table_index,
+                None,
+                &None,
+            )?;
             if let Some(alias) = alias {
                 bind_context.apply_table_alias(alias, &self.name_resolution_ctx)?;
             }

src/query/sql/src/planner/binder/column_binding.rs

@@ -28,6 +28,8 @@ pub struct ColumnBinding {
     pub column_position: Option<usize>,
     /// Table index of this `ColumnBinding` in current context
     pub table_index: Option<IndexType>,
+    /// Source table index of this `ColumnBinding` in current context
+    pub source_table_index: Option<IndexType>,
     /// Column name of this `ColumnBinding` in current context
     pub column_name: String,
     /// Column index of ColumnBinding
@@ -72,6 +74,7 @@ impl ColumnBinding {
             table_name: None,
             column_position: None,
             table_index: None,
+            source_table_index: None,
             column_name: name,
             index,
             data_type,
@@ -83,6 +86,25 @@ impl ColumnBinding {
     pub fn is_dummy(&self) -> bool {
         self.index >= DummyColumnType::Other.type_identifier()
     }
+
+    // Only table_index and column_position are retained to determine whether two columns are the same column of the same table.
+    // Avoid situations where aliases and other situations may cause inability to judge
+    pub fn as_source(&self) -> Option<ColumnBinding> {
+        self.source_table_index
+            .or(self.table_index)
+            .map(|table_index| ColumnBinding {
+                database_name: None,
+                table_name: None,
+                column_position: self.column_position,
+                table_index: Some(table_index),
+                source_table_index: None,
+                column_name: "".to_string(),
+                index: 0,
+                data_type: self.data_type.clone(),
+                visibility: self.visibility.clone(),
+                virtual_expr: None,
+            })
+    }
 }
 
 impl ColumnIndex for ColumnBinding {}
@@ -96,6 +118,8 @@ pub struct ColumnBindingBuilder {
     pub column_position: Option<usize>,
     /// Table index of this `ColumnBinding` in current context
     pub table_index: Option<IndexType>,
+    /// Source table index of this `ColumnBinding` in current context
+    pub source_table_index: Option<IndexType>,
     /// Column name of this `ColumnBinding` in current context
     pub column_name: String,
     /// Column index of ColumnBinding
@@ -120,6 +144,7 @@ impl ColumnBindingBuilder {
             table_name: None,
             column_position: None,
             table_index: None,
+            source_table_index: None,
             column_name,
             index,
             data_type,
@@ -148,6 +173,11 @@ impl ColumnBindingBuilder {
         self
     }
 
+    pub fn source_table_index(mut self, index: Option<IndexType>) -> ColumnBindingBuilder {
+        self.source_table_index = index;
+        self
+    }
+
     pub fn virtual_expr(mut self, virtual_expr: Option<String>) -> ColumnBindingBuilder {
         self.virtual_expr = virtual_expr;
         self
@@ -159,6 +189,7 @@ impl ColumnBindingBuilder {
             table_name: self.table_name,
             column_position: self.column_position,
             table_index: self.table_index,
+            source_table_index: self.source_table_index,
             column_name: self.column_name,
             index: self.index,
             data_type: self.data_type,

src/query/sql/src/planner/binder/table.rs

@@ -132,7 +132,7 @@ impl Binder {
             None
         };
 
-        let table_index = self.metadata.write().add_table(
+        let (table_index, source_table_index) = self.metadata.write().add_table(
             CATALOG_DEFAULT.to_string(),
             "system".to_string(),
             table.clone(),
@@ -143,8 +143,14 @@ impl Binder {
             None,
         );
 
-        let (s_expr, mut bind_context) =
-            self.bind_base_table(bind_context, "system", table_index, None, &None)?;
+        let (s_expr, mut bind_context) = self.bind_base_table(
+            bind_context,
+            "system",
+            table_index,
+            source_table_index,
+            None,
+            &None,
+        )?;
         if let Some(alias) = alias {
             bind_context.apply_table_alias(alias, &self.name_resolution_ctx)?;
         }
@@ -353,6 +359,7 @@ impl Binder {
         bind_context: &BindContext,
         database_name: &str,
         table_index: IndexType,
+        source_table_index: Option<IndexType>,
         change_type: Option<ChangeType>,
         sample: &Option<SampleConfig>,
     ) -> Result<(SExpr, BindContext)> {
@@ -388,6 +395,7 @@ impl Binder {
                     .table_name(Some(table_name.to_string()))
                     .database_name(Some(database_name.to_string()))
                     .table_index(Some(*table_index))
+                    .source_table_index(source_table_index)
                     .column_position(*column_position)
                     .virtual_expr(virtual_expr.clone())
                     .build();

src/query/sql/src/planner/dataframe.rs

@@ -93,7 +93,7 @@ impl Dataframe {
                 query_ctx.clone().get_abort_checker(),
             )?;
 
-            let table_index = metadata.write().add_table(
+            let (table_index, source_table_index) = metadata.write().add_table(
                 CATALOG_DEFAULT.to_owned(),
                 database.to_string(),
                 table_meta,
@@ -104,7 +104,14 @@ impl Dataframe {
                 None,
             );
 
-            binder.bind_base_table(&bind_context, database, table_index, None, &None)
+            binder.bind_base_table(
+                &bind_context,
+                database,
+                table_index,
+                source_table_index,
+                None,
+                &None,
+            )
         } else {
             binder.bind_table_reference(&mut bind_context, &table)
         }?;

src/query/sql/src/planner/expression_parser.rs

@@ -60,7 +60,7 @@ use crate::Visibility;
 pub fn bind_table(table_meta: Arc<dyn Table>) -> Result<(BindContext, MetadataRef)> {
     let mut bind_context = BindContext::new();
     let metadata = Arc::new(RwLock::new(Metadata::default()));
-    let table_index = metadata.write().add_table(
+    let (table_index, _) = metadata.write().add_table(
         CATALOG_DEFAULT.to_owned(),
         "default".to_string(),
         table_meta,

src/query/sql/src/planner/metadata.rs

@@ -123,6 +123,25 @@ impl Metadata {
             .map(|table| table.index)
     }
 
+    fn get_source_table_index(
+        &self,
+        database_name: Option<&str>,
+        table_name: &str,
+    ) -> Option<IndexType> {
+        self.tables
+            .iter()
+            .find(|table| match database_name {
+                Some(database_name) => {
+                    table.database == database_name && table.name == table_name
+                        || table.alias_name == Some(table_name.to_string())
+                }
+                None => {
+                    table.name == table_name || table.alias_name == Some(table_name.to_string())
+                }
+            })
+            .map(|table| table.index)
+    }
+
     pub fn column(&self, index: IndexType) -> &ColumnEntry {
         self.columns
             .get(index)
@@ -358,9 +377,11 @@ impl Metadata {
         source_of_index: bool,
         source_of_stage: bool,
         cte_suffix_name: Option<String>,
-    ) -> IndexType {
+    ) -> (IndexType, Option<IndexType>) {
         let table_name = table_meta.name().to_string();
         let table_name = Self::remove_cte_suffix(table_name, cte_suffix_name);
+        let source_table_index =
+            self.get_source_table_index(Some(database.as_str()), table_name.as_str());
 
         let table_index = self.tables.len();
         // If exists table alias name, use it instead of origin name
@@ -455,7 +476,7 @@ impl Metadata {
             }
         }
 
-        table_index
+        (table_index, source_table_index)
     }
 
     pub fn change_derived_column_alias(&mut self, index: IndexType, alias: String) {

src/query/sql/src/planner/optimizer/aggregate/normalize_aggregate.rs

@@ -143,6 +143,7 @@ impl RuleNormalizeAggregateOptimizer {
                             column: ColumnBinding {
                                 table_name: None,
                                 table_index: None,
+                                source_table_index: None,
                                 database_name: None,
                                 column_position: None,
                                 index: work_index,

src/query/sql/src/planner/optimizer/decorrelate/subquery_rewriter.rs

@@ -601,20 +601,20 @@ impl SubqueryRewriter {
             SubqueryType::Any => {
                 let output_column = subquery.output_column.clone();
                 let column_name = format!("subquery_{}", output_column.index);
-                let left_condition = wrap_cast(
-                    &ScalarExpr::BoundColumnRef(BoundColumnRef {
-                        span: subquery.span,
-                        column: ColumnBindingBuilder::new(
-                            column_name,
-                            output_column.index,
-                            output_column.data_type,
-                            Visibility::Visible,
-                        )
-                        .table_index(output_column.table_index)
-                        .build(),
-                    }),
-                    &subquery.data_type,
-                );
+                let mut left_condition = ScalarExpr::BoundColumnRef(BoundColumnRef {
+                    span: subquery.span,
+                    column: ColumnBindingBuilder::new(
+                        column_name,
+                        output_column.index,
+                        output_column.data_type,
+                        Visibility::Visible,
+                    )
+                    .table_index(output_column.table_index)
+                    .build(),
+                });
+                if !left_condition.data_type()?.eq(&subquery.data_type) {
+                    left_condition = wrap_cast(&left_condition, &subquery.data_type)
+                }
                 let child_expr = *subquery.child_expr.as_ref().unwrap().clone();
                 let op = *subquery.compare_op.as_ref().unwrap();
                 let (right_condition, is_non_equi_condition) =

src/query/sql/src/planner/optimizer/rule/factory.rs

@@ -46,6 +46,7 @@ use super::rewrite::RulePushDownSortEvalScalar;
 use super::rewrite::RulePushDownSortScan;
 use super::rewrite::RuleSemiToInnerJoin;
 use super::rewrite::RuleSplitAggregate;
+use super::rewrite::RuleSubqueryNotInToIn;
 use super::rewrite::RuleTryApplyAggIndex;
 use super::transform::RuleCommuteJoinBaseTable;
 use super::transform::RuleEagerAggregation;
@@ -111,6 +112,7 @@ impl RuleFactory {
             RuleID::MergeFilterIntoMutation => {
                 Ok(Box::new(RuleMergeFilterIntoMutation::new(ctx.metadata)))
             }
+            RuleID::SubqueryNotInToIn => Ok(Box::new(RuleSubqueryNotInToIn::new(ctx.metadata))),
         }
     }
 }

src/query/sql/src/planner/optimizer/rule/rewrite/mod.rs

@@ -47,6 +47,7 @@ mod rule_push_down_sort_expression;
 mod rule_push_down_sort_scan;
 mod rule_semi_to_inner_join;
 mod rule_split_aggregate;
+mod rule_subquery_not_in_to_in;
 mod rule_try_apply_agg_index;
 
 pub use rule_commute_join::RuleCommuteJoin;
@@ -83,4 +84,5 @@ pub use rule_push_down_sort_expression::RulePushDownSortEvalScalar;
 pub use rule_push_down_sort_scan::RulePushDownSortScan;
 pub use rule_semi_to_inner_join::RuleSemiToInnerJoin;
 pub use rule_split_aggregate::RuleSplitAggregate;
+pub use rule_subquery_not_in_to_in::RuleSubqueryNotInToIn;
 pub use rule_try_apply_agg_index::RuleTryApplyAggIndex;