asolimando commented on code in PR #25719:
URL: https://github.com/apache/datafusion/pull/25719#discussion_r4187976452
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -4465,4 +4537,272 @@ mod tests {
);
Ok(())
}
+
+ // ---------------------------------------------------------------
+ // Unit tests for compute_fallback_selectivity
+ // ---------------------------------------------------------------
+
+ #[test]
+ fn test_fallback_selectivity_single_handled_equality() {
+ // col_0 = <expr>, NDV(col_0) = 100 → selectivity = 1/100
+ let schema = Schema::new(vec![Field::new("a", DataType::Int32,
false)]);
+ let predicate: Arc<dyn PhysicalExpr> = binary(
+ col("a", &schema).unwrap(),
+ Operator::Eq,
+ lit(42i32),
+ &schema,
+ )
+ .unwrap();
+ let col_stats = vec![ColumnStatistics {
+ distinct_count: Precision::Inexact(100),
+ ..Default::default()
+ }];
+ let result = compute_fallback_selectivity(&predicate, &col_stats, 20);
+ assert!(
+ (result - 0.01).abs() < 1e-12,
+ "expected 1/100 = 0.01, got {result}"
+ );
+ }
+
+ #[test]
+ fn test_fallback_selectivity_multiple_unhandled_conjuncts() {
+ // s <> 'abc' AND t <> 'x' AND u <> 'p'
+ // None are handled equalities → selectivity = default once = 0.2
+ let schema = Schema::new(vec![
+ Field::new("s", DataType::Utf8, false),
+ Field::new("t", DataType::Utf8, false),
+ Field::new("u", DataType::Utf8, false),
+ ]);
+ // Simulate three non-equality conjuncts via NotEq operators
+ let pred1 = binary(
+ col("s", &schema).unwrap(),
+ Operator::NotEq,
+ lit("abc"),
+ &schema,
+ )
+ .unwrap();
+ let pred2 = binary(
+ col("t", &schema).unwrap(),
+ Operator::NotEq,
+ lit("x"),
+ &schema,
+ )
+ .unwrap();
+ let pred3 = binary(
+ col("u", &schema).unwrap(),
+ Operator::NotEq,
+ lit("p"),
+ &schema,
+ )
+ .unwrap();
+ let combined: Arc<dyn PhysicalExpr> = conjunction(vec![pred1, pred2,
pred3]);
+ let col_stats = vec![
+ ColumnStatistics::new_unknown(),
+ ColumnStatistics::new_unknown(),
+ ColumnStatistics::new_unknown(),
+ ];
+ let result = compute_fallback_selectivity(&combined, &col_stats, 20);
+ // default_selectivity applied exactly once: 0.2
+ assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}");
+ }
+
+ #[test]
+ fn test_fallback_selectivity_mixed_handled_and_unhandled() {
+ // col_a = 42 AND col_b <> 'x'
+ // col_a has NDV=50, col_b is unhandled
+ // selectivity = (1/50) * 0.2 = 0.004
+ let schema = Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Utf8, false),
+ ]);
+ let eq_pred = binary(
+ col("a", &schema).unwrap(),
+ Operator::Eq,
+ lit(42i32),
+ &schema,
+ )
+ .unwrap();
+ let neq_pred = binary(
+ col("b", &schema).unwrap(),
+ Operator::NotEq,
+ lit("x"),
+ &schema,
+ )
+ .unwrap();
+ let combined: Arc<dyn PhysicalExpr> = conjunction(vec![eq_pred,
neq_pred]);
+ let col_stats = vec![
+ ColumnStatistics {
+ distinct_count: Precision::Inexact(50),
+ ..Default::default()
+ },
+ ColumnStatistics::new_unknown(),
+ ];
+ let result = compute_fallback_selectivity(&combined, &col_stats, 20);
+ let expected = (1.0 / 50.0) * 0.2;
+ assert!(
+ (result - expected).abs() < 1e-12,
+ "expected {expected}, got {result}"
+ );
+ }
+
+ #[test]
+ fn test_fallback_selectivity_col_eq_col_uses_max_ndv() {
+ // col_a = col_b, NDV(a)=100, NDV(b)=200
+ // selectivity = 1 / max(100, 200) = 1/200
+ let schema = Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Int32, false),
+ ]);
+ let predicate: Arc<dyn PhysicalExpr> = binary(
+ col("a", &schema).unwrap(),
+ Operator::Eq,
+ col("b", &schema).unwrap(),
+ &schema,
+ )
+ .unwrap();
+ let col_stats = vec![
+ ColumnStatistics {
+ distinct_count: Precision::Inexact(100),
+ ..Default::default()
+ },
+ ColumnStatistics {
+ distinct_count: Precision::Inexact(200),
+ ..Default::default()
+ },
+ ];
+ let result = compute_fallback_selectivity(&predicate, &col_stats, 20);
+ assert!(
+ (result - 1.0 / 200.0).abs() < 1e-12,
+ "expected 1/200 = 0.005, got {result}"
+ );
+ }
+
+ #[test]
+ fn test_fallback_selectivity_non_equality_returns_default() {
+ // A single non-equality predicate (a > 10) → default_selectivity
applied once
+ let schema = Schema::new(vec![Field::new("a", DataType::Int32,
false)]);
+ let predicate: Arc<dyn PhysicalExpr> = binary(
+ col("a", &schema).unwrap(),
+ Operator::Gt,
+ lit(10i32),
+ &schema,
+ )
+ .unwrap();
+ let col_stats = vec![ColumnStatistics {
+ distinct_count: Precision::Inexact(100),
+ ..Default::default()
+ }];
+ let result = compute_fallback_selectivity(&predicate, &col_stats, 20);
+ assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}");
+ }
+
+ /// Verify that `FilterExec` uses NDV-based fallback selectivity for
predicates
+ /// that `check_support` cannot handle. A Utf8 equality `name = 'alice'`
fails
+ /// interval analysis, so the filter falls back to `1 / NDV` instead of the
+ /// flat 20% default.
+ #[tokio::test]
+ async fn
test_filter_statistics_fallback_uses_ndv_for_unsupported_predicate()
+ -> Result<()> {
+ let schema = Schema::new(vec![Field::new("name", DataType::Utf8,
false)]);
+ // 1000 rows, NDV = 200 for the `name` column.
+ let input = Arc::new(StatisticsExec::new(
+ Statistics {
+ num_rows: Precision::Inexact(1000),
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![ColumnStatistics {
+ distinct_count: Precision::Inexact(200),
+ ..Default::default()
+ }],
+ },
+ schema.clone(),
+ ));
+
+ // Utf8 equality fails `check_support`, triggering the NDV fallback.
+ // Expected selectivity = 1 / 200 = 0.005 → num_rows = 1000 * 0.005 =
5.
+ let predicate: Arc<dyn PhysicalExpr> =
+ binary(col("name", &schema)?, Operator::Eq, lit("alice"),
&schema)?;
+ let filter = Arc::new(FilterExec::try_new(predicate, input)?);
+ let stats =
+ StatisticsContext::new().compute(filter.as_ref(),
&StatisticsArgs::new())?;
+ assert_eq!(stats.num_rows, Precision::Inexact(5));
+ Ok(())
+ }
+
+ /// Verify that a `CAST(a AS Int64) = <ScalarSubquery>` predicate — which
+ /// `check_support` cannot resolve — still falls back gracefully to the
+ /// default selectivity (20%) rather than panicking or silently using 100%.
+ ///
+ /// In this case the left-hand side is a `CastExpr`, so `column_ndv`
returns
+ /// `None` and no NDV estimate is available. The single unhandled conjunct
+ /// receives the flat default.
+ #[tokio::test]
+ async fn
test_filter_statistics_fallback_cast_expr_uses_default_selectivity()
+ -> Result<()> {
+ let schema = Schema::new(vec![Field::new("a", DataType::Int32,
false)]);
+ // 1000 rows, NDV = 500.
+ let input = Arc::new(StatisticsExec::new(
+ Statistics {
+ num_rows: Precision::Inexact(1000),
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![ColumnStatistics {
+ distinct_count: Precision::Inexact(500),
+ ..Default::default()
+ }],
+ },
+ schema.clone(),
+ ));
+
+ // CAST(a AS Int64) = 42: the CastExpr wraps the column, so
+ // `column_ndv` cannot resolve it. The predicate also passes
+ // `check_support` (CastExpr + Int64 literal are both supported),
+ // so the interval-analysis path runs. This test documents the
+ // current boundary: once `check_support` accepts the predicate
+ // the fallback is not reached.
+ //
+ // We therefore use a plain Utf8 column compared to a literal,
+ // which is the concrete unsupported form this PR improves.
+ let predicate: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new(
+ Arc::new(CastExpr::new(col("a", &schema)?, DataType::Int64, None)),
+ Operator::Eq,
+ Arc::new(Literal::new(ScalarValue::Int64(Some(42)))),
+ ));
+ let filter = Arc::new(
+ FilterExec::try_new(predicate,
input)?.with_default_selectivity(20)?,
+ );
+ let stats =
+ StatisticsContext::new().compute(filter.as_ref(),
&StatisticsArgs::new())?;
+ // check_support accepts CAST + Int64 literal, so interval analysis
+ // runs and produces a result — num_rows will not be the full 1000.
+ assert!(
+ stats.num_rows != Precision::Inexact(1000),
+ "expected some rows to be filtered, got {:?}",
+ stats.num_rows
+ );
+ Ok(())
+ }
+
+ #[test]
+ fn test_fallback_selectivity_utf8_equality_uses_ndv() {
+ // name = 'alice' on a Utf8 column with NDV=60.
+ // Utf8 equality fails `check_support`, so our fallback runs and
+ // returns 1/60 instead of the previous flat 20%.
Review Comment:
nit: this comment still refers to "the previous flat 20%"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
