diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index b9ecc59d1a1ff..61e495f979689 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1373,10 +1373,12 @@ config_namespace! { /// rewrite; other predicates and Bloom-filter pruning remain available. /// /// Within the cap, nonempty lists of at most 20 values use the existing - /// per-value rewrite. Larger non-null literal string lists on a string - /// column use a compact sorted domain, for both `IN` and `NOT IN`. - /// Other lists retain the existing per-value rewrite, so raising the cap - /// can make those predicates expensive to build and evaluate. + /// per-value rewrite. Larger literal string lists on a string column use + /// a compact representation, for both `IN` and `NOT IN`, including lists + /// with NULL members. `NOT IN` with NULL and all-NULL `IN` lists cannot + /// match any rows. Other lists retain the existing per-value rewrite, so + /// raising the cap can make those predicates expensive to build and + /// evaluate. /// /// Defaults to 20. pub max_in_list_size: usize, default = 20 diff --git a/datafusion/core/tests/parquet/string_in_list_pruning.rs b/datafusion/core/tests/parquet/string_in_list_pruning.rs index 56a83449a1320..52d2c58bdd943 100644 --- a/datafusion/core/tests/parquet/string_in_list_pruning.rs +++ b/datafusion/core/tests/parquet/string_in_list_pruning.rs @@ -440,23 +440,24 @@ async fn string_not_in_list_with_truncated_bounds() { } #[tokio::test] -async fn string_not_in_list_with_null_does_not_bypass_row_filter() { +async fn string_in_list_with_null_preserves_filter_semantics() { let mut file = tempfile::Builder::new() - .prefix("string_not_in_list_pruning") + .prefix("string_in_list_null_pruning") .suffix(".parquet") .tempfile() .unwrap(); let schema = Arc::new(Schema::new(vec![Field::new("value", DataType::Utf8, true)])); // The first row group has a known zero null count, and every value lies - // in a gap in the IN list. Dropping the NULL list member while inverting - // NOT IN would incorrectly prove that this entire row group matches. + // in a gap in the IN list. The matching value in the second row group is + // deliberately not first, so incorrectly bypassing the row filter changes + // the result when the scan has a limit. let values = vec![ Some("v000001"), Some("v000001"), Some("v000001"), Some("v000001"), - Some("v000000"), Some("v000001"), + Some("v000000"), None, Some("v999999"), ]; @@ -475,13 +476,11 @@ async fn string_not_in_list_with_null_does_not_bypass_row_filter() { assert_eq!(writer.close().unwrap().num_row_groups(), 2); // Build the physical source directly so a logical optimizer cannot fold - // the SQL NOT IN (..., NULL) filter to an empty relation before the scan. + // NOT IN (..., NULL) to an empty relation before the scan. let mut list = (0..21) .map(|index| lit(format!("v{:06}", index * 10))) .collect::>(); list.push(lit(ScalarValue::Utf8(None))); - let predicate = - in_list(col("value", &schema).unwrap(), list, &true, &schema).unwrap(); let location = Path::from_filesystem_path(file.path()).unwrap(); let partitioned_file = PartitionedFile::new( location.to_string(), @@ -490,48 +489,79 @@ async fn string_not_in_list_with_null_does_not_bypass_row_filter() { let ctx = SessionContext::new_with_config(SessionConfig::new().with_target_partitions(1)); - for max_in_list_size in [0, 32] { - let mut options = TableParquetOptions::default(); - options.global.max_in_list_size = max_in_list_size; - let source = Arc::new( - ParquetSource::new(Arc::clone(&schema)) - .with_table_parquet_options(options) - .with_predicate(Arc::clone(&predicate)) - .with_pushdown_filters(true) - .with_enable_page_index(false) - .with_bloom_filter_on_read(false), - ); - let config = - FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source) - .with_file(partitioned_file.clone()) - .with_limit(Some(1)) - .build(); - let plan: Arc = - Arc::new(DataSourceExec::new(Arc::new(config))); - let plan_text = displayable(plan.as_ref()).indent(true).to_string(); - assert!(plan_text.contains("NOT IN"), "{plan_text}"); - let batches = collect(Arc::clone(&plan), ctx.task_ctx()).await.unwrap(); - let output = ScanOutput { - batches, - plan: plan_text, - metrics: MetricsFinder::find_metrics(plan.as_ref()).unwrap(), - }; - - assert_eq!( - output - .batches - .iter() - .map(RecordBatch::num_rows) - .sum::(), - 0, - "cap={max_in_list_size}, plan={}, metrics={}", - output.plan, - output.metrics - ); - assert_eq!(output.fully_matched("row_groups_pruned_statistics"), 0); - assert_eq!(output.pruned("row_groups_pruned_statistics"), 0); - assert_eq!(output.pruned("limit_pruned_row_groups"), 0); - assert_eq!(output.counter("pushdown_rows_pruned"), 8); - assert_eq!(output.counter("predicate_evaluation_errors"), 0); + for negated in [false, true] { + let predicate = in_list( + col("value", &schema).unwrap(), + list.clone(), + &negated, + &schema, + ) + .unwrap(); + for max_in_list_size in [0, 32] { + let mut options = TableParquetOptions::default(); + options.global.max_in_list_size = max_in_list_size; + let source = Arc::new( + ParquetSource::new(Arc::clone(&schema)) + .with_table_parquet_options(options) + .with_predicate(Arc::clone(&predicate)) + .with_pushdown_filters(true) + .with_enable_page_index(false) + .with_bloom_filter_on_read(false), + ); + let config = + FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source) + .with_file(partitioned_file.clone()) + .with_limit(Some(1)) + .build(); + let plan: Arc = + Arc::new(DataSourceExec::new(Arc::new(config))); + let plan_text = displayable(plan.as_ref()).indent(true).to_string(); + assert!(plan_text.contains("IN"), "{plan_text}"); + let batches = collect(Arc::clone(&plan), ctx.task_ctx()).await.unwrap(); + let output = ScanOutput { + batches, + plan: plan_text, + metrics: MetricsFinder::find_metrics(plan.as_ref()).unwrap(), + }; + + if negated { + assert!(output.batches.iter().all(|batch| batch.num_rows() == 0)); + } else { + assert_batches_eq!( + [ + "+---------+", + "| value |", + "+---------+", + "| v000000 |", + "+---------+", + ], + &output.batches + ); + } + assert_eq!( + output.counter("pushdown_rows_pruned"), + match (negated, max_in_list_size) { + (false, 0) => 7, + (false, _) => 3, + (true, 0) => 8, + (true, _) => 0, + } + ); + assert_eq!(output.fully_matched("row_groups_pruned_statistics"), 0); + assert_eq!( + output.pruned("row_groups_pruned_statistics"), + if max_in_list_size == 0 { + 0 + } else if negated { + 2 + } else { + 1 + }, + "negated={negated}, cap={max_in_list_size}, metrics={}", + output.metrics + ); + assert_eq!(output.pruned("limit_pruned_row_groups"), 0); + assert_eq!(output.counter("predicate_evaluation_errors"), 0); + } } } diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs index 2db4cb718d364..5ca99f2498e75 100644 --- a/datafusion/datasource-parquet/src/row_group_filter.rs +++ b/datafusion/datasource-parquet/src/row_group_filter.rs @@ -395,6 +395,11 @@ impl RowGroupAccessPlanFilter { if candidate_row_group_indices.is_empty() { return; } + // Some pruning rewrites preserve only whether rows can be TRUE, while + // full-match inference must distinguish FALSE from UNKNOWN. + if !predicate.can_be_inverted_for_full_match() { + return; + } let arrow_schema = pruning_stats.arrow_schema; let mut inverted_expr: Arc = @@ -432,6 +437,7 @@ impl RowGroupAccessPlanFilter { let Ok(inverted_predicate) = PruningPredicateBuilder::new() .with_file_schema(Arc::clone(predicate.schema())) + .with_max_in_list_size(predicate.max_in_list_size()) .try_build(inverted_expr) else { return; diff --git a/datafusion/pruning/benches/string_in_list_pruning.rs b/datafusion/pruning/benches/string_in_list_pruning.rs index 7431d08d25041..54d30922d7300 100644 --- a/datafusion/pruning/benches/string_in_list_pruning.rs +++ b/datafusion/pruning/benches/string_in_list_pruning.rs @@ -24,9 +24,10 @@ //! without making the baseline depend on a deeply nested expression. //! //! The main cases alternate intervals pinned to a domain member with intervals -//! that span a sparse gap. Supplemental `NOT IN` cases measure uniform singleton -//! containers, long bounds, and long literals that share bounds' prefixes. Bloom -//! filters are not involved. +//! that span a sparse gap. NULL-containing variants cover the optimized +//! filter-semantics paths, including the constant-false `NOT IN` rewrite. +//! Supplemental `NOT IN` cases measure uniform singleton containers, long bounds, +//! and long literals that share bounds' prefixes. Bloom filters are not involved. //! //! Run with `cargo bench -p datafusion-pruning --bench string_in_list_pruning`. //! The construction benchmarks reuse their input physical expressions; the @@ -188,12 +189,16 @@ struct BenchmarkCase { size: usize, schema: SchemaRef, in_list: PhysicalExprRef, + in_list_with_null: PhysicalExprRef, expanded_or: PhysicalExprRef, not_in_list: PhysicalExprRef, + not_in_list_with_null: PhysicalExprRef, expanded_and: PhysicalExprRef, in_list_predicate: PruningPredicate, + in_list_with_null_predicate: PruningPredicate, expanded_or_predicate: PruningPredicate, not_in_list_predicate: PruningPredicate, + not_in_list_with_null_predicate: PruningPredicate, expanded_and_predicate: PruningPredicate, statistics: IntervalStatistics, } @@ -209,15 +214,30 @@ impl BenchmarkCase { let values = (0..size) .map(|index| lit(ScalarValue::new_utf8view(value(index * 10)))) .collect::>(); + let mut values_with_null = values.clone(); + values_with_null.push(lit(ScalarValue::Utf8View(None))); let not_in_list = in_list(Arc::clone(&column), values.clone(), &true, &schema).unwrap(); + let not_in_list_with_null = in_list( + Arc::clone(&column), + values_with_null.clone(), + &true, + &schema, + ) + .unwrap(); + let in_list_with_null = + in_list(Arc::clone(&column), values_with_null, &false, &schema).unwrap(); let in_list = in_list(Arc::clone(&column), values.clone(), &false, &schema).unwrap(); let expanded_or = expanded(&column, &values, Operator::Eq, Operator::Or); let expanded_and = expanded(&column, &values, Operator::NotEq, Operator::And); let in_list_predicate = build_predicate(&in_list, &schema, size); + let in_list_with_null_predicate = + build_predicate(&in_list_with_null, &schema, size + 1); let expanded_or_predicate = build_predicate(&expanded_or, &schema, size); let not_in_list_predicate = build_predicate(¬_in_list, &schema, size); + let not_in_list_with_null_predicate = + build_predicate(¬_in_list_with_null, &schema, size + 1); let expanded_and_predicate = build_predicate(&expanded_and, &schema, size); eprintln!( "string_in_list_pruning: {size} values, compact in={}, compact not in={}", @@ -239,6 +259,10 @@ impl BenchmarkCase { .collect::>(); let negated_kept = kept.iter().map(|keep| !keep).collect::>(); assert_eq!(in_list_predicate.prune(&statistics).unwrap(), kept); + assert_eq!( + in_list_with_null_predicate.prune(&statistics).unwrap(), + kept + ); assert_eq!(expanded_or_predicate.prune(&statistics).unwrap(), kept); assert_eq!( not_in_list_predicate.prune(&statistics).unwrap(), @@ -248,17 +272,28 @@ impl BenchmarkCase { expanded_and_predicate.prune(&statistics).unwrap(), negated_kept ); + assert!( + not_in_list_with_null_predicate + .prune(&statistics) + .unwrap() + .iter() + .all(|keep| !keep) + ); Self { size, schema, in_list, + in_list_with_null, expanded_or, not_in_list, + not_in_list_with_null, expanded_and, in_list_predicate, + in_list_with_null_predicate, expanded_or_predicate, not_in_list_predicate, + not_in_list_with_null_predicate, expanded_and_predicate, statistics, } @@ -269,22 +304,28 @@ fn criterion_benchmark(criterion: &mut Criterion) { let cases = DOMAIN_SIZES.map(BenchmarkCase::new); let mut construction = criterion.benchmark_group("string_in_list_pruning/construct"); for case in &cases { - construction.throughput(Throughput::Elements(case.size as u64)); - for (name, expression) in [ - ("in_list", &case.in_list), - ("expanded_or", &case.expanded_or), - ("not_in_list", &case.not_in_list), - ("expanded_and", &case.expanded_and), + for (name, expression, max_in_list_size) in [ + ("in_list", &case.in_list, case.size), + ("in_list_with_null", &case.in_list_with_null, case.size + 1), + ("expanded_or", &case.expanded_or, case.size), + ("not_in_list", &case.not_in_list, case.size), + ( + "not_in_list_with_null_constant_false", + &case.not_in_list_with_null, + case.size + 1, + ), + ("expanded_and", &case.expanded_and, case.size), ] { + construction.throughput(Throughput::Elements(max_in_list_size as u64)); construction.bench_with_input( - BenchmarkId::new(name, case.size), + BenchmarkId::new(name, max_in_list_size), expression, |bencher, expression| { bencher.iter(|| { black_box(build_predicate( black_box(expression), &case.schema, - case.size, + max_in_list_size, )) }); }, @@ -296,14 +337,24 @@ fn criterion_benchmark(criterion: &mut Criterion) { let mut evaluation = criterion.benchmark_group("string_in_list_pruning/evaluate"); evaluation.throughput(Throughput::Elements(CONTAINERS as u64)); for case in &cases { - for (name, predicate) in [ - ("in_list", &case.in_list_predicate), - ("expanded_or", &case.expanded_or_predicate), - ("not_in_list", &case.not_in_list_predicate), - ("expanded_and", &case.expanded_and_predicate), + for (name, predicate, list_size) in [ + ("in_list", &case.in_list_predicate, case.size), + ( + "in_list_with_null", + &case.in_list_with_null_predicate, + case.size + 1, + ), + ("expanded_or", &case.expanded_or_predicate, case.size), + ("not_in_list", &case.not_in_list_predicate, case.size), + ( + "not_in_list_with_null_constant_false", + &case.not_in_list_with_null_predicate, + case.size + 1, + ), + ("expanded_and", &case.expanded_and_predicate, case.size), ] { evaluation.bench_with_input( - BenchmarkId::new(name, case.size), + BenchmarkId::new(name, list_size), predicate, |bencher, predicate| { bencher.iter(|| { diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index 7ef255cb5aebb..c3362c63299e1 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -380,6 +380,17 @@ pub struct PruningPredicate { /// /// See [`PruningPredicate::literal_guarantees`] for more details. literal_guarantees: Vec, + /// Maximum IN-list size used to build this predicate. + max_in_list_size: usize, + /// Whether its logical inverse can safely prove every row matches. + can_be_inverted_for_full_match: bool, +} + +#[derive(Default)] +struct PruningExpressionProperties { + /// The rewrite preserves which rows can be TRUE, but not whether rejected + /// rows are FALSE or UNKNOWN. + has_filter_semantics_only: bool, } /// Build a pruning predicate from an optional predicate expression. @@ -450,7 +461,7 @@ impl<'a> PruningPredicateBuilder<'a> { /// | Condition | Pruning representation | /// | --- | --- | /// | `N <= min(20, C)` | Existing per-value rewrite | - /// | `20 < N <= C`, non-null literal strings on a string column | Compact sorted domain | + /// | `20 < N <= C`, literal strings with optional NULLs on a string column | Compact pruning expression | /// | `20 < N <= C`, other lists | Existing per-value rewrite | /// | `N > C` | Unhandled-predicate hook, normally "keep the container" | /// @@ -459,9 +470,9 @@ impl<'a> PruningPredicateBuilder<'a> { /// pruning (such as Bloom filters). The default cap is [`MAX_IN_LIST_SIZE`] /// (20), so the compact path requires an explicitly raised cap. /// - /// The compact form covers `IN` and `NOT IN` alike. Raising the cap can - /// still build large comparison trees for non-string lists or lists - /// containing NULL; their handling is unchanged. + /// The compact form covers `IN` and `NOT IN` alike, including lists with + /// NULL members. Raising the cap can still build large comparison trees for + /// other eligible lists. /// /// Query engines typically pass /// `datafusion.execution.parquet.max_in_list_size` here. @@ -528,12 +539,14 @@ impl<'a> PruningPredicateBuilder<'a> { // build predicate expression once let mut required_columns = RequiredColumns::new(); + let mut properties = PruningExpressionProperties::default(); let predicate_expr = build_predicate_expression( &predicate, &file_schema, &mut required_columns, &unhandled_hook, self.max_in_list_size, + &mut properties, ); let predicate_schema = required_columns.schema(); // Simplify the newly created predicate to get rid of redundant casts, comparisons, etc. @@ -547,6 +560,8 @@ impl<'a> PruningPredicateBuilder<'a> { required_columns, orig_expr: predicate, literal_guarantees, + max_in_list_size: self.max_in_list_size, + can_be_inverted_for_full_match: !properties.has_filter_semantics_only, }) } } @@ -719,6 +734,17 @@ impl PruningPredicate { is_always_true(&self.predicate_expr) && self.literal_guarantees.is_empty() } + /// Returns the configured maximum IN-list size used to build this predicate. + pub fn max_in_list_size(&self) -> usize { + self.max_in_list_size + } + + /// Returns whether pruning the logical inverse can safely prove that every + /// row in a container satisfies the original predicate. + pub fn can_be_inverted_for_full_match(&self) -> bool { + self.can_be_inverted_for_full_match + } + pub fn required_columns(&self) -> &RequiredColumns { &self.required_columns } @@ -1471,7 +1497,7 @@ fn build_is_null_column_expr( } } -/// Keep large literal string domains compact instead of building a per-value +/// Keep large literal string lists compact instead of building a per-value /// tree: an OR tree for `IN`, an AND chain for `NOT IN`. /// /// `IN` excludes a container whose interval is disjoint from the domain. That @@ -1482,10 +1508,14 @@ fn build_is_null_column_expr( /// value the domain holds, which is exactly what makes the per-value /// `min != v OR v != max` chain false. Its decisions match that chain /// everywhere, including absent and inverted bounds. +/// +/// A NULL list member makes `NOT IN` and an all-NULL `IN` list never TRUE. For +/// other `IN` lists, NULL does not change which rows can make the predicate TRUE. fn build_string_in_list_expr( in_list: &phys_expr::InListExpr, schema: &Schema, required_columns: &mut RequiredColumns, + properties: &mut PruningExpressionProperties, ) -> Option> { let membership = if in_list.negated() { SetMembership::NotIn @@ -1501,14 +1531,30 @@ fn build_string_in_list_expr( if field.name() != column.name() || !data_type.is_string() { return None; } - // NULLs must remain unhandled: the inverse predicate is also used to prove - // that every row matches, and IN (..., NULL) can evaluate to UNKNOWN. - // Tracked in https://github.com/apache/datafusion/issues/24711. - let values = in_list - .list() - .iter() - .map(|expr| extract_string_literal(expr).map(str::to_owned)) - .collect::>>()?; + let mut values = Vec::with_capacity(in_list.list().len()); + let mut contains_null = false; + for expr in in_list.list() { + if let Some(value) = extract_string_literal(expr) { + values.push(value.to_owned()); + } else if expr + .downcast_ref::() + .is_some_and(|literal| literal.value().is_null()) + { + contains_null = true; + } else { + return None; + } + } + + // Pruning asks only whether the predicate can be TRUE. UNKNOWN and FALSE + // both reject a row, so NOT IN with a NULL member and an all-NULL IN list + // can never match. IN can otherwise ignore NULL and search its non-null domain. + if contains_null && (membership == SetMembership::NotIn || values.is_empty()) { + properties.has_filter_semantics_only = true; + return Some(Arc::new(phys_expr::Literal::new(ScalarValue::Boolean( + Some(false), + )))); + } let min = required_columns .min_column_expr(column, in_list.expr(), field) .ok()?; @@ -1518,6 +1564,9 @@ fn build_string_in_list_expr( let non_null = build_is_null_column_expr(in_list.expr(), schema, required_columns, true)?; let may_match = Arc::new(StringInListPruningExpr::new(membership, min, max, values)); + if contains_null { + properties.has_filter_semantics_only = true; + } Some(Arc::new(phys_expr::BinaryExpr::new( non_null, Operator::And, @@ -1595,12 +1644,14 @@ impl PredicateRewriter { schema: &Schema, ) -> Arc { let mut required_columns = RequiredColumns::new(); + let mut properties = PruningExpressionProperties::default(); build_predicate_expression( expr, &Arc::new(schema.clone()), &mut required_columns, &self.unhandled_hook, self.max_in_list_size, + &mut properties, ) } } @@ -1614,7 +1665,7 @@ impl PredicateRewriter { /// Returns the pruning predicate as an [`PhysicalExpr`] /// /// `max_in_list_size` is the largest `IN (...)` list eligible for statistics -/// pruning. Large literal string lists use a compact sorted domain, for both +/// pruning. Large literal string lists use a compact representation, for both /// `IN` and `NOT IN`; other eligible lists use per-value checks. Longer lists /// fall back to `unhandled_hook`. fn build_predicate_expression( @@ -1623,6 +1674,7 @@ fn build_predicate_expression( required_columns: &mut RequiredColumns, unhandled_hook: &Arc, max_in_list_size: usize, + properties: &mut PruningExpressionProperties, ) -> Arc { if is_always_false(expr) { // Shouldn't return `unhandled_hook.handle(expr)` @@ -1664,7 +1716,7 @@ fn build_predicate_expression( if in_list.list().len() > MAX_IN_LIST_SIZE && in_list.list().len() <= max_in_list_size && let Some(pruning_expr) = - build_string_in_list_expr(in_list, schema, required_columns) + build_string_in_list_expr(in_list, schema, required_columns, properties) { return pruning_expr; } @@ -1697,6 +1749,7 @@ fn build_predicate_expression( required_columns, unhandled_hook, max_in_list_size, + properties, ); } else { return unhandled_hook.handle(expr); @@ -1737,6 +1790,7 @@ fn build_predicate_expression( required_columns, unhandled_hook, max_in_list_size, + properties, ); let right_expr = build_predicate_expression( &right, @@ -1744,6 +1798,7 @@ fn build_predicate_expression( required_columns, unhandled_hook, max_in_list_size, + properties, ); // simplify boolean expression if applicable let expr = match (&left_expr, op, &right_expr) { @@ -3921,6 +3976,8 @@ mod tests { !compact, "negated={negated}, limit={limit}" ); + assert!(predicate.can_be_inverted_for_full_match()); + assert_eq!(predicate.max_in_list_size(), limit); } let default = PruningPredicateBuilder::new() @@ -3935,10 +3992,23 @@ mod tests { } #[test] - fn large_string_in_list_keeps_null_semantics() -> Result<()> { + fn large_string_in_list_compacts_null_literals() -> Result<()> { let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Utf8, true)])); let values = (0..21).map(|i| lit(format!("a{i:03}"))).collect::>(); let stats = TestStatistics::new().with( + "c1", + ContainerStats::new_utf8( + [Some("a005"), Some("other")], + [Some("a005"), Some("other")], + ) + .with_null_counts([Some(0); 2]) + .with_row_counts([Some(1); 2]), + ); + + let mut with_null = values; + with_null.push(lit(ScalarValue::Utf8(None))); + + let or_stats = TestStatistics::new().with( "c1", ContainerStats::new_utf8( [Some("middle"), Some("other")], @@ -3947,39 +4017,68 @@ mod tests { .with_null_counts([Some(0); 2]) .with_row_counts([Some(1); 2]), ); - let positive = col("c1").in_list(values.clone(), false); + let positive_or = col("c1") + .in_list(with_null.clone(), false) + .or(col("c1").eq(lit("middle"))); let predicate = large_string_pruning_predicate( - logical2physical(&positive.or(col("c1").eq(lit("middle"))), &schema), + logical2physical(&positive_or, &schema), Arc::clone(&schema), )?; - assert_eq!(predicate.prune(&stats)?, [true, false]); + assert_eq!(predicate.prune(&or_stats)?, [true, false]); + assert!(!predicate.can_be_inverted_for_full_match()); - let mut with_null = values; - with_null.push(lit(ScalarValue::Utf8(None))); - for expr in [ - col("c1").in_list(with_null.clone(), false), - col("c1").in_list(with_null.clone(), true), - ] { + for negated in [false, true] { + let expr = col("c1").in_list(with_null.clone(), negated); let physical = logical2physical(&expr, &schema); let default = PruningPredicateBuilder::new() .with_file_schema(Arc::clone(&schema)) .try_build(Arc::clone(&physical))?; assert!(is_always_true(default.predicate_expr()), "{expr}"); assert_eq!(default.prune(&stats)?, [true, true]); + assert!(default.can_be_inverted_for_full_match()); - // Raising the cap retains the existing per-value rewrite for lists - // containing NULL, in either direction. Dropping the NULL literal - // would turn UNKNOWN into FALSE, which the inverse-predicate proof - // in identify_fully_matched_row_groups cannot absorb. - // See https://github.com/apache/datafusion/issues/24711. let raised = large_string_pruning_predicate(physical, Arc::clone(&schema))?; - let raised = raised.predicate_expr().to_string(); - assert!(!raised.contains("IN_SET_INTERSECTS"), "{expr}"); - assert!(!raised.contains("NOT_IN_SET_MAY_MATCH"), "{expr}"); + assert!(!raised.can_be_inverted_for_full_match()); + if negated { + assert!(is_always_false(raised.predicate_expr()), "{expr}"); + assert_eq!(raised.prune(&stats)?, [false, false]); + } else { + assert!( + raised + .predicate_expr() + .to_string() + .contains("IN_SET_INTERSECTS"), + "{expr}" + ); + assert_eq!(raised.prune(&stats)?, [true, false]); + } } - // Inverting NOT IN (..., NULL) must not prove a full match and bypass - // the original row filter, which returns UNKNOWN for both rows. + let all_null = + std::iter::repeat_n(lit(ScalarValue::Utf8(None)), 21).collect::>(); + for negated in [false, true] { + let expr = col("c1").in_list(all_null.clone(), negated); + let predicate = large_string_pruning_predicate( + logical2physical(&expr, &schema), + Arc::clone(&schema), + )?; + assert!(is_always_false(predicate.predicate_expr()), "{expr}"); + assert_eq!(predicate.prune(&stats)?, [false, false]); + assert!(!predicate.can_be_inverted_for_full_match()); + } + + let integer_schema = + Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)])); + let mut integer_values = (0..21).map(lit).collect::>(); + integer_values.push(lit(ScalarValue::Int32(None))); + let integer_predicate = large_string_pruning_predicate( + logical2physical(&col("c1").in_list(integer_values, false), &integer_schema), + integer_schema, + )?; + assert!(integer_predicate.can_be_inverted_for_full_match()); + + // The compact false predicate has the same filter result as the original + // NOT IN expression, which returns UNKNOWN for values outside the list. let not_in = logical2physical(&col("c1").in_list(with_null, true), &schema); let batch = RecordBatch::try_new( schema, @@ -4142,10 +4241,9 @@ mod tests { /// `identify_fully_matched_row_groups` proves "every row matches" by showing /// the pruning predicate for `NOT P OR IsNull(col)` excludes the container. - /// That inference needs `P` to be two-valued, so the compact `NOT IN` form - /// and the compact `IN` it inverts to must stay exact. Both sides use a - /// raised cap here, which is what wiring the configured cap into the - /// inverted builder, or removing the lower bound, would produce. + /// For non-NULL lists, the compact `NOT IN` form and the compact `IN` it + /// inverts to are precise enough for this proof. Both sides use a raised cap + /// here, matching the configuration used by row-group pruning. #[test] fn large_string_not_in_list_inverts_without_false_full_match() -> Result<()> { let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Utf8, true)])); @@ -6693,12 +6791,14 @@ mod tests { ) -> Arc { let expr = logical2physical(expr, schema); let unhandled_hook = Arc::new(ConstantUnhandledPredicateHook::default()) as _; + let mut properties = PruningExpressionProperties::default(); build_predicate_expression( &expr, &Arc::new(schema.clone()), required_columns, &unhandled_hook, MAX_IN_LIST_SIZE, + &mut properties, ) } diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index e895b6b8be1c1..261efc3f28141 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -413,7 +413,7 @@ datafusion.execution.parquet.dictionary_page_size_limit 1048576 (writing) Sets b datafusion.execution.parquet.enable_page_index true (reading) If true, reads the Parquet data page level metadata (the Page Index), if present, to reduce the I/O and number of rows decoded. datafusion.execution.parquet.encoding NULL (writing) Sets default encoding for any column. Valid values are: plain, plain_dictionary, rle, bit_packed, delta_binary_packed, delta_length_byte_array, delta_byte_array, rle_dictionary, and byte_stream_split. These values are not case sensitive. If NULL, uses default parquet writer setting datafusion.execution.parquet.force_filter_selections false (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. -datafusion.execution.parquet.max_in_list_size 20 Maximum number of input values in an `IN (...)` list eligible for min/max pruning. Lists above this cap, or a cap of 0, skip this rewrite; other predicates and Bloom-filter pruning remain available. Within the cap, nonempty lists of at most 20 values use the existing per-value rewrite. Larger non-null literal string lists on a string column use a compact sorted domain, for both `IN` and `NOT IN`. Other lists retain the existing per-value rewrite, so raising the cap can make those predicates expensive to build and evaluate. Defaults to 20. +datafusion.execution.parquet.max_in_list_size 20 Maximum number of input values in an `IN (...)` list eligible for min/max pruning. Lists above this cap, or a cap of 0, skip this rewrite; other predicates and Bloom-filter pruning remain available. Within the cap, nonempty lists of at most 20 values use the existing per-value rewrite. Larger literal string lists on a string column use a compact representation, for both `IN` and `NOT IN`, including lists with NULL members. `NOT IN` with NULL and all-NULL `IN` lists cannot match any rows. Other lists retain the existing per-value rewrite, so raising the cap can make those predicates expensive to build and evaluate. Defaults to 20. datafusion.execution.parquet.max_predicate_cache_size NULL (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching. datafusion.execution.parquet.max_row_group_bytes NULL (writing) Target maximum size of each row group in bytes. When set, the writer flushes whenever either this limit or `max_row_group_size` is reached, whichever comes first. Useful for bounding writer memory on wide schemas where a row-count limit can map to very different byte sizes. Matches the behavior of `parquet.block.size` in parquet-mr. If `None` (the default), only the row-count limit applies. Currently only honored when `allow_single_file_parallelism` is `false`; by default the parallel file writer ignores this limit. datafusion.execution.parquet.max_row_group_size 1048576 (writing) Target maximum number of rows in each row group (defaults to 1M rows). Writing larger row groups requires more memory to write, but can get better compression and be faster to read. When `max_row_group_bytes` is also set, the writer flushes a row group when either limit is reached, whichever comes first. diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 313aa985ed854..987986b40409b 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -93,7 +93,7 @@ The following configuration settings are available: | datafusion.execution.parquet.coerce_int96_tz | NULL | (reading) Optional timezone applied to INT96 columns when `coerce_int96` is set. When `Some`, INT96 columns coerce to `Timestamp(, Some())` instead of the default `Timestamp(, None)`. Spark and other systems write INT96 values as UTC-adjusted instants, so callers that need the resulting Arrow type to be timezone-aware (e.g. for Spark `TimestampType` semantics) should set this to `"UTC"`. No effect when `coerce_int96` is `None`. | | datafusion.execution.parquet.bloom_filter_on_read | true | (reading) Use any available bloom filters when reading parquet files | | datafusion.execution.parquet.max_predicate_cache_size | NULL | (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching. | -| datafusion.execution.parquet.max_in_list_size | 20 | Maximum number of input values in an `IN (...)` list eligible for min/max pruning. Lists above this cap, or a cap of 0, skip this rewrite; other predicates and Bloom-filter pruning remain available. Within the cap, nonempty lists of at most 20 values use the existing per-value rewrite. Larger non-null literal string lists on a string column use a compact sorted domain, for both `IN` and `NOT IN`. Other lists retain the existing per-value rewrite, so raising the cap can make those predicates expensive to build and evaluate. Defaults to 20. | +| datafusion.execution.parquet.max_in_list_size | 20 | Maximum number of input values in an `IN (...)` list eligible for min/max pruning. Lists above this cap, or a cap of 0, skip this rewrite; other predicates and Bloom-filter pruning remain available. Within the cap, nonempty lists of at most 20 values use the existing per-value rewrite. Larger literal string lists on a string column use a compact representation, for both `IN` and `NOT IN`, including lists with NULL members. `NOT IN` with NULL and all-NULL `IN` lists cannot match any rows. Other lists retain the existing per-value rewrite, so raising the cap can make those predicates expensive to build and evaluate. Defaults to 20. | | datafusion.execution.parquet.data_pagesize_limit | 1048576 | (writing) Sets best effort maximum size of data page in bytes | | datafusion.execution.parquet.write_batch_size | 1024 | (writing) Sets write_batch_size in rows | | datafusion.execution.parquet.writer_version | 1.0 | (writing) Sets parquet writer version valid values are "1.0" and "2.0" |