From 0342ae41725adf1e9616cd43ea5adc41ae27e6e6 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:24:14 -0500 Subject: [PATCH 1/2] fix(query): enforce Exact time filters on in-memory staging and listing scans StandardTableProvider reports any ` {<,<=,>,>=} ` filter as Exact, for any column, so DataFusion drops the FilterExec above the scan. Parquet scans enforce these through the source predicate with pushdown_filters, but MemTable (in-memory staging records) and ListingTable (legacy listing) ignore pushed filters, so their rows were returned unfiltered, e.g. `WHERE created_at >= '2020-01-01'` or a minute-aligned p_timestamp bound inside the staging window. The listing scan also applied LIMIT before any filtering. Scan both through scan_enforcing_exact_filters, which scans the projected plus filter columns, applies the Exact conjunction in a FilterExec, then projects and applies the limit. Co-Authored-By: Claude Opus 5.5 --- src/query/stream_schema_provider.rs | 84 +++++- .../exact_filter_tests.rs | 248 ++++++++++++++++++ 2 files changed, 327 insertions(+), 5 deletions(-) create mode 100644 src/query/stream_schema_provider/exact_filter_tests.rs diff --git a/src/query/stream_schema_provider.rs b/src/query/stream_schema_provider.rs index 7dce6f732..b0df46a09 100644 --- a/src/query/stream_schema_provider.rs +++ b/src/query/stream_schema_provider.rs @@ -45,7 +45,9 @@ use datafusion::{ physical_planning_context::PhysicalPlanningContext, utils::conjunction, }, physical_expr::create_physical_expr, - physical_plan::{ExecutionPlan, Statistics, empty::EmptyExec, union::UnionExec}, + physical_plan::{ + ExecutionPlan, Statistics, empty::EmptyExec, filter::FilterExecBuilder, union::UnionExec, + }, prelude::Expr, scalar::ScalarValue, }; @@ -162,6 +164,68 @@ pub fn exact_source_filters(filters: &[Expr]) -> Vec { .collect() } +/// Scans `table`, whose scan ignores pushed filters, and enforces the filters +/// `StandardTableProvider` reports as `Exact`. +/// +/// DataFusion removes the `FilterExec` above the scan for `Exact` filters. Parquet +/// scans enforce them through the source predicate (see `exact_source_filters`), +/// but `MemTable` and `ListingTable` do not, so their rows are filtered here. The +/// limit is applied after the filter, and the filter's columns are scanned even +/// when they are not projected. +async fn scan_enforcing_exact_filters( + table: &dyn TableProvider, + state: &dyn Session, + projection: Option<&Vec>, + filters: &[Expr], + limit: Option, +) -> Result, DataFusionError> { + let Some(predicate) = conjunction(exact_source_filters(filters)) else { + return table.scan(state, projection, filters, limit).await; + }; + + let schema = table.schema(); + let scan_projection = projection + .map(|projection| { + let mut columns = projection.clone(); + for column in predicate.column_refs() { + columns.push(schema.index_of(&column.name)?); + } + columns.sort_unstable(); + columns.dedup(); + Ok::<_, DataFusionError>(columns) + }) + .transpose()?; + // Positions of the requested columns in the scan's output. + let output_projection = + projection + .zip(scan_projection.as_ref()) + .map(|(projection, scanned)| { + projection + .iter() + .map(|index| { + scanned + .binary_search(index) + .expect("projected column is scanned") + }) + .collect_vec() + }); + + let input = table + .scan(state, scan_projection.as_ref(), filters, None) + .await?; + let predicate = create_physical_expr( + &predicate, + &input.schema().to_dfschema()?, + state.execution_props(), + &PhysicalPlanningContext::default(), + )?; + let filter = FilterExecBuilder::new(predicate, input) + .apply_projection(output_projection)? + .with_fetch(limit) + .build()?; + Ok(Arc::new(filter)) +} + pub fn build_parquet_scan_components( schema: SchemaRef, filters: &[Expr], @@ -404,9 +468,9 @@ impl StandardTableProvider { ) -> Result<(), DataFusionError> { // Both sources were captured before the remote snapshot was read. let (records, partitioned_files) = captured; - let arrow_exec = reversed_mem_table(records, self.schema.clone())? - .scan(state, projection, filters, limit) - .await?; + let mem_table = reversed_mem_table(records, self.schema.clone())?; + let arrow_exec = + scan_enforcing_exact_filters(&mem_table, state, projection, filters, limit).await?; execution_plans.push(arrow_exec); // Keep the native scan and its file extensions: each extension owns the @@ -442,7 +506,14 @@ impl StandardTableProvider { let table = builder.build(self.schema.clone(), |x| glob_storage.query_prefixes(x))?; if let Some(table) = table { - let plan = table.scan(state, projection, filters, limit).await?; + let plan = scan_enforcing_exact_filters( + table.as_ref(), + state, + projection, + filters, + limit, + ) + .await?; execution_plans.push(plan); } @@ -1931,3 +2002,6 @@ mod tests { #[cfg(test)] mod ordering_tests; + +#[cfg(test)] +mod exact_filter_tests; diff --git a/src/query/stream_schema_provider/exact_filter_tests.rs b/src/query/stream_schema_provider/exact_filter_tests.rs new file mode 100644 index 000000000..30f2ae8a3 --- /dev/null +++ b/src/query/stream_schema_provider/exact_filter_tests.rs @@ -0,0 +1,248 @@ +/* + * Parseable Server (C) 2022 - 2025 Parseable, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + */ + +//! Filters `StandardTableProvider` reports as `Exact` must be enforced by every scan it unions. +//! +//! DataFusion drops the `FilterExec` above a scan for `Exact` filters. Parquet scans enforce them +//! through the source predicate, but `MemTable` (in-memory staging) and `ListingTable` (legacy +//! listing) ignore pushed filters, so their rows must be filtered by the provider. + +use std::{fs, path::PathBuf}; + +use arrow_array::{Int64Array, TimestampMillisecondArray}; +use arrow_schema::{DataType, Field, TimeUnit}; +use chrono::NaiveDate; +use datafusion::{ + physical_plan::collect, + prelude::{SessionConfig, SessionContext}, +}; +use parquet::arrow::ArrowWriter; +use temp_dir::TempDir; + +use super::*; +use crate::storage::LocalFS; + +/// A provider with `StandardTableProvider`'s filter pushdown and its scan of a table that ignores +/// pushed filters. +#[derive(Debug)] +struct ExactPushdownTable { + inner: Arc, +} + +#[async_trait::async_trait] +impl TableProvider for ExactPushdownTable { + fn schema(&self) -> SchemaRef { + self.inner.schema() + } + + fn table_type(&self) -> TableType { + TableType::Base + } + + async fn scan( + &self, + state: &dyn Session, + projection: Option<&Vec>, + filters: &[Expr], + limit: Option, + ) -> DataFusionResult> { + scan_enforcing_exact_filters(self.inner.as_ref(), state, projection, filters, limit).await + } + + fn supports_filters_pushdown( + &self, + filters: &[&Expr], + ) -> DataFusionResult> { + Ok(filters + .iter() + .map(|filter| { + if expr_in_boundary(filter) { + TableProviderFilterPushDown::Exact + } else { + TableProviderFilterPushDown::Inexact + } + }) + .collect()) + } +} + +fn millis(hour: u32, minute: u32, second: u32) -> i64 { + NaiveDate::from_ymd_opt(2026, 10, 17) + .unwrap() + .and_hms_opt(hour, minute, second) + .unwrap() + .and_utc() + .timestamp_millis() +} + +fn schema() -> SchemaRef { + let timestamp = DataType::Timestamp(TimeUnit::Millisecond, None); + Arc::new(Schema::new(vec![ + Field::new(DEFAULT_TIMESTAMP_KEY, timestamp.clone(), false), + Field::new("created_at", timestamp, false), + Field::new("id", DataType::Int64, false), + ])) +} + +/// Ids 0..4 are at 12:29:30 and created in 2019; ids 4..8 are at 12:30:30 and created in 2021. +fn batch(schema: &SchemaRef) -> RecordBatch { + let old_created = DateTime::parse_from_rfc3339("2019-06-01T00:00:00Z") + .unwrap() + .timestamp_millis(); + let new_created = DateTime::parse_from_rfc3339("2021-06-01T00:00:00Z") + .unwrap() + .timestamp_millis(); + let (times, created): (Vec<_>, Vec<_>) = (0..8) + .map(|id| { + if id < 4 { + (millis(12, 29, 30), old_created) + } else { + (millis(12, 30, 30), new_created) + } + }) + .unzip(); + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(TimestampMillisecondArray::from(times)), + Arc::new(TimestampMillisecondArray::from(created)), + Arc::new(Int64Array::from_iter_values(0..8)), + ], + ) + .unwrap() +} + +/// Queries whose time predicates `expr_in_boundary` reports as `Exact`, and the ids they select. +const QUERIES: [(&str, &[i64]); 5] = [ + ( + "SELECT id FROM t WHERE created_at >= '2020-01-01T00:00:00' ORDER BY id", + &[4, 5, 6, 7], + ), + ( + "SELECT id FROM t WHERE p_timestamp < '2026-10-17T12:30:00' ORDER BY id", + &[0, 1, 2, 3], + ), + ( + "SELECT id FROM t WHERE p_timestamp >= '2026-10-17T12:30:00' AND id < 6 ORDER BY id", + &[4, 5], + ), + ( + "SELECT p_timestamp, id FROM t WHERE created_at < '2020-01-01T00:00:00' ORDER BY id", + &[0, 1, 2, 3], + ), + ( + "SELECT * FROM t WHERE created_at > '2020-01-01T00:00:00' ORDER BY id", + &[4, 5, 6, 7], + ), +]; + +async fn ids(context: &SessionContext, sql: &str) -> Vec { + let plan = context + .sql(sql) + .await + .unwrap() + .create_physical_plan() + .await + .unwrap(); + let batches = collect(plan, context.task_ctx()).await.unwrap(); + batches + .iter() + .flat_map(|batch| { + let ids = batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + ids.values().to_vec() + }) + .collect() +} + +/// Checks `QUERIES`, a `LIMIT` that must count only matching rows, and a count without projected +/// columns against `table`. +async fn check(table: Arc) { + let mut config = SessionConfig::new(); + config.options_mut().execution.parquet.pushdown_filters = true; + let context = SessionContext::new_with_config(config); + context + .register_table("t", Arc::new(ExactPushdownTable { inner: table })) + .unwrap(); + let mut failures = Vec::new(); + for (sql, expected) in QUERIES { + let actual = ids(&context, sql).await; + if actual != expected { + failures.push(format!("{sql}: {actual:?}")); + } + } + let sql = "SELECT id FROM t WHERE created_at >= '2020-01-01T00:00:00' LIMIT 2"; + let limited = ids(&context, sql).await; + if limited.len() != 2 || limited.iter().any(|id| *id < 4) { + failures.push(format!("{sql}: {limited:?}")); + } + // No column is projected for the count. + let sql = "SELECT count(*) AS id FROM t WHERE created_at >= '2020-01-01T00:00:00'"; + let count = ids(&context, sql).await; + if count != [4] { + failures.push(format!("{sql}: {count:?}")); + } + assert!(failures.is_empty(), "{failures:#?}"); +} + +#[tokio::test] +async fn staging_memory_scan_enforces_exact_filters() { + let schema = schema(); + let table = reversed_mem_table(vec![batch(&schema)], schema).unwrap(); + check(Arc::new(table)).await; +} + +#[tokio::test] +async fn legacy_listing_scan_enforces_exact_filters() { + let schema = schema(); + let store = TempDir::new().unwrap(); + let stream_dir = store.path().join("exact"); + let minute_dir = stream_dir.join("date=2026-10-17/hour=12/minute=29"); + fs::create_dir_all(&minute_dir).unwrap(); + let file = fs::File::create(minute_dir.join("listed.data.parquet")).unwrap(); + let mut writer = ArrowWriter::try_new(file, schema.clone(), None).unwrap(); + writer.write(&batch(&schema)).unwrap(); + writer.close().unwrap(); + + // `LocalFS::query_prefixes` resolves prefixes against `/`, so the stream is named by its + // absolute path without the leading `/`. + let storage: Arc = Arc::new(LocalFS::new(PathBuf::from("/"))); + let midnight = |day| { + NaiveDate::from_ymd_opt(2026, 10, day) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap() + }; + let time_filters = [ + PartialTimeFilter::Low(Bound::Included(midnight(16))), + PartialTimeFilter::High(Bound::Excluded(midnight(18))), + ]; + let stream = stream_dir.to_str().unwrap()[1..].to_owned(); + let table = ListingTableBuilder::new(stream) + .populate_via_listing(storage.clone(), &time_filters) + .await + .unwrap() + .build(schema, |prefixes| storage.query_prefixes(prefixes)) + .unwrap() + .expect("the file is listed"); + check(table).await; +} From a93e80753a351299dd6e5ffeb999d3e8e18a7e6d Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:34:52 -0500 Subject: [PATCH 2/2] fix(query): report Exact only for minute-aligned time-partition bounds expr_in_boundary accepted a minute-aligned timestamp bound on any column, so supports_filters_pushdown reported e.g. `created_at >= '...T00:00:00'` as Exact. DataFusion then drops the FilterExec and pushes LIMIT into the scan, and collect_from_snapshot truncates the file list once the files' row counts reach the limit, assuming every row of a surviving file matches. For any column other than the one data files are partitioned by, that undercounts: `... LIMIT 100` could return 5 rows. Report Exact only when the filter's left side is the stream's time partition column (p_timestamp unless configured): data files hold one minute of that column, so a minute-aligned bound keeps or prunes each file whole. StandardTableProvider now carries the stream's time partition for this. Co-Authored-By: Claude Opus 5.5 --- src/query/stream_schema_provider.rs | 150 +++++++++++++++++++++++----- 1 file changed, 125 insertions(+), 25 deletions(-) diff --git a/src/query/stream_schema_provider.rs b/src/query/stream_schema_provider.rs index b0df46a09..4f5419e4f 100644 --- a/src/query/stream_schema_provider.rs +++ b/src/query/stream_schema_provider.rs @@ -96,13 +96,14 @@ impl SchemaProvider for GlobalSchemaProvider { async fn table(&self, name: &str) -> DataFusionResult>> { if self.table_exist(name) { + let stream = PARSEABLE + .get_stream(name, &self.tenant_id) + .expect(STREAM_EXISTS); Ok(Some(Arc::new(StandardTableProvider { - schema: PARSEABLE - .get_stream(name, &self.tenant_id) - .expect(STREAM_EXISTS) - .get_schema(), + schema: stream.get_schema(), stream: name.to_owned(), tenant_id: self.tenant_id.clone(), + time_partition: stream.get_time_partition(), hot_tier_guarded: false, }))) } else { @@ -121,6 +122,8 @@ struct StandardTableProvider { // prefix under which to find snapshot stream: String, tenant_id: Option, + // Column the stream's data files are partitioned by; `p_timestamp` when `None`. + time_partition: Option, hot_tier_guarded: bool, } @@ -152,6 +155,7 @@ pub(super) fn guarded_hot_tier_table_source( schema: provider.schema.clone(), stream: provider.stream.clone(), tenant_id: provider.tenant_id.clone(), + time_partition: provider.time_partition.clone(), hot_tier_guarded: true, }))) } @@ -165,7 +169,8 @@ pub fn exact_source_filters(filters: &[Expr]) -> Vec { } /// Scans `table`, whose scan ignores pushed filters, and enforces the filters -/// `StandardTableProvider` reports as `Exact`. +/// `StandardTableProvider` reports as `Exact` (`exact_source_filters` is a +/// superset of them: see `filter_pushdown`). /// /// DataFusion removes the `FilterExec` above the scan for `Exact` filters. Parquet /// scans enforce them through the source predicate (see `exact_source_filters`), @@ -989,20 +994,10 @@ impl TableProvider for StandardTableProvider { &self, filters: &[&Expr], ) -> Result, DataFusionError> { - let res_vec = filters + Ok(filters .iter() - .map(|filter| { - if expr_in_boundary(filter) { - // if filter can be handled by time partiton pruning, it is exact - TableProviderFilterPushDown::Exact - } else { - // otherwise, we still might be able to handle the filter with file - // level mechanisms such as Parquet row group pruning. - TableProviderFilterPushDown::Inexact - } - }) - .collect_vec(); - Ok(res_vec) + .map(|filter| filter_pushdown(filter, &self.time_partition)) + .collect_vec()) } } @@ -1187,6 +1182,30 @@ pub fn is_within_staging_window(time_filters: &[PartialTimeFilter]) -> bool { !has_upper_bound } +/// How `StandardTableProvider` can handle `filter`. +/// +/// Data files hold one minute of the stream's time partition column, so a +/// minute-aligned bound on that column either keeps or prunes each file whole. +/// Only then is the filter `Exact`: DataFusion drops the `FilterExec` above the +/// scan and may push `LIMIT` into it, and `collect_from_snapshot` truncates the +/// file list by row counts assuming every row of a surviving file matches. A +/// bound on any other column (including `p_timestamp` on a stream with a custom +/// time partition) only prunes files partially, so it stays `Inexact`. +fn filter_pushdown(filter: &Expr, time_partition: &Option) -> TableProviderFilterPushDown { + let partition_column = time_partition.as_deref().unwrap_or(DEFAULT_TIMESTAMP_KEY); + let is_partition_column = matches!( + filter, + Expr::BinaryExpr(BinaryExpr { left, .. }) + if matches!(left.as_ref(), Expr::Column(column) if column.name == partition_column) + ); + if is_partition_column && expr_in_boundary(filter) { + TableProviderFilterPushDown::Exact + } else { + // File level mechanisms such as Parquet row group pruning may still apply. + TableProviderFilterPushDown::Inexact + } +} + pub fn expr_in_boundary(filter: &Expr) -> bool { let Expr::BinaryExpr(binexpr) = filter else { return false; @@ -1408,6 +1427,7 @@ mod tests { schema: Arc::new(Schema::empty()), stream: "logs".to_owned(), tenant_id: Some("other-tenant".to_owned()), + time_partition: Some("event_time".to_owned()), hot_tier_guarded: false, })); @@ -1423,6 +1443,10 @@ mod tests { let guarded_provider = standard_table_provider(&guarded_source).unwrap(); assert_eq!(guarded_provider.stream, "logs"); assert_eq!(guarded_provider.tenant_id.as_deref(), Some("other-tenant")); + assert_eq!( + guarded_provider.time_partition.as_deref(), + Some("event_time") + ); assert!(guarded_provider.hot_tier_guarded); } @@ -1505,13 +1529,7 @@ mod tests { ) -> datafusion::error::Result> { Ok(filters .iter() - .map(|filter| { - if super::expr_in_boundary(filter) { - TableProviderFilterPushDown::Exact - } else { - TableProviderFilterPushDown::Inexact - } - }) + .map(|filter| super::filter_pushdown(filter, &None)) .collect()) } } @@ -1616,6 +1634,88 @@ mod tests { assert!(predicate.contains("api")); } + #[test] + fn only_minute_aligned_partition_column_bounds_are_exact() { + let minute = ScalarValue::TimestampMillisecond(Some(1_672_531_200_000), None); + let mid_minute = ScalarValue::TimestampMillisecond(Some(1_672_531_230_000), None); + let bound = |column: &str, value: &ScalarValue| { + Expr::Column(column.into()).gt_eq(Expr::Literal(value.clone(), None)) + }; + let event_time = Some("event_time".to_owned()); + + let cases = [ + (bound("p_timestamp", &minute), &None, true), + (bound("p_timestamp", &mid_minute), &None, false), + (bound("created_at", &minute), &None, false), + (bound("event_time", &minute), &event_time, true), + // Ingestion time on a stream partitioned by event time. + (bound("p_timestamp", &minute), &event_time, false), + ( + Expr::Literal(minute.clone(), None).lt(Expr::Column("p_timestamp".into())), + &None, + false, + ), + ]; + for (filter, time_partition, exact) in cases { + assert_eq!( + matches!( + super::filter_pushdown(&filter, time_partition), + TableProviderFilterPushDown::Exact + ), + exact, + "{filter} with time partition {time_partition:?}" + ); + } + } + + #[tokio::test] + async fn limit_is_pushed_into_scan_only_for_exact_partition_bounds() { + // `collect_from_snapshot` truncates the file list by row counts when a + // limit reaches the scan, so it must not see one when a filter only + // partially matches the surviving files. + let schema = Arc::new(Schema::new(vec![ + Field::new( + "p_timestamp", + DataType::Timestamp(arrow_schema::TimeUnit::Millisecond, None), + false, + ), + Field::new( + "created_at", + DataType::Timestamp(arrow_schema::TimeUnit::Millisecond, None), + false, + ), + ])); + let context = SessionContext::new(); + context + .register_table("logs", Arc::new(TestParquetProvider { schema })) + .unwrap(); + + for (column, limit_pushed) in [("p_timestamp", true), ("created_at", false)] { + let plan = context + .sql(&format!( + "SELECT * FROM logs WHERE {column} >= '2023-01-01T00:00:00' LIMIT 5" + )) + .await + .unwrap() + .create_physical_plan() + .await + .unwrap(); + let plan = displayable(plan.as_ref()).indent(true).to_string(); + let source_line = plan + .lines() + .find(|line| line.contains("DataSourceExec")) + .expect("parquet source line"); + + assert_eq!(source_line.contains("limit=5"), limit_pushed, "{plan}"); + assert_eq!( + plan.lines() + .any(|line| line.contains("FilterExec") && line.contains(column)), + !limit_pushed, + "{plan}" + ); + } + } + #[tokio::test] async fn optimized_plan_enforces_exact_time_once_without_duplicating_inexact_filter() { let schema = Arc::new(Schema::new(vec![