diff --git a/src/datafusion-local/src/tests/column_free_conjunct.rs b/src/datafusion-local/src/tests/column_free_conjunct.rs new file mode 100644 index 00000000..fafea8e0 --- /dev/null +++ b/src/datafusion-local/src/tests/column_free_conjunct.rs @@ -0,0 +1,145 @@ +//! Regression tests for issue #19: a pushed-down conjunct that references no column. +//! +//! DataFusion simplifies `NOT (s = s)` to `s IS NULL AND NULL`. The bare `NULL` +//! is a conjunct with an empty column set, and the row-filter builder used to +//! drop every candidate whose column set was empty. The liquid scan is the only +//! place the pushed-down predicate is applied — DataFusion has already removed +//! the `FilterExec` — so dropping a conjunct *widens* the filter: rows where the +//! predicate is UNKNOWN came back as if it were TRUE. +//! +//! That shows up as a ternary-logic partitioning violation. `WHERE p`, +//! `WHERE NOT p` and `WHERE p IS NULL` must partition the table, since every row +//! satisfies exactly one of the three; with the conjunct dropped the three +//! buckets returned more rows than the table holds. + +use std::path::Path; +use std::sync::Arc; + +use arrow::array::{Array, Float64Array, Int64Array, StringArray}; +use arrow::record_batch::RecordBatch; +use arrow_schema::{DataType, Field, Schema}; +use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; +use parquet::arrow::ArrowWriter; +use tempfile::TempDir; + +use crate::LiquidCacheLocalBuilder; + +/// The predicate from the issue. Over [`write_t1`]'s data it is TRUE for eight +/// rows, UNKNOWN for four, and FALSE for none: +/// +/// - `id BETWEEN 1 AND 7` is FALSE everywhere — `id` starts at 10. +/// - `f <= 333.0` is TRUE for the first four rows only. +/// - `s = s` is TRUE where `s` is set and UNKNOWN where it is NULL. +/// +/// So the last four NULL-`s` rows are UNKNOWN, and belong to the `p IS NULL` +/// bucket alone. +const P: &str = "(f <= 333.0 OR s = s) OR id BETWEEN 1 AND 7"; + +/// Twelve rows. `s` is NULL on even rows, `f` climbs past the 333 threshold at +/// row four, and `id` stays clear of the 1..7 range. +fn write_t1(path: &Path) { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("f", DataType::Float64, false), + Field::new("s", DataType::Utf8, true), + ])); + + let id: Int64Array = (0..12).map(|i| i + 10).collect(); + let f: Float64Array = (0..12).map(|i| i as f64 * 100.0).collect(); + let s: StringArray = (0..12) + .map(|i| (i % 2 == 1).then(|| format!("s{i}"))) + .collect(); + + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(id), Arc::new(f), Arc::new(s)], + ) + .unwrap(); + let file = std::fs::File::create(path).unwrap(); + let mut writer = ArrowWriter::try_new(file, schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); +} + +async fn liquid_ctx(cache_dir: &Path, parquet: &Path) -> SessionContext { + std::fs::create_dir_all(cache_dir).unwrap(); + let (ctx, _cache) = LiquidCacheLocalBuilder::new() + .with_cache_dir(cache_dir.to_path_buf()) + .build(SessionConfig::new()) + .await + .unwrap(); + ctx.register_parquet( + "t1", + parquet.to_str().unwrap(), + ParquetReadOptions::default(), + ) + .await + .unwrap(); + ctx +} + +/// The `s` values matching `where`, sorted, with NULL spelled out. +async fn projected_rows(ctx: &SessionContext, r#where: &str) -> Vec { + let batches = ctx + .sql(&format!("SELECT s FROM t1 {}", r#where)) + .await + .unwrap() + .collect() + .await + .unwrap(); + let mut rows = Vec::new(); + for batch in batches { + let column = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + for i in 0..column.len() { + rows.push(match column.is_valid(i) { + true => column.value(i).to_string(), + false => "NULL".to_string(), + }); + } + } + rows.sort(); + rows +} + +/// The scan projects `s` while the filter reads `f`, `s` and `id`, so the +/// predicate columns are materialized separately from the projected one. +/// +/// `NOT p` is the telling bucket: it is never TRUE, because `p` is TRUE or +/// UNKNOWN for every row. Dropping the column-free `NULL` conjunct left +/// `f > 333 AND s IS NULL AND id NOT BETWEEN 1 AND 7` behind, which matches the +/// four UNKNOWN rows — so they came back in two buckets at once. +#[tokio::test] +async fn ternary_partitions_cover_every_row_once() { + let dir = TempDir::new().unwrap(); + let parquet = dir.path().join("t1.parquet"); + write_t1(&parquet); + let ctx = liquid_ctx(&dir.path().join("cache"), &parquet).await; + + // The first query reads the predicate columns through the parquet fallback + // and fills the cache; everything after it is served from the cache, which + // is a separate evaluation path. Running the buckets twice gives each one a + // warm pass, and the first one a cold pass. + for pass in ["fills-cache", "cached"] { + let matched = projected_rows(&ctx, &format!("WHERE {P}")).await; + let negated = projected_rows(&ctx, &format!("WHERE NOT ({P})")).await; + let unknown = projected_rows(&ctx, &format!("WHERE ({P}) IS NULL")).await; + + assert_eq!(matched.len(), 8, "{pass}"); + assert_eq!(negated, Vec::::new(), "{pass}"); + assert_eq!(unknown, vec!["NULL"; 4], "{pass}"); + + let mut partitioned = matched; + partitioned.extend(negated); + partitioned.extend(unknown); + partitioned.sort(); + assert_eq!( + projected_rows(&ctx, "").await, + partitioned, + "{pass}: the three buckets are not a partition of the table" + ); + } +} diff --git a/src/datafusion-local/src/tests/mod.rs b/src/datafusion-local/src/tests/mod.rs index 2f6f0e6e..2feff914 100644 --- a/src/datafusion-local/src/tests/mod.rs +++ b/src/datafusion-local/src/tests/mod.rs @@ -23,6 +23,7 @@ use datafusion::{ use crate::LiquidCacheLocalBuilder; mod batch_size_alignment; +mod column_free_conjunct; mod date_optimizer; mod filter_limit; mod page_index; diff --git a/src/datafusion/src/cache/mod.rs b/src/datafusion/src/cache/mod.rs index 4dbb1f1d..814d0d32 100644 --- a/src/datafusion/src/cache/mod.rs +++ b/src/datafusion/src/cache/mod.rs @@ -5,7 +5,7 @@ use crate::io::ParquetCacheMetadata; use crate::reader::{LiquidPredicate, extract_multi_column_or}; use crate::sync::Mutex; use ahash::AHashMap; -use arrow::array::{BooleanArray, RecordBatch}; +use arrow::array::{BooleanArray, RecordBatch, RecordBatchOptions}; use arrow::buffer::BooleanBuffer; use arrow_schema::{ArrowError, Field, Schema, SchemaRef}; use liquid_cache::cache::squeeze_policies::SqueezePolicy; @@ -173,9 +173,14 @@ impl CachedRowGroup { fields.push(column.field()); } let schema = Arc::new(Schema::new(fields)); - let record_batch = RecordBatch::try_new(schema, arrays).unwrap(); - let boolean_array = predicate.evaluate(record_batch).unwrap(); - Some(Ok(boolean_array)) + // The row count has to be carried explicitly: a column-free conjunct + // (`NULL`, `false`) projects no arrays, and an array-less batch would + // otherwise claim zero rows. + let options = RecordBatchOptions::new().with_row_count(Some(selection.count_set_bits())); + Some( + RecordBatch::try_new_with_options(schema, arrays, &options) + .and_then(|batch| predicate.evaluate(batch)), + ) } } @@ -432,7 +437,7 @@ mod tests { use super::*; use crate::cache::{CachedRowGroupRef, LiquidCacheParquet}; use crate::reader::FilterCandidateBuilder; - use arrow::array::Int32Array; + use arrow::array::{Array, Int32Array}; use arrow::buffer::BooleanBuffer; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; @@ -465,6 +470,54 @@ mod tests { file.create_row_group(0, vec![]) } + /// Issue #19: `NOT (s = s)` simplifies to `s IS NULL AND NULL`, so a conjunct + /// that reads no column reaches the row filter. It has to survive candidate + /// building and then evaluate against the selection's row count — an + /// array-less batch would otherwise report zero rows and hand back a mask of + /// the wrong length, which silently widens the filter. + #[tokio::test] + async fn evaluate_column_free_conjunct() { + let batch_size = 8; + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])); + let row_group = setup_cache(batch_size, schema.clone()).await; + + let array = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6, 7, 8])); + let batch_id = BatchID::from_row_id(0, batch_size); + let column = row_group.get_column(0).unwrap(); + assert!(column.insert(batch_id, array.clone()).await.is_ok()); + + let tmp_meta = tempfile::NamedTempFile::new().unwrap(); + let mut writer = + ArrowWriter::try_new(tmp_meta.reopen().unwrap(), Arc::clone(&schema), None).unwrap(); + let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array]).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let file_reader = std::fs::File::open(tmp_meta.path()).unwrap(); + let metadata = ArrowReaderMetadata::load(&file_reader, ArrowReaderOptions::new()).unwrap(); + + let expr: Arc = Arc::new(Literal::new(ScalarValue::Boolean(None))); + let builder = FilterCandidateBuilder::new(expr, Arc::clone(&schema)); + let candidate = builder + .build(metadata.metadata()) + .unwrap() + .expect("a column-free conjunct must still produce a candidate"); + let projection = candidate.projection(metadata.metadata()); + let mut predicate = LiquidPredicate::try_new(candidate, projection).unwrap(); + assert!(predicate.predicate_column_ids().is_empty()); + + // Four of the eight rows are selected, so the mask must be four long. + let selection = BooleanBuffer::collect_bool(batch_size, |i| i % 2 == 0); + let result = row_group + .evaluate_selection_with_predicate(batch_id, &selection, &mut predicate) + .await + .unwrap() + .unwrap(); + + assert_eq!(result.len(), selection.count_set_bits()); + assert_eq!(result.true_count(), 0); + assert_eq!(result.null_count(), result.len()); + } + #[tokio::test] async fn evaluate_or_on_cached_columns() { let batch_size = 4; diff --git a/src/datafusion/src/reader/plantime/row_filter.rs b/src/datafusion/src/reader/plantime/row_filter.rs index fe9c5ea1..3412cfd9 100644 --- a/src/datafusion/src/reader/plantime/row_filter.rs +++ b/src/datafusion/src/reader/plantime/row_filter.rs @@ -71,7 +71,7 @@ use arrow_schema::SchemaRef; use datafusion::datasource::physical_plan::ParquetFileMetrics; use datafusion::logical_expr::Operator; use datafusion::physical_expr::utils::reassign_expr_columns; -use datafusion::physical_plan::expressions::{BinaryExpr, LikeExpr}; +use datafusion::physical_plan::expressions::{BinaryExpr, LikeExpr, Literal}; use datafusion::physical_plan::metrics; use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::ArrowPredicate; @@ -289,10 +289,9 @@ impl FilterCandidateBuilder { return Ok(None); }; - if required_indices_into_file_schema.is_empty() { - return Ok(None); - } - + // A conjunct that references no column (`NULL`, `false`, a volatile + // function) is still a conjunct: dropping it here would widen the filter, + // because the scan is the only place the predicate is applied. let projected_file_schema = Arc::new( self.file_schema .project(&required_indices_into_file_schema)?, @@ -418,13 +417,15 @@ fn columns_sorted(_columns: &[usize], _metadata: &ParquetMetaData) -> Result, physical_file_schema: &SchemaRef, @@ -499,17 +500,23 @@ pub fn build_row_filter( } fn get_priority(expr: &Arc) -> u8 { + // A constant conjunct reads no column and can empty the selection outright, + // which skips every predicate after it, so it goes first. + if expr.is::() { + return 0; + } + if let Some(binary) = expr.downcast_ref::() { match binary.op() { - Operator::Eq | Operator::NotEq => 0, // Highest priority - Operator::LikeMatch | Operator::ILikeMatch => 1, - Operator::NotLikeMatch | Operator::NotILikeMatch => 2, - Operator::Lt | Operator::LtEq | Operator::Gt | Operator::GtEq => 3, - _ => 4, + Operator::Eq | Operator::NotEq => 1, + Operator::LikeMatch | Operator::ILikeMatch => 2, + Operator::NotLikeMatch | Operator::NotILikeMatch => 3, + Operator::Lt | Operator::LtEq | Operator::Gt | Operator::GtEq => 4, + _ => 5, } } else if expr.is::() { - 1 // LIKE expressions + 2 // LIKE expressions } else { - 5 // All other expression types + 6 // All other expression types } }