fix

Author: SkyFan2002

src/query/sql/src/planner/optimizer/optimizers/common_subexpression/analyze.rs

@@ -26,7 +26,7 @@ use crate::plans::MaterializedCTERef;
 use crate::plans::RelOperator;
 pub fn analyze_common_subexpression(
     s_expr: &SExpr,
-    metadata: &Metadata,
+    metadata: &mut Metadata,
 ) -> Result<(Vec<SExprReplacement>, Vec<SExpr>)> {
     // Skip CSE optimization if the expression contains recursive CTE
     if contains_recursive_cte(s_expr) {
@@ -46,7 +46,7 @@ fn process_candidate_expressions(
     candidates: &[(Vec<usize>, SExpr)],
     replacements: &mut Vec<SExprReplacement>,
     materialized_ctes: &mut Vec<SExpr>,
-    _metadata: &Metadata,
+    metadata: &mut Metadata,
 ) -> Result<()> {
     if candidates.len() < 2 {
         return Ok(());
@@ -60,6 +60,7 @@ fn process_candidate_expressions(
         new_scan.push_down_predicates = None;
         new_scan.limit = None;
         new_scan.order_by = None;
+        new_scan.scan_id = metadata.next_scan_id();
         Arc::new(SExpr::create_leaf(Arc::new(RelOperator::Scan(new_scan))))
     } else {
         Arc::new(cte_def.clone())

src/query/sql/src/planner/optimizer/optimizers/common_subexpression/optimizer.rs

@@ -31,8 +31,9 @@ pub struct CommonSubexpressionOptimizer {
 impl Optimizer for CommonSubexpressionOptimizer {
     async fn optimize(&mut self, s_expr: &SExpr) -> Result<SExpr> {
         let metadata = self._opt_ctx.get_metadata();
-        let metadata = metadata.read();
-        let (replacements, materialized_ctes) = analyze_common_subexpression(s_expr, &metadata)?;
+        let mut metadata = metadata.write();
+        let (replacements, materialized_ctes) =
+            analyze_common_subexpression(s_expr, &mut metadata)?;
         rewrite_sexpr(s_expr, replacements, materialized_ctes)
     }
 

src/query/storages/fuse/src/operations/read/parquet_data_source_deserializer.rs

@@ -140,7 +140,7 @@ impl DeserializeDataTransform {
     fn runtime_filter(&mut self, data_block: DataBlock) -> Result<Option<Bitmap>> {
         // Check if already cached runtime filters
         if self.cached_runtime_filter.is_none() {
-            let bloom_filters = self.ctx.get_bloom_runtime_filter_with_id(self.table_index);
+            let bloom_filters = self.ctx.get_bloom_runtime_filter_with_id(self.scan_id);
             let bloom_filters = bloom_filters
                 .into_iter()
                 .filter_map(|filter| {