fix(query): handle single-row inequality joins safely (#20155)

* fix(query): avoid range join for single-row side (#20062)

* fix(query): clear single_to_inner for cross join fallback from range join

When a scalar subquery inequality join (LeftSingle/RightSingle converted
to Inner with single_to_inner marker) has no equi-conditions and falls
back from RangeJoin to HashJoin, the hash join executes as a cross join
+ filter. The FROM_LEFT_SINGLE runtime check fires during the
cross-product matching phase before the filter is applied, causing a
false-positive "Scalar subquery can't return more than one row" error.

Clear single_to_inner when the join has no equi-conditions so the
runtime doesn't enforce it in this context. The scalar guarantee is
already satisfied by construction (the aggregate always produces exactly
one row).

This is a follow-up to #20062 which introduced the single-row guard on
both sides but did not handle the cross-join false-positive.

* fix(query): preserve scalar guard for one-row outers

* test(query): cover scalar probe cross join guard

* test(query): remove duplicate scalar guard coverage

* refactor(query): clarify scalar join cardinality wording

* refactor(query): keep single-row helper concise

* docs(query): clarify scalar-side cardinality comment

* test(query): mask read cost in join explain

---------

Co-authored-by: sundyli <543950155@qq.com>

Author: dantengsky

src/query/service/src/physical_plans/physical_join.rs

@@ -27,6 +27,30 @@ use crate::physical_plans::PhysicalPlanBuilder;
 use crate::physical_plans::explain::PlanStatsInfo;
 use crate::physical_plans::physical_plan::PhysicalPlan;
 
+fn is_single_row(stat_info: &databend_common_sql::optimizer::ir::StatInfo) -> bool {
+    matches!(stat_info.statistics.precise_cardinality, Some(1)) || stat_info.cardinality == 1.0
+}
+
+fn is_precise_single_row(stat_info: &databend_common_sql::optimizer::ir::StatInfo) -> bool {
+    matches!(stat_info.statistics.precise_cardinality, Some(1))
+}
+
+fn single_join_scalar_side_is_precise_single_row(join: &Join, s_expr: &SExpr) -> Result<bool> {
+    match join.single_to_inner {
+        Some(JoinType::LeftSingle) => {
+            let right_rel_expr = RelExpr::with_s_expr(s_expr.right_child());
+            let right_stat_info = right_rel_expr.derive_cardinality()?;
+            Ok(is_precise_single_row(&right_stat_info))
+        }
+        Some(JoinType::RightSingle) => {
+            let left_rel_expr = RelExpr::with_s_expr(s_expr.left_child());
+            let left_stat_info = left_rel_expr.derive_cardinality()?;
+            Ok(is_precise_single_row(&left_stat_info))
+        }
+        _ => Ok(false),
+    }
+}
+
 enum PhysicalJoinType {
     Hash,
     // The first arg is range conditions, the second arg is other conditions
@@ -63,6 +87,7 @@ fn physical_join(join: &Join, s_expr: &SExpr) -> Result<PhysicalJoinType> {
 
     let left_rel_expr = RelExpr::with_s_expr(s_expr.left_child());
     let right_rel_expr = RelExpr::with_s_expr(s_expr.right_child());
+    let left_stat_info = left_rel_expr.derive_cardinality()?;
     let right_stat_info = right_rel_expr.derive_cardinality()?;
 
     if !join.equi_conditions.is_empty() {
@@ -75,10 +100,10 @@ fn physical_join(join: &Join, s_expr: &SExpr) -> Result<PhysicalJoinType> {
         return Ok(PhysicalJoinType::Hash);
     }
 
-    if matches!(right_stat_info.statistics.precise_cardinality, Some(1))
-        || right_stat_info.cardinality == 1.0
-    {
-        // If the output rows of build side is equal to 1, we use CROSS JOIN + FILTER instead of RANGE JOIN.
+    if is_single_row(&left_stat_info) || is_single_row(&right_stat_info) {
+        // Prefer CROSS JOIN + FILTER when statistics prove or estimate one side at one row.
+        // HashJoin remains correct if the estimate is wrong and avoids the result-block
+        // overhead that RangeJoin can incur for this shape after join commutation.
         return Ok(PhysicalJoinType::Hash);
     }
 
@@ -221,8 +246,25 @@ impl PhysicalPlanBuilder {
         } else {
             match physical_join(join, s_expr)? {
                 PhysicalJoinType::Hash => {
+                    // When a LeftSingle/RightSingle join (scalar subquery) has no
+                    // equi-conditions, the hash join executes as a cross join + filter.
+                    // The single-join runtime check can fire during the cross-product
+                    // matching phase before the filter is applied. It is only safe to
+                    // clear the marker when the scalar side itself is proven to produce
+                    // exactly one row. A precise one-row cardinality on the other input
+                    // does not prove that.
+                    let join = if join.equi_conditions.is_empty()
+                        && join.single_to_inner.is_some()
+                        && single_join_scalar_side_is_precise_single_row(join, s_expr)?
+                    {
+                        let mut j = join.clone();
+                        j.single_to_inner = None;
+                        j
+                    } else {
+                        join.clone()
+                    };
                     self.build_hash_join(
-                        join,
+                        &join,
                         s_expr,
                         required,
                         others_required,

src/query/service/tests/it/sql/exec/range_join.rs

@@ -18,6 +18,122 @@ use databend_common_expression::types::number::NumberScalar;
 use databend_query::test_kits::TestFixture;
 use futures_util::TryStreamExt;
 
+async fn explain_text(fixture: &TestFixture, query: &str) -> anyhow::Result<String> {
+    let blocks = fixture
+        .execute_query(&format!("EXPLAIN {query}"))
+        .await?
+        .try_collect::<Vec<DataBlock>>()
+        .await?;
+    let block = DataBlock::concat(&blocks).expect("concat explain blocks");
+    let column = block.get_by_offset(0);
+    let mut lines = Vec::with_capacity(block.num_rows());
+    for row in 0..block.num_rows() {
+        if let Some(ScalarRef::String(line)) = column.index(row) {
+            lines.push(line.to_string());
+        }
+    }
+    Ok(lines.join("\n"))
+}
+
+fn extract_one_u64(blocks: Vec<DataBlock>) -> u64 {
+    let block = DataBlock::concat(&blocks).expect("concat result blocks");
+    assert_eq!(block.num_rows(), 1);
+    match block.get_by_offset(0).index(0) {
+        Some(ScalarRef::Number(NumberScalar::UInt64(v))) => v,
+        other => panic!("unexpected count scalar: {other:?}"),
+    }
+}
+
+/// An inequality join with a precise-one-row aggregate side should use CROSS JOIN + FILTER
+/// instead of RangeJoin. Exercise both equivalent predicate operand orders and verify the
+/// selected physical operator and query result.
+#[tokio::test(flavor = "multi_thread")]
+async fn test_inequality_join_with_scalar_side_avoids_range_join() -> anyhow::Result<()> {
+    let fixture = TestFixture::setup().await?;
+    fixture.create_default_database().await?;
+    let db = fixture.default_db_name();
+
+    fixture
+        .execute_command(&format!(
+            "CREATE TABLE {db}.big(ts INT) AS SELECT number FROM numbers(1000)"
+        ))
+        .await?;
+    fixture
+        .execute_command(&format!(
+            "CREATE TABLE {db}.threshold(ts INT) AS SELECT number FROM numbers(1000)"
+        ))
+        .await?;
+
+    // Scalar subquery on the right side of the inequality.
+    let q_right = format!(
+        "SELECT count(*) FROM {db}.big b \
+         WHERE b.ts >= (SELECT min(ts) FROM {db}.threshold)"
+    );
+    // Equivalent predicate with the scalar subquery written on the left.
+    let q_left = format!(
+        "SELECT count(*) FROM {db}.big b \
+         WHERE (SELECT min(ts) FROM {db}.threshold) <= b.ts"
+    );
+
+    for query in [&q_right, &q_left] {
+        let plan = explain_text(&fixture, query).await?;
+        assert!(
+            !plan.contains("RangeJoin"),
+            "single-row inequality side must not produce a RangeJoin, plan was:\n{plan}"
+        );
+    }
+
+    // Both forms must produce the correct result: all 1000 rows match `ts >= min(ts)`.
+    for query in [&q_right, &q_left] {
+        let blocks = fixture
+            .execute_query(query)
+            .await?
+            .try_collect::<Vec<DataBlock>>()
+            .await?;
+        assert_eq!(extract_one_u64(blocks), 1000, "query: {query}");
+    }
+
+    Ok(())
+}
+
+/// Force the exact CROSS JOIN + FILTER shape that used to report a false scalar-subquery
+/// cardinality violation. The pushed-down expression is true for all four rows, but its
+/// estimated selectivity places the four-row table on the build side and the exact-one-row
+/// scalar aggregate on the probe side. The CROSS JOIN + FILTER execution must not treat the
+/// four cross-product candidates as four scalar-subquery rows.
+#[tokio::test(flavor = "multi_thread")]
+async fn test_scalar_aggregate_probe_cross_join_filter() -> anyhow::Result<()> {
+    let fixture = TestFixture::setup().await?;
+    fixture.create_default_database().await?;
+    let db = fixture.default_db_name();
+
+    fixture
+        .execute_command(&format!(
+            "CREATE TABLE {db}.four_rows(ts INT) AS SELECT number FROM numbers(4)"
+        ))
+        .await?;
+
+    let query = format!(
+        "SELECT count(*) \
+         FROM (SELECT ts FROM {db}.four_rows WHERE length(to_string(ts)) > 0) b \
+         WHERE b.ts >= (SELECT min(number) FROM numbers(1))"
+    );
+
+    let plan = explain_text(&fixture, &query).await?;
+    assert!(plan.contains("HashJoin"), "plan was:\n{plan}");
+    assert!(plan.contains("TableScan(Build)"), "plan was:\n{plan}");
+    assert!(plan.contains("AggregateFinal(Probe)"), "plan was:\n{plan}");
+
+    let blocks = fixture
+        .execute_query(&query)
+        .await?
+        .try_collect::<Vec<DataBlock>>()
+        .await?;
+    assert_eq!(extract_one_u64(blocks), 4);
+
+    Ok(())
+}
+
 fn extract_two_u64(blocks: Vec<DataBlock>) -> (u64, u64) {
     let block = DataBlock::concat(&blocks).expect("concat blocks");
     assert_eq!(block.num_rows(), 1, "unexpected rows: {}", block.num_rows());

src/query/sql/src/planner/optimizer/optimizers/rule/join_rules/rule_commute_join.rs

@@ -113,6 +113,7 @@ impl Rule for RuleCommuteJoin {
                     (condition.right.clone(), condition.left.clone());
             }
             join.join_type = join.join_type.opposite();
+            join.single_to_inner = join.single_to_inner.map(|join_type| join_type.opposite());
             let mut result = SExpr::create_binary(
                 Arc::new(join.into()),
                 Arc::new(right_child.clone()),

src/query/sql/src/planner/optimizer/optimizers/rule/join_rules/rule_commute_join_base_table.rs

@@ -94,6 +94,7 @@ impl Rule for RuleCommuteJoinBaseTable {
                         (condition.right.clone(), condition.left.clone());
                 }
                 join.join_type = join.join_type.opposite();
+                join.single_to_inner = join.single_to_inner.map(|join_type| join_type.opposite());
                 let mut result = SExpr::create_binary(
                     Arc::new(join.into()),
                     Arc::new(right_child.clone()),

src/query/sql/src/planner/optimizer/optimizers/rule/join_rules/rule_left_exchange_join.rs

@@ -102,6 +102,12 @@ impl Rule for RuleLeftExchangeJoin {
         let t2 = s_expr.child(0)?.child(1)?;
         let t3 = s_expr.child(1)?;
 
+        // Reassociation does not preserve the scalar-cardinality semantics encoded by
+        // single_to_inner, so do not exchange either marked join.
+        if join1.single_to_inner.is_some() || join2.single_to_inner.is_some() {
+            return Ok(());
+        }
+
         // Ensure inner joins or cross joins.
         if !matches!(join1.join_type, JoinType::Inner | JoinType::Cross)
             || !matches!(join2.join_type, JoinType::Inner | JoinType::Cross)

tests/sqllogictests/suites/mode/standalone/explain/join.test

@@ -1026,6 +1026,36 @@ HashJoin
     ├── push downs: [filters: [], limit: NONE]
     └── estimated rows: 4.00
 
+# The filter estimates the table side at zero rows, so join commutation places the
+# precise-one-row aggregate on Probe. Keep this no-equi join on HashJoin, not RangeJoin.
+query T
+EXPLAIN SELECT * FROM (SELECT MAX(a) AS a FROM t2) AS s, t1 WHERE s.a <= t1.a AND t1.a > 100;
+----
+HashJoin
+├── output columns: [MAX(a) (#1), t1.a (#2)]
+├── join type: INNER
+├── build keys: []
+├── probe keys: []
+├── keys is null equal: []
+├── filters: [s.a (#1) <= t1.a (#2)]
+├── estimated rows: 0.00
+├── TableScan(Build)
+│   ├── table: default.default.t1
+│   ├── scan id: 1
+│   ├── output columns: [a (#2)]
+│   ├── read rows: 0
+│   ├── read size: 0
+│   ├── partitions total: 1
+│   ├── partitions scanned: 0
+│   ├── pruning stats: [segments: <read cost: <slt:ignore>, range pruning: 1 to 0 cost: <slt:ignore>>]
+│   ├── push downs: [filters: [is_true(t1.a (#2) > 100)], limit: NONE]
+│   └── estimated rows: 0.00
+└── EvalScalar(Probe)
+    ├── output columns: [MAX(a) (#1)]
+    ├── expressions: [1]
+    ├── estimated rows: 1.00
+    └── DummyTableScan
+
 query T
 EXPLAIN SELECT * FROM t1 JOIN t3 ON t1.a = t3.a;
 ----

tests/sqllogictests/suites/query/subquery.test

@@ -890,6 +890,9 @@ insert into t1 values(1), (2), (3);
 query error 1001.*Scalar subquery can't return more than one row
 select (select sum(a) from t1 where t1.a >= t2.a group by t1.a) from t1 as t2;
 
+query error 1001.*Scalar subquery can't return more than one row
+SELECT 1 WHERE (SELECT number FROM numbers(2)) >= 0;
+
 
 
 statement ok