From 46b508eb4019868a1d4ee676f6604f298cb3a862 Mon Sep 17 00:00:00 2001 From: Saad Tajwar <59696464+saadtajwar@users.noreply.github.com> Date: Mon, 22 Jun 2026 20:35:16 -0400 Subject: [PATCH 1/3] perf: optimize object store requests when reading CSV (#22962) ## Which issue does this PR close? - Closes #21419 ## Rationale for this change The CSV scanner currently uses `calculate_range` which issues two extra `get_opts` requests per byte range to find newline boundaries (one for the start boundary, one for the end boundary), plus one GET for the actual data. For a file split into 3 partitions, this results in 8 total object store requests. #20823 solved this same problem for the JSON scanner by introducing `AlignedBoundaryStream`, which wraps the raw byte stream and lazily aligns to newline boundaries as data is read, eliminating the extra boundary-seeking requests entirely. This PR applies the same approach to CSV. ## What changes are included in this PR? Based on the approach from #20823: Moved `AlignedBoundaryStream` from `datasource-json` to the shared `datasource` crate so it can be reused by both JSON and CSV scanners. Updated `CsvOpener` to use instead of `calculate_range`, and removed the `calculate_range` & `find_first_newline` as they no longer had any callers. Updated tests to reflect. Public API changes include: - Removal of the `RangeCalculation` enum - Removal of the `calculate_change` function - `boundary_stream` (containing `AlignedBoundaryStream`) moved from `datafusion_datasource_json` to `datafusion_datasource` ## Are these changes tested? Yes. The existing `AlignedBoundaryStream` unit tests (16 tests covering boundary alignment edge cases) were moved along with the implementation and continue to pass. The `query_csv_file_with_byte_range_partitions` snapshot test in `object_store_access.rs` has been updated to verify the new request pattern (4 requests instead of 8). ## Are there any user-facing changes? No. --- .../tests/datasource/object_store_access.rs | 28 +--- datafusion/datasource-csv/src/source.rs | 85 +++++----- datafusion/datasource-json/src/mod.rs | 1 - datafusion/datasource-json/src/source.rs | 3 +- .../src/boundary_stream.rs | 4 +- datafusion/datasource/src/mod.rs | 149 +----------------- datafusion/datasource/src/test_util.rs | 29 ++++ 7 files changed, 90 insertions(+), 209 deletions(-) rename datafusion/{datasource-json => datasource}/src/boundary_stream.rs (99%) diff --git a/datafusion/core/tests/datasource/object_store_access.rs b/datafusion/core/tests/datasource/object_store_access.rs index 25150ae284cc0..2503de862e06a 100644 --- a/datafusion/core/tests/datasource/object_store_access.rs +++ b/datafusion/core/tests/datasource/object_store_access.rs @@ -231,23 +231,13 @@ async fn query_multi_csv_file() { ); } -/// Test that a CSV file split into byte ranges via repartitioning exercises -/// range-based object store access. +/// Test that a CSV file split into byte ranges via repartitioning produces +/// exactly one GET request per byte range — no extra requests for boundary seeking. /// /// With a single file and `target_partitions=3`, the repartitioner produces -/// exactly 3 ranges. For each range, `calculate_range` calls -/// `find_first_newline` via a GET for every non-file boundary it touches -/// (the start boundary if `start > 0`, the end boundary if `end < file_size`), -/// plus one GET for the actual data — so 2 GETs for the first range (end scan -/// + data), 3 for the middle range (start scan + end scan + data), and 2 for -/// the last range (start scan + data) = 7 data GETs total. Additionally, -/// adjacent ranges share a boundary position, so each shared boundary is scanned -/// twice — once as the left range's end and again as the right range's start — -/// producing the duplicate GETs visible in the snapshot. Add the 1 HEAD for -/// file-size metadata = **8 total**. -/// -/// This differs from the JSON reader which uses [`AlignedBoundaryStream`] and -/// needs only 1 GET per range. +/// exactly 3 ranges. Each range is served by a single [`AlignedBoundaryStream`] +/// which issues exactly one bounded `get_opts` call, so there are 3 data GETs +/// plus 1 HEAD (to determine file size) = **4 total**. /// /// This test documents the current request pattern to catch regressions. #[tokio::test] @@ -275,15 +265,11 @@ async fn query_csv_file_with_byte_range_partitions() { +---------+-------+-------+ ------- Object Store Request Summary ------- RequestCountingObjectStore() - Total Requests: 8 + Total Requests: 4 - GET (opts) path=csv_range_table.csv head=true + - GET (opts) path=csv_range_table.csv range=0-129 - GET (opts) path=csv_range_table.csv range=42-129 - - GET (opts) path=csv_range_table.csv range=0-49 - - GET (opts) path=csv_range_table.csv range=42-129 - - GET (opts) path=csv_range_table.csv range=85-129 - - GET (opts) path=csv_range_table.csv range=49-89 - GET (opts) path=csv_range_table.csv range=85-129 - - GET (opts) path=csv_range_table.csv range=89-129 " ); } diff --git a/datafusion/datasource-csv/src/source.rs b/datafusion/datasource-csv/src/source.rs index 638279f827344..25ec311880405 100644 --- a/datafusion/datasource-csv/src/source.rs +++ b/datafusion/datasource-csv/src/source.rs @@ -17,24 +17,23 @@ //! Execution plan for reading CSV files +use datafusion_datasource::boundary_stream::AlignedBoundaryStream; use datafusion_datasource::projection::{ProjectionOpener, SplitProjection}; use datafusion_physical_plan::projection::ProjectionExprs; use std::fmt; -use std::io::{Read, Seek, SeekFrom}; +use std::io::Read; use std::sync::Arc; -use std::task::Poll; use datafusion_datasource::decoder::{DecoderDeserializer, deserialize_stream}; use datafusion_datasource::file_compression_type::FileCompressionType; use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener}; use datafusion_datasource::{ - FileRange, ListingTableUrl, PartitionedFile, RangeCalculation, TableSchema, - as_file_source, calculate_range, + FileRange, ListingTableUrl, PartitionedFile, TableSchema, as_file_source, }; use arrow::csv; use datafusion_common::config::CsvOptions; -use datafusion_common::{DataFusionError, Result}; +use datafusion_common::{DataFusionError, Result, exec_datafusion_err}; use datafusion_common_runtime::JoinSet; use datafusion_datasource::file::FileSource; use datafusion_datasource::file_scan_config::FileScanConfig; @@ -367,43 +366,53 @@ impl FileOpener for CsvOpener { Ok(Box::pin(async move { // Current partition contains bytes [start_byte, end_byte) (might contain incomplete lines at boundaries) + let file_size = partitioned_file.object_meta.size; + let location = partitioned_file.object_meta.location; + + if let Some(file_range) = partitioned_file.range.as_ref() { + let raw_start: u64 = file_range.start.try_into().map_err(|_| { + exec_datafusion_err!( + "Expected start range to fit in u64, got {}", + file_range.start + ) + })?; + let raw_end: u64 = file_range.end.try_into().map_err(|_| { + exec_datafusion_err!( + "Expected end range to fit in u64, got {}", + file_range.end + ) + })?; + + let aligned_stream = AlignedBoundaryStream::new( + Arc::clone(&store), + location.clone(), + raw_start, + raw_end, + file_size, + terminator.unwrap_or(b'\n'), + ) + .await? + .map_err(DataFusionError::from); + + let decoder = config.builder().build_decoder(); + let input = file_compression_type + .convert_stream(aligned_stream.boxed())? + .fuse(); + let stream = deserialize_stream( + input, + DecoderDeserializer::new(CsvDecoder::new(decoder)), + ); + return Ok(stream.map_err(Into::into).boxed()); + } - let calculated_range = - calculate_range(&partitioned_file, &store, terminator).await?; - - let range = match calculated_range { - RangeCalculation::Range(None) => None, - RangeCalculation::Range(Some(range)) => Some(range.into()), - RangeCalculation::TerminateEarly => { - return Ok( - futures::stream::poll_fn(move |_| Poll::Ready(None)).boxed() - ); - } - }; - - let options = GetOptions { - range, - ..Default::default() - }; - - let result = store - .get_opts(&partitioned_file.object_meta.location, options) - .await?; + // No range specified — read the entire file + let options = GetOptions::default(); + let result = store.get_opts(&location, options).await?; match result.payload { #[cfg(not(target_arch = "wasm32"))] - GetResultPayload::File(mut file, _) => { - let is_whole_file_scanned = partitioned_file.range.is_none(); - let decoder = if is_whole_file_scanned { - // Don't seek if no range as breaks FIFO files - file_compression_type.convert_read(file)? - } else { - file.seek(SeekFrom::Start(result.range.start as _))?; - file_compression_type.convert_read( - file.take((result.range.end - result.range.start) as u64), - )? - }; - + GetResultPayload::File(file, _) => { + let decoder = file_compression_type.convert_read(file)?; let mut reader = config.open(decoder)?; // Use std::iter::from_fn to wrap execution of iterator's next() method. diff --git a/datafusion/datasource-json/src/mod.rs b/datafusion/datasource-json/src/mod.rs index f7932c8a21d95..ec93fc9c387e2 100644 --- a/datafusion/datasource-json/src/mod.rs +++ b/datafusion/datasource-json/src/mod.rs @@ -20,7 +20,6 @@ // https://github.com/apache/datafusion/issues/11143 #![cfg_attr(not(test), deny(clippy::clone_on_ref_ptr))] -pub mod boundary_stream; pub mod file_format; pub mod source; pub mod utils; diff --git a/datafusion/datasource-json/src/source.rs b/datafusion/datasource-json/src/source.rs index 179870673d426..8632d6b942bc1 100644 --- a/datafusion/datasource-json/src/source.rs +++ b/datafusion/datasource-json/src/source.rs @@ -25,11 +25,10 @@ use std::task::{Context, Poll}; use crate::file_format::JsonDecoder; use crate::utils::{ChannelReader, JsonArrayToNdjsonReader}; -use crate::boundary_stream::AlignedBoundaryStream; - use datafusion_common::error::{DataFusionError, Result}; use datafusion_common::exec_datafusion_err; use datafusion_common_runtime::{JoinSet, SpawnedTask}; +use datafusion_datasource::boundary_stream::AlignedBoundaryStream; use datafusion_datasource::decoder::{DecoderDeserializer, deserialize_stream}; use datafusion_datasource::file_compression_type::FileCompressionType; use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener}; diff --git a/datafusion/datasource-json/src/boundary_stream.rs b/datafusion/datasource/src/boundary_stream.rs similarity index 99% rename from datafusion/datasource-json/src/boundary_stream.rs rename to datafusion/datasource/src/boundary_stream.rs index 847c80279a53e..7b1cfb814df31 100644 --- a/datafusion/datasource-json/src/boundary_stream.rs +++ b/datafusion/datasource/src/boundary_stream.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! Streaming boundary-aligned wrapper for newline-delimited JSON range reads. +//! Streaming boundary-aligned wrapper for newline-delimited JSON and CSV range reads. //! //! [`AlignedBoundaryStream`] wraps a raw byte stream and lazily aligns to //! record (newline) boundaries, avoiding the need for separate `get_opts` @@ -398,7 +398,7 @@ impl Stream for AlignedBoundaryStream { #[cfg(test)] mod tests { use super::*; - use crate::test_utils::{CHUNK_SIZES, make_chunked_store}; + use crate::test_util::{CHUNK_SIZES, make_chunked_store}; use futures::TryStreamExt; async fn collect_stream(stream: AlignedBoundaryStream) -> Vec { diff --git a/datafusion/datasource/src/mod.rs b/datafusion/datasource/src/mod.rs index 82030e545a42e..d4dfa1180ecf2 100644 --- a/datafusion/datasource/src/mod.rs +++ b/datafusion/datasource/src/mod.rs @@ -28,6 +28,7 @@ //! A table that uses the `ObjectStore` listing capability //! to get the list of files to process. +pub mod boundary_stream; pub mod decoder; pub mod display; pub mod file; @@ -56,11 +57,10 @@ pub use self::url::ListingTableUrl; use crate::file_groups::FileGroup; use chrono::TimeZone; use datafusion_common::stats::Precision; -use datafusion_common::{ColumnStatistics, Result, TableReference, exec_datafusion_err}; +use datafusion_common::{ColumnStatistics, Result, TableReference}; use datafusion_common::{ScalarValue, Statistics}; use datafusion_physical_expr::LexOrdering; -use futures::{Stream, StreamExt}; -use object_store::{GetOptions, GetRange, ObjectStore}; +use futures::Stream; use object_store::{ObjectMeta, path::Path}; pub use table_schema::{TableSchema, TableSchemaBuilder}; // Remove when add_row_stats is remove @@ -69,7 +69,6 @@ use arrow::datatypes::SchemaRef; pub use statistics::add_row_stats; pub use statistics::compute_all_files_statistics; use std::any::Any; -use std::ops::Range; use std::pin::Pin; use std::sync::Arc; @@ -411,119 +410,6 @@ impl From for PartitionedFile { } } -/// Represents the possible outcomes of a range calculation. -/// -/// This enum is used to encapsulate the result of calculating the range of -/// bytes to read from an object (like a file) in an object store. -/// -/// Variants: -/// - `Range(Option>)`: -/// Represents a range of bytes to be read. It contains an `Option` wrapping a -/// `Range`. `None` signifies that the entire object should be read, -/// while `Some(range)` specifies the exact byte range to read. -/// - `TerminateEarly`: -/// Indicates that the range calculation determined no further action is -/// necessary, possibly because the calculated range is empty or invalid. -pub enum RangeCalculation { - Range(Option>), - TerminateEarly, -} - -/// Calculates an appropriate byte range for reading from an object based on the -/// provided metadata. -/// -/// This asynchronous function examines the [`PartitionedFile`] of an object in an object store -/// and determines the range of bytes to be read. The range calculation may adjust -/// the start and end points to align with meaningful data boundaries (like newlines). -/// -/// Returns a `Result` wrapping a [`RangeCalculation`], which is either a calculated byte range or an indication to terminate early. -/// -/// Returns an `Error` if any part of the range calculation fails, such as issues in reading from the object store or invalid range boundaries. -pub async fn calculate_range( - file: &PartitionedFile, - store: &Arc, - terminator: Option, -) -> Result { - let location = &file.object_meta.location; - let file_size = file.object_meta.size; - let newline = terminator.unwrap_or(b'\n'); - - match file.range { - None => Ok(RangeCalculation::Range(None)), - Some(FileRange { start, end }) => { - let start: u64 = start.try_into().map_err(|_| { - exec_datafusion_err!("Expect start range to fit in u64, got {start}") - })?; - let end: u64 = end.try_into().map_err(|_| { - exec_datafusion_err!("Expect end range to fit in u64, got {end}") - })?; - - let start_delta = if start != 0 { - find_first_newline(store, location, start - 1, file_size, newline).await? - } else { - 0 - }; - - if start + start_delta > end { - return Ok(RangeCalculation::TerminateEarly); - } - - let end_delta = if end != file_size { - find_first_newline(store, location, end - 1, file_size, newline).await? - } else { - 0 - }; - - let range = start + start_delta..end + end_delta; - - if range.start >= range.end { - return Ok(RangeCalculation::TerminateEarly); - } - - Ok(RangeCalculation::Range(Some(range))) - } - } -} - -/// Asynchronously finds the position of the first newline character in a specified byte range -/// within an object, such as a file, in an object store. -/// -/// This function scans the contents of the object starting from the specified `start` position -/// up to the `end` position, looking for the first occurrence of a newline character. -/// It returns the position of the first newline relative to the start of the range. -/// -/// Returns a `Result` wrapping a `usize` that represents the position of the first newline character found within the specified range. If no newline is found, it returns the length of the scanned data, effectively indicating the end of the range. -/// -/// The function returns an `Error` if any issues arise while reading from the object store or processing the data stream. -async fn find_first_newline( - object_store: &Arc, - location: &Path, - start: u64, - end: u64, - newline: u8, -) -> Result { - let options = GetOptions { - range: Some(GetRange::Bounded(start..end)), - ..Default::default() - }; - - let result = object_store.get_opts(location, options).await?; - let mut result_stream = result.into_stream(); - - let mut index = 0; - - while let Some(chunk) = result_stream.next().await.transpose()? { - if let Some(position) = chunk.iter().position(|&byte| byte == newline) { - let position = position as u64; - return Ok(index + position); - } - - index += chunk.len() as u64; - } - - Ok(index) -} - /// Generates test files with min-max statistics in different overlap patterns. /// /// Used by tests and benchmarks. @@ -647,7 +533,7 @@ mod tests { use datafusion_execution::object_store::{ DefaultObjectStoreRegistry, ObjectStoreRegistry, }; - use object_store::{ObjectStoreExt, local::LocalFileSystem, path::Path}; + use object_store::{local::LocalFileSystem, path::Path}; use std::{collections::HashMap, ops::Not, sync::Arc}; use url::Url; @@ -844,31 +730,4 @@ mod tests { // testing an empty path with `ignore_subdirectory` set to false assert!(url.contains(&Path::parse("/var/data/mytable/").unwrap(), false)); } - - /// Regression test for - #[tokio::test] - async fn test_calculate_range_single_line_file() { - use super::{PartitionedFile, RangeCalculation, calculate_range}; - use object_store::ObjectStore; - use object_store::memory::InMemory; - - let content = r#"{"id":1,"data":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}"#; - let file_size = content.len() as u64; - - let store: Arc = Arc::new(InMemory::new()); - let path = Path::from("test.json"); - store.put(&path, content.into()).await.unwrap(); - - let mid = file_size / 2; - let partitioned_file = PartitionedFile::new_with_range( - path.to_string(), - file_size, - mid as i64, - file_size as i64, - ); - - let result = calculate_range(&partitioned_file, &store, None).await; - - assert!(matches!(result, Ok(RangeCalculation::TerminateEarly))); - } } diff --git a/datafusion/datasource/src/test_util.rs b/datafusion/datasource/src/test_util.rs index d35ed5feb51de..5bef6d44e1408 100644 --- a/datafusion/datasource/src/test_util.rs +++ b/datafusion/datasource/src/test_util.rs @@ -131,3 +131,32 @@ impl FileSource for MockSource { pub(crate) fn col(name: &str, schema: &Schema) -> Result> { Ok(Arc::new(Column::new_with_schema(name, schema)?)) } + +/// Chunk sizes exercised by every parameterised test. +/// +/// `usize::MAX` is intentionally included: `ChunkedStore` treats it as +/// "one chunk containing everything", giving the single-chunk fast path. +pub(crate) const CHUNK_SIZES: &[usize] = &[1, 2, 3, 4, 5, 7, 8, 11, 13, 16, usize::MAX]; + +/// Seed a fresh `InMemory` store with `data` and wrap it in a +/// [`ChunkedStore`] that splits every GET response into `chunk_size`-byte +/// pieces. +pub(crate) async fn make_chunked_store( + data: &[u8], + chunk_size: usize, +) -> (Arc, object_store::path::Path) { + use bytes::Bytes; + use object_store::ObjectStoreExt; + use object_store::PutPayload; + use object_store::chunked::ChunkedStore; + use object_store::memory::InMemory; + use object_store::path::Path; + + let inner = Arc::new(InMemory::new()); + let path = Path::from("test"); + inner + .put(&path, PutPayload::from(Bytes::copy_from_slice(data))) + .await + .unwrap(); + (Arc::new(ChunkedStore::new(inner, chunk_size)), path) +} From e2960893595b422ec99f7e706b81e80566e4ed58 Mon Sep 17 00:00:00 2001 From: Moe Date: Mon, 22 Jun 2026 16:46:33 -0700 Subject: [PATCH 2/3] Kept null-aware anti-join NULLs in the pushed dynamic filter. The hash-join dynamic filter pushed `key IN build_keys` down to the probe scan for null-aware anti joins too. That drops the probe-side NULL, but `NOT IN` three-valued logic needs it to collapse the result to zero rows, so the join silently returned rows. OR `probe_key IS NULL` into the pushed predicate. Non-NULL probe rows still get filtered; only the NULL additionally survives. --- .../physical-plan/src/joins/hash_join/exec.rs | 1 + .../src/joins/hash_join/shared_bounds.rs | 36 +++++++++++++++++-- 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index b1d387ea74557..a11b89175f3eb 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1363,6 +1363,7 @@ impl ExecutionPlan for HashJoinExec { filter, on_right, repartition_random_state, + self.null_aware, )) }))) }) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 0af4015ff7239..2658a7020cd52 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -37,7 +37,7 @@ use datafusion_common::{DataFusionError, Result, ScalarValue, SharedResult}; use datafusion_expr::Operator; use datafusion_functions::core::r#struct as struct_func; use datafusion_physical_expr::expressions::{ - BinaryExpr, CaseExpr, DynamicFilterPhysicalExpr, InListExpr, lit, + BinaryExpr, CaseExpr, DynamicFilterPhysicalExpr, InListExpr, IsNullExpr, lit, }; use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef, ScalarFunctionExpr}; @@ -255,6 +255,9 @@ pub(crate) struct SharedBuildAccumulator { repartition_random_state: SeededRandomState, /// Schema of the probe (right) side for evaluating filter expressions probe_schema: Arc, + /// Null-aware anti join (`NOT IN`). A probe-side NULL must reach the join so its + /// three-valued logic can collapse the result, so the pushed filter keeps NULL rows. + null_aware: bool, } /// Strategy for filter pushdown (decided at collection time) @@ -358,6 +361,7 @@ impl SharedBuildAccumulator { dynamic_filter: Arc, on_right: Vec, repartition_random_state: SeededRandomState, + null_aware: bool, ) -> Self { // Troubleshooting: If partition counts are incorrect, verify this logic matches // the actual execution pattern in collect_build_side() @@ -404,6 +408,7 @@ impl SharedBuildAccumulator { on_right, repartition_random_state, probe_schema: right_child.schema(), + null_aware, } } @@ -579,7 +584,8 @@ impl SharedBuildAccumulator { if let Some(filter_expr) = combine_membership_and_bounds(membership_expr, bounds_expr) { - self.dynamic_filter.update(filter_expr)?; + self.dynamic_filter + .update(self.null_aware_filter(filter_expr))?; } } PartitionStatus::Pending => { @@ -685,12 +691,35 @@ impl SharedBuildAccumulator { )?) as Arc }; - self.dynamic_filter.update(filter_expr)?; + self.dynamic_filter + .update(self.null_aware_filter(filter_expr))?; } } Ok(()) } + + /// Wraps a pushdown filter so a null-aware anti join keeps its probe-side NULL rows. + /// + /// The build-side predicate drops probe rows whose key is NULL, but `NOT IN` three-valued + /// logic needs that NULL to reach the join. OR-ing `probe_key IS NULL` preserves the dynamic + /// filter's selectivity for non-NULL rows while letting the NULL through. + fn null_aware_filter( + &self, + filter_expr: Arc, + ) -> Arc { + if !self.null_aware { + return filter_expr; + } + // A null-aware anti join is validated to a single probe key. + let probe_key_is_null: Arc = + Arc::new(IsNullExpr::new(Arc::clone(&self.on_right[0]))); + Arc::new(BinaryExpr::new( + filter_expr, + Operator::Or, + probe_key_is_null, + )) + } } impl fmt::Debug for SharedBuildAccumulator { @@ -722,6 +751,7 @@ pub(super) fn make_partitioned_accumulator_for_test( on_right: vec![], repartition_random_state: SeededRandomState::with_seed(1), probe_schema, + null_aware: false, } } From f30d9bcae5a6f22f738b4da74519353243f48fde Mon Sep 17 00:00:00 2001 From: Moe Date: Mon, 22 Jun 2026 17:37:03 -0700 Subject: [PATCH 3/3] Added a null-aware anti-join dynamic-filter regression test. Exercises the pushdown path the existing in-memory tests miss: parquet with row-level filtering, so the pushed dynamic filter actually drops rows. Without the fix `id NOT IN (SELECT eid ...)` returns 1 and 3 instead of zero rows. --- .../test_files/null_aware_anti_join.slt | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt index b18f3b3ae7a99..3f04f8b865623 100644 --- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt +++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt @@ -451,3 +451,68 @@ DROP TABLE customers_test; statement ok DROP TABLE all_null_banned; + +############# +## Test: dynamic filter pushdown must not drop the inner NULL. +## With join dynamic filter pushdown on, the build-side filter pushed to the probe scan would drop +## the inner NULL, but NOT IN three-valued logic needs it to collapse the result to zero rows. The +## in-memory VALUES scans above never apply the pushed filter, so this case needs a parquet scan. +############# + +statement ok +set datafusion.optimizer.enable_join_dynamic_filter_pushdown = true; + +# Row-level parquet filtering, so the pushed filter actually drops matching rows instead of only +# pruning row groups. Without this the single row group is read whole and the NULL never gets dropped. +statement ok +set datafusion.execution.parquet.pushdown_filters = true; + +statement ok +CREATE TABLE asa_outer(id INT) AS VALUES (1), (2), (3); + +statement ok +CREATE TABLE asa_inner(eid INT) AS VALUES (2), (NULL); + +query I +COPY asa_outer TO 'test_files/scratch/null_aware_anti_join/asa_outer.parquet' STORED AS PARQUET; +---- +3 + +query I +COPY asa_inner TO 'test_files/scratch/null_aware_anti_join/asa_inner.parquet' STORED AS PARQUET; +---- +2 + +statement ok +CREATE EXTERNAL TABLE asa_outer_parquet(id INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/null_aware_anti_join/asa_outer.parquet'; + +statement ok +CREATE EXTERNAL TABLE asa_inner_parquet(eid INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/null_aware_anti_join/asa_inner.parquet'; + +# Expected: zero rows. Before the fix the pushed dynamic filter dropped the inner NULL, so the join +# wrongly returned id = 1 and id = 3. +query I +SELECT id FROM asa_outer_parquet WHERE id NOT IN (SELECT eid FROM asa_inner_parquet) ORDER BY id; +---- + +statement ok +DROP TABLE asa_outer; + +statement ok +DROP TABLE asa_inner; + +statement ok +DROP TABLE asa_outer_parquet; + +statement ok +DROP TABLE asa_inner_parquet; + +statement ok +set datafusion.execution.parquet.pushdown_filters = false; + +statement ok +set datafusion.optimizer.enable_join_dynamic_filter_pushdown = true;