fix: cross join return wrong column (#18355)

* fix: cross join return wrong column

* add test

Author: SkyFan2002

src/query/service/src/pipelines/builders/builder_join.rs

@@ -139,6 +139,7 @@ impl PipelineBuilder {
             self.func_ctx.clone(),
             state.clone(),
             &join.probe_projections,
+            &join.build_projections,
             &join.probe_keys,
             join.probe.output_schema()?,
             &join.join_type,

src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_probe_state.rs

@@ -81,6 +81,8 @@ pub struct HashJoinProbeState {
     /// `probe_projections` only contains the columns from upstream required columns
     /// and columns from other_condition which are in probe schema.
     pub(crate) probe_projections: ColumnSet,
+    // used in cross join
+    pub(crate) build_projections: ColumnSet,
     /// Todo(xudong): add more detailed comments for the following fields.
     /// Final scan tasks
     pub(crate) final_scan_tasks: RwLock<VecDeque<usize>>,
@@ -99,6 +101,7 @@ impl HashJoinProbeState {
         func_ctx: FunctionContext,
         hash_join_state: Arc<HashJoinState>,
         probe_projections: &ColumnSet,
+        build_projections: &ColumnSet,
         probe_keys: &[RemoteExpr],
         mut probe_schema: DataSchemaRef,
         join_type: &JoinType,
@@ -138,6 +141,7 @@ impl HashJoinProbeState {
             merge_into_final_partial_unmodified_scan_tasks: RwLock::new(VecDeque::new()),
             mark_scan_map_lock: Mutex::new(()),
             hash_method: method,
+            build_projections: build_projections.clone(),
         })
     }
 

src/query/service/src/pipelines/processors/transforms/hash_join/probe_join/cross_join.rs

@@ -35,6 +35,7 @@ impl HashJoinProbeState {
         }
         let mut probe_block = input.project(&self.probe_projections);
         let build_block = DataBlock::concat(build_blocks)?;
+        let build_block = build_block.project(&self.build_projections);
         if build_num_rows == 1 {
             for col in build_block.columns() {
                 let scalar = unsafe { col.index_unchecked(0) };

tests/sqllogictests/suites/query/join/cross.test

@@ -20,4 +20,171 @@ statement ok
 drop table onecolumn
 
 statement ok
-drop table empty
+drop table empty
+
+statement ok
+CREATE TABLE cte_test1 (
+  "Price Branch Region" VARCHAR NULL,
+  "PriceBranchID" VARCHAR NULL,
+  "ProductType" VARCHAR NULL,
+  "Product Type" VARCHAR NULL,
+  "BILL TO_CustomerID" VARCHAR NULL,
+  "BILL TO_CustomerName" VARCHAR NULL,
+  "CustomerType" VARCHAR NULL,
+  "MarketCategory" VARCHAR NULL,
+  "OrderChannel" VARCHAR NULL,
+  "Price Override" VARCHAR NULL,
+  "Value" DECIMAL(38, 10) NULL
+);
+
+statement ok
+INSERT INTO cte_test1 VALUES (
+    'Price Branch Region'
+    , 'PriceBranchID'
+    , 'ProductType'
+    , 'Product Type'
+    , 'BILL TO_CustomerID'
+    , 'BILL TO_CustomerName'
+    , 'CustomerType'
+    , 'MarketCategory'
+    , 'OrderChannel'
+    , 'Price Override'
+    , 10
+);
+
+statement ok
+CREATE TABLE cte_test2 (
+  "Year" VARCHAR NULL,
+  "PriceBranchGlRegionID" VARCHAR NULL,
+  "Price Branch Region" VARCHAR NULL,
+  "PriceBranchID" VARCHAR NULL,
+  "Enterprise Profit" VARCHAR NULL,
+  "Tier Group" VARCHAR NULL,
+  "Tier Group 2" VARCHAR NULL,
+  "Tier Group Sort" VARCHAR NULL,
+  "Tier" VARCHAR NULL,
+  "Tier Name" VARCHAR NULL,
+  "Tier Label" VARCHAR NULL,
+  "Short Name" VARCHAR NULL,
+  "Economic Tree" VARCHAR NULL,
+  "Tier Group v2" VARCHAR NULL,
+  "Tier Name v2" VARCHAR NULL,
+  "Tier Sort v2" VARCHAR NULL,
+  "Tier Group 2 v2" VARCHAR NULL,
+  "Tier Group v3" VARCHAR NULL,
+  "Tier Name v3" VARCHAR NULL,
+  "Short Name v3" VARCHAR NULL,
+  "Tier Sort v3" VARCHAR NULL,
+  "ProductType" VARCHAR NULL,
+  "Product Type" VARCHAR NULL,
+  "BILL TO_CustomerID" VARCHAR NULL,
+  "BILL TO_CustomerName" VARCHAR NULL,
+  "CustomerType" VARCHAR NULL,
+  "MarketCategory" VARCHAR NULL,
+  "OrderChannel" VARCHAR NULL,
+  "Price Override" VARCHAR NULL,
+  "Value" DECIMAL(38, 10) NULL,
+  "BuyLineID" VARCHAR NULL
+);
+
+statement ok
+INSERT INTO cte_test2 VALUES (
+  'Year',
+  'PriceBranchGlRegionID',
+  'Price Branch Region',
+  'PriceBranchID',
+  'Enterprise Profit',
+  'Tier Group',
+  'Tier Group 2',
+  'Tier Group Sort',
+  'Tier',
+  'Tier Name',
+  'Tier Label',
+  'Short Name',
+  'Economic Tree',
+  'Tier Group v2',
+  'Tier Name v2',
+  'Tier Sort v2',
+  'Tier Group 2 v2',
+  'Tier Group v3',
+  'Tier Name v3',
+  'Short Name v3',
+  'Tier Sort v3',
+  'ProductType',
+  'Product Type',
+  'BILL TO_CustomerID',
+  'BILL TO_CustomerName',
+  'CustomerType',
+  'MarketCategory',
+  'OrderChannel',
+  'Price Override',
+  20,
+  'BuyLineID'
+);
+
+query T
+WITH GrossRev AS
+(
+  SELECT
+      "ProductType"
+    , "Product Type"
+    , "BILL TO_CustomerID"
+    , "BILL TO_CustomerName"
+    , "CustomerType"
+    , "OrderChannel"
+    , "MarketCategory"
+    , "Price Branch Region"
+    , "PriceBranchID"
+    , "Price Override"
+    , SUM(0) "Value"
+    , SUM("Value") "GrossRevenue"
+  FROM cte_test1
+  GROUP BY
+      "ProductType"
+    , "Product Type"
+    , "BILL TO_CustomerID"
+    , "BILL TO_CustomerName"
+    , "CustomerType"
+    , "OrderChannel"
+    , "MarketCategory"
+    , "Price Branch Region"
+    , "PriceBranchID"
+    , "Price Override"
+)
+, ShortName AS
+(
+  SELECT
+      "Tier Group v3"
+    , "Short Name v3"
+    , "Tier Sort v3"
+  FROM cte_test2
+  GROUP BY
+      "Tier Group v3"
+    , "Short Name v3"
+    , "Tier Sort v3"
+)
+, res AS
+(
+SELECT
+      'GrossRevData' "DataSet"
+    , sn."Tier Group v3"
+    , sn."Short Name v3"
+    , sn."Tier Sort v3"
+    , gr."ProductType"
+    , gr."Product Type"
+    , gr."BILL TO_CustomerID"
+    , gr."BILL TO_CustomerName"
+    , gr."CustomerType"
+    , gr."OrderChannel"
+    , gr."MarketCategory"
+    , gr."Price Branch Region"
+    , gr."PriceBranchID"
+    , gr."Price Override"
+    , gr."Value"
+    , gr."GrossRevenue"
+FROM ShortName sn
+CROSS JOIN GrossRev gr
+)
+select "BILL TO_CustomerName" from res;
+----
+BILL TO_CustomerName