This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 8207dfd1 perf(core): push read filters into the parquet reader (#911)
8207dfd1 is described below
commit 8207dfd19ec570cd7596c66dcdc09def9759c809
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Oct 29 23:53:30 2026 -0700
perf(core): push read filters into the parquet reader (#911)
`file_group::pushdown::filter_pushdown` turns `ReadOptions.filters` into a
parquet `RowFilter` (the filter columns are decoded first, the rest only for
the rows that pass) and a row-group selector over the footer statistics, and
`FileGroupReader` hands them to the merge-on-read reader for every slice it
reads: every filter for a slice without log files, and only the record key
filters for a slice whose log files merge. The reader's existing gate then
pushes nothing under a position-ba [...]
---
crates/core/src/file_group/mod.rs | 1 +
crates/core/src/file_group/pushdown.rs | 337 +++++++++++++++++++++
crates/core/src/file_group/reader.rs | 183 ++++++++++-
crates/core/src/file_group/reader_v2/adapter.rs | 19 ++
.../src/file_group/reader_v2/reader_context.rs | 13 +-
.../core/tests/table_read_filter_pushdown_tests.rs | 211 +++++++++++++
6 files changed, 752 insertions(+), 12 deletions(-)
diff --git a/crates/core/src/file_group/mod.rs
b/crates/core/src/file_group/mod.rs
index 15b7cbf1..b2ca4402 100644
--- a/crates/core/src/file_group/mod.rs
+++ b/crates/core/src/file_group/mod.rs
@@ -25,6 +25,7 @@ pub mod base_file;
pub mod builder;
pub mod file_slice;
pub mod log_file;
+pub(crate) mod pushdown;
pub mod reader;
pub(crate) mod reader_v2;
pub mod record_batches;
diff --git a/crates/core/src/file_group/pushdown.rs
b/crates/core/src/file_group/pushdown.rs
new file mode 100644
index 00000000..f46dcf6f
--- /dev/null
+++ b/crates/core/src/file_group/pushdown.rs
@@ -0,0 +1,337 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! What a slice read pushes into the parquet reader from the caller's filters.
+
+use std::sync::Arc;
+
+use arrow_array::RecordBatch;
+use arrow_schema::{ArrowError, DataType, Schema};
+use parquet::arrow::arrow_reader::{ArrowPredicateFn, RowFilter};
+use parquet::arrow::{ProjectionMask, parquet_to_arrow_schema};
+use parquet::file::metadata::ParquetMetaData;
+
+use crate::expr::filter::{Filter, filters_to_row_mask};
+use crate::metadata::meta_field::MetaField;
+use crate::statistics::StatisticsContainer;
+use crate::storage::{RowFilterBuilder, RowGroupSelector};
+use crate::table::file_pruner::FilePruner;
+
+/// The caller's filters compiled for the parquet reader: a row-group selector
+/// over the footer statistics and a row filter that decodes the filter
+/// columns first and the rest only for the rows that pass.
+///
+/// Both are conservative. A file without one of the filter columns gets no
+/// row filter, and a row group without statistics is kept; the row-level mask
+/// the reader applies afterwards still decides every row.
+pub(crate) struct FilterPushdown {
+ pub row_filter: RowFilterBuilder,
+ pub row_groups: RowGroupSelector,
+ /// Every filter is on a record key field, which no log record can change,
+ /// so the filter may drop base rows before a merge-on-read merge. Always
+ /// true for a slice that merges, which keeps only those filters.
+ pub mor_pk_safe: bool,
+}
+
+/// `table_schema` is the schema the read returns rows in. A file written
+/// before a type widening stores a filter column in a narrower type, and a
+/// literal compared in that type can disagree with the same comparison in
+/// the table's type, so such a file gets nothing pushed.
+///
+/// A slice that `merges` log records into its base file pushes only the
+/// record key filters: a filter on any other column could drop a base row
+/// that a log record would have lost to.
+pub(crate) fn filter_pushdown(
+ filters: &[Filter],
+ record_key_fields: &[String],
+ table_schema: &Schema,
+ merges: bool,
+) -> Option<FilterPushdown> {
+ let is_key = |filter: &&Filter| {
+ filter.field == MetaField::RecordKey.as_ref() ||
record_key_fields.contains(&filter.field)
+ };
+ let filters: Vec<Filter> = if merges {
+ filters.iter().filter(is_key).cloned().collect()
+ } else {
+ filters.to_vec()
+ };
+ if filters.is_empty() {
+ return None;
+ }
+ let mor_pk_safe = filters.iter().all(|filter| is_key(&filter));
+ let table_types: Vec<(String, DataType)> = filters
+ .iter()
+ .map(|filter| {
+ let data_type = table_schema
+ .field_with_name(&filter.field)
+ .map(|f| f.data_type().clone())
+ .unwrap_or(DataType::Null);
+ (filter.field.clone(), data_type)
+ })
+ .collect();
+ // Every filter column, stored in the file as the table has it.
+ let same_types = |file_schema: &Schema, table_types: &[(String,
DataType)]| {
+ table_types.iter().all(|(name, data_type)| {
+ file_schema
+ .field_with_name(name)
+ .is_ok_and(|f| f.data_type() == data_type)
+ })
+ };
+ let for_rows = filters.clone();
+ let row_types = table_types.clone();
+ // Checked against the schema the reader decodes with (its embedded Arrow
+ // hint included), which is what the predicate sees.
+ let row_filter: RowFilterBuilder = Arc::new(move |parquet_schema,
read_schema| {
+ if !same_types(read_schema, &row_types) {
+ return None;
+ }
+ let root = parquet_schema.root_schema();
+ let mut columns: Vec<usize> = for_rows
+ .iter()
+ .map(|filter| {
+ root.get_fields()
+ .iter()
+ .position(|field| field.name() == filter.field)
+ })
+ .collect::<Option<_>>()?;
+ columns.sort_unstable();
+ columns.dedup();
+ let mask = ProjectionMask::roots(parquet_schema, columns);
+ let filters = for_rows.clone();
+ let predicate = ArrowPredicateFn::new(mask, move |batch: RecordBatch| {
+ filters_to_row_mask(&filters, &batch)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))
+ });
+ Some(RowFilter::new(vec![Box::new(predicate)]))
+ });
+ // Statistics leave NaN out of a float column's range, so a range check on
+ // one could drop a row group holding NaN; those columns are left to the
+ // row filter.
+ let for_groups: Vec<Filter> = filters
+ .iter()
+ .filter(|filter| {
+ !matches!(
+ table_schema
+ .field_with_name(&filter.field)
+ .map(|f| f.data_type()),
+ Ok(DataType::Float16 | DataType::Float32 | DataType::Float64)
+ )
+ })
+ .cloned()
+ .collect();
+ let group_types = table_types;
+ let row_groups: RowGroupSelector = Arc::new(move |metadata:
&ParquetMetaData| {
+ let file = metadata.file_metadata();
+ let schema =
+ parquet_to_arrow_schema(file.schema_descr(),
file.key_value_metadata()).ok()?;
+ if !same_types(&schema, &group_types) {
+ return None;
+ }
+ let pruner = FilePruner::new(&for_groups, &schema,
&Schema::empty()).ok()?;
+ Some(
+ (0..metadata.num_row_groups())
+ .filter(|&index| {
+ pruner.should_include(&StatisticsContainer::from_row_group(
+ metadata.row_group(index),
+ &schema,
+ ))
+ })
+ .collect(),
+ )
+ });
+ Some(FilterPushdown {
+ row_filter,
+ row_groups,
+ mor_pk_safe,
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::expr::filter::col;
+ use arrow_array::{Int64Array, StringArray};
+ use arrow_schema::{DataType, Field};
+ use bytes::Bytes;
+ use parquet::arrow::ArrowWriter;
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+ use parquet::file::properties::WriterProperties;
+
+ fn schema() -> Schema {
+ Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("v", DataType::Int64, false),
+ ])
+ }
+
+ /// Ten rows `k00..k09` with values `0..9`, two rows per row group.
+ fn parquet_file() -> Bytes {
+ let schema = Arc::new(schema());
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(StringArray::from(
+ (0..10).map(|i| format!("k{i:02}")).collect::<Vec<_>>(),
+ )),
+ Arc::new(Int64Array::from_iter_values(0..10)),
+ ],
+ )
+ .unwrap();
+ let props = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(2))
+ .build();
+ let mut out = Vec::new();
+ let mut writer = ArrowWriter::try_new(&mut out, schema,
Some(props)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ Bytes::from(out)
+ }
+
+ fn ids_through(pushdown: &FilterPushdown, file: Bytes) -> (Vec<usize>,
Vec<String>) {
+ let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
+ let groups = (pushdown.row_groups)(builder.metadata()).unwrap();
+ let filter = (pushdown.row_filter)(builder.parquet_schema(),
builder.schema()).unwrap();
+ let reader = builder
+ .with_row_groups(groups.clone())
+ .with_row_filter(filter)
+ .build()
+ .unwrap();
+ let ids = reader
+ .map(|b| b.unwrap())
+ .flat_map(|b| {
+ b.column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .iter()
+ .map(|v| v.unwrap().to_string())
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ (groups, ids)
+ }
+
+ #[test]
+ fn test_pushdown_selects_row_groups_and_rows() {
+ let file = parquet_file();
+ let one = filter_pushdown(
+ &[col("id").eq("k05")],
+ &["id".to_string()],
+ &schema(),
+ false,
+ )
+ .unwrap();
+ assert!(one.mor_pk_safe);
+ assert_eq!(
+ ids_through(&one, file.clone()),
+ (vec![2], vec!["k05".to_string()])
+ );
+
+ let two =
+ filter_pushdown(&[col("id").in_list(["k01", "k08"])], &[],
&schema(), false).unwrap();
+ assert!(!two.mor_pk_safe);
+ assert_eq!(
+ ids_through(&two, file.clone()),
+ (vec![0, 4], vec!["k01".to_string(), "k08".to_string()])
+ );
+
+ let both = filter_pushdown(
+ &[col("v").gte("6"), col("id").ne("k07")],
+ &[],
+ &schema(),
+ false,
+ )
+ .unwrap();
+ assert_eq!(
+ ids_through(&both, file.clone()),
+ (
+ vec![3, 4],
+ vec!["k06".to_string(), "k08".to_string(), "k09".to_string()]
+ )
+ );
+
+ let none = filter_pushdown(&[col("v").gt("100")], &[], &schema(),
false).unwrap();
+ assert_eq!(ids_through(&none, file), (vec![], vec![]));
+ }
+
+ /// A slice that merges keeps only the record key half of a mixed filter.
+ #[test]
+ fn test_a_merging_slice_pushes_only_record_key_filters() {
+ let file = parquet_file();
+ let keys = ["id".to_string()];
+ let mixed = [col("id").in_list(["k01", "k08"]), col("v").eq("8")];
+
+ let merging = filter_pushdown(&mixed, &keys, &schema(), true).unwrap();
+ assert!(merging.mor_pk_safe);
+ assert_eq!(
+ ids_through(&merging, file.clone()),
+ (vec![0, 4], vec!["k01".to_string(), "k08".to_string()])
+ );
+ let plain = filter_pushdown(&mixed, &keys, &schema(), false).unwrap();
+ assert!(!plain.mor_pk_safe);
+ assert_eq!(
+ ids_through(&plain, file),
+ (vec![4], vec!["k08".to_string()])
+ );
+ assert!(filter_pushdown(&[col("v").eq("8")], &keys, &schema(),
true).is_none());
+ }
+
+ /// A file holding a filter column in another type than the table (a file
+ /// from before a widening) gets nothing pushed, and a float column gets no
+ /// row-group pruning.
+ #[test]
+ fn test_pushdown_skips_a_narrower_file_type_and_float_statistics() {
+ let file = parquet_file();
+ let wide = Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("v", DataType::Decimal128(20, 0), false),
+ ]);
+ let pushdown = filter_pushdown(&[col("v").gt("5")], &[], &wide,
false).unwrap();
+ let builder =
ParquetRecordBatchReaderBuilder::try_new(file.clone()).unwrap();
+ assert!((pushdown.row_filter)(builder.parquet_schema(),
builder.schema()).is_none());
+ assert!((pushdown.row_groups)(builder.metadata()).is_none());
+
+ let floats = Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("v", DataType::Float64, false),
+ ]);
+ let pushdown = filter_pushdown(
+ &[col("v").gt("5"), col("id").eq("k01")],
+ &[],
+ &floats,
+ false,
+ )
+ .unwrap();
+ let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
+ // The file stores `v` as Int64, so the type check skips the file; with
+ // a Float64 file the selector would still ignore the `v` filter.
+ assert!((pushdown.row_groups)(builder.metadata()).is_none());
+ }
+
+ /// A column the file does not have gets no row filter and no row-group
+ /// pruning; the reader's mask still handles it.
+ #[test]
+ fn test_pushdown_skips_a_column_the_file_lacks() {
+ let file = parquet_file();
+ let pushdown = filter_pushdown(&[col("missing").eq("x")], &[],
&schema(), false).unwrap();
+ let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
+ assert!((pushdown.row_filter)(builder.parquet_schema(),
builder.schema()).is_none());
+ // No opinion on row groups either: the reader fetches them all.
+ assert!((pushdown.row_groups)(builder.metadata()).is_none());
+ assert!(filter_pushdown(&[], &[], &schema(), false).is_none());
+ }
+}
diff --git a/crates/core/src/file_group/reader.rs
b/crates/core/src/file_group/reader.rs
index 4381d4b9..e2996e28 100644
--- a/crates/core/src/file_group/reader.rs
+++ b/crates/core/src/file_group/reader.rs
@@ -354,15 +354,42 @@ impl FileGroupReader {
if unchanged { None } else { Some(target) }
}
+ /// The caller's filters, compiled for the parquet reader: all of them for
+ /// a slice read without log files, only the record key filters for one
+ /// that `merges` them. The row-level mask applied afterwards still decides
+ /// every row.
+ fn filter_pushdown(
+ &self,
+ options: &ReadOptions,
+ data_schema: &arrow_schema::Schema,
+ merges: bool,
+ ) -> Option<crate::file_group::pushdown::FilterPushdown> {
+ let record_key_fields: Vec<String> = self
+ .hudi_configs
+ .try_get(HudiTableConfig::RecordKeyFields)
+ .ok()
+ .flatten()
+ .map(Into::into)
+ .unwrap_or_default();
+ crate::file_group::pushdown::filter_pushdown(
+ &options.filters,
+ &record_key_fields,
+ data_schema,
+ merges,
+ )
+ }
+
/// Read one slice through the merge-on-read reader.
///
- /// Filters and projection are not applied here: the caller runs the result
- /// through the same `apply_eager_options` the other paths use, so the two
- /// engines cannot disagree about what a filter means.
+ /// The caller's filters reach the parquet reader as a pushdown (see
+ /// [`Self::filter_pushdown`]); they are still applied, with the
+ /// projection, by the caller's `apply_eager_options`, so the two engines
+ /// cannot disagree about what a filter means.
async fn read_via_v2(
&self,
base_file_path: &str,
log_file_paths: Vec<String>,
+ options: &ReadOptions,
) -> Result<RecordBatch> {
let data_schema = self.resolved_data_schema(base_file_path).await?;
if let Some(reason) =
crate::file_group::reader_v2::adapter::refuse_reason(
@@ -382,12 +409,14 @@ impl FileGroupReader {
// Per file slice, so debug: a wide read would repeat this once per
// slice, and which reader served the table is not news N times over.
log::debug!("reading '{base_file_path}' with file group reader version
2");
+ let pushdown = self.filter_pushdown(options, &data_schema,
!log_file_paths.is_empty());
let merged = crate::file_group::reader_v2::adapter::read_file_slice(
self.hudi_configs.clone(),
self.storage.clone(),
base_file_path,
log_file_paths,
partition_path,
+ pushdown,
Some(data_schema),
self.completion_gate_inputs.clone(),
)
@@ -620,7 +649,8 @@ impl FileGroupReader {
} else {
log_file_paths
};
- self.read_via_v2(base_file_path, log_file_paths).await?
+ self.read_via_v2(base_file_path, log_file_paths, &options)
+ .await?
} else if base_file_only {
// Nothing to merge — a copy-on-write slice, or a read-optimized
read
// that ignores the log files. Served by the base file reader, not
by
@@ -833,12 +863,14 @@ impl FileGroupReader {
// Per file slice — see the eager sibling.
log::debug!("streaming '{base_file_path}' with file group reader
version 2");
+ let pushdown = self.filter_pushdown(options, &data_schema,
!log_file_paths.is_empty());
let merged =
crate::file_group::reader_v2::adapter::read_file_slice_stream(
self.hudi_configs.clone(),
self.storage.clone(),
base_file_path,
log_file_paths,
partition_path,
+ pushdown,
Some(data_schema),
self.completion_gate_inputs.clone(),
)
@@ -1222,6 +1254,149 @@ mod tests {
use std::sync::Arc;
use url::Url;
+ /// A filtered slice read reaches the parquet reader: with two rows per
+ /// row group, a key filter fetches one row group of the file.
+ #[tokio::test]
+ async fn test_filters_prune_row_groups_of_a_slice_read() {
+ use crate::table::Table;
+ use arrow_array::{Int64Array, StringArray};
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let dir = tempfile::tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .with_option("hoodie.parquet.block.size", "2048")
+ .with_option("hoodie.copyonwrite.record.size.estimate", "1024")
+ .create()
+ .await
+ .unwrap();
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(
+ (0..10).map(|i| format!("k{i:02}")).collect::<Vec<_>>(),
+ )),
+ Arc::new(Int64Array::from_iter_values(0..10)),
+ ],
+ )
+ .unwrap();
+ table.append([batch]).await.unwrap();
+ let base_file = std::fs::read_dir(dir.path())
+ .unwrap()
+ .filter_map(|e| e.ok())
+ .map(|e| e.file_name().to_string_lossy().to_string())
+ .find(|name| name.ends_with(".parquet"))
+ .unwrap();
+
+ let table = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ let reader = table
+ .create_file_group_reader_with_options(
+ Some(&ReadOptions::new()),
+ std::iter::empty::<(&str, &str)>(),
+ )
+ .await
+ .unwrap();
+ let volume = reader.storage.read_volume();
+ let options = ReadOptions::new()
+ .with_filters([("id", "=", "k05")])
+ .unwrap();
+ let batch = reader
+ .read_file_slice_from_paths(&base_file, Vec::<String>::new(),
&options)
+ .await
+ .unwrap();
+ assert_eq!(batch.num_rows(), 1);
+ assert_eq!(volume.file_row_groups.load(Relaxed), 5);
+ assert_eq!(volume.row_groups_read.load(Relaxed), 1);
+ }
+
+ /// On a slice whose log files merge, the record key half of a mixed
+ /// filter still reaches the parquet reader, and the merged update wins.
+ #[tokio::test]
+ async fn test_key_filters_prune_row_groups_of_a_merging_slice_read() {
+ use crate::config::table::TableTypeValue;
+ use crate::table::Table;
+ use arrow_array::{Int64Array, StringArray};
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let dir = tempfile::tempdir().unwrap();
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ let batch = |ids: Vec<String>, values: Vec<i64>| {
+ RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(StringArray::from(ids)),
+ Arc::new(Int64Array::from(values)),
+ ],
+ )
+ .unwrap()
+ };
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .with_table_type(TableTypeValue::MergeOnRead)
+ .with_option("hoodie.parquet.block.size", "2048")
+ .with_option("hoodie.copyonwrite.record.size.estimate", "1024")
+ .create()
+ .await
+ .unwrap();
+ table
+ .append([batch(
+ (0..10).map(|i| format!("k{i:02}")).collect(),
+ (0..10).collect(),
+ )])
+ .await
+ .unwrap();
+ table
+ .upsert([batch(vec!["k05".to_string()], vec![500])])
+ .await
+ .unwrap();
+ let names: Vec<String> = std::fs::read_dir(dir.path())
+ .unwrap()
+ .filter_map(|e| e.ok())
+ .map(|e| e.file_name().to_string_lossy().to_string())
+ .collect();
+ let base_file = names.iter().find(|n|
n.ends_with(".parquet")).unwrap();
+ let log_files: Vec<&String> = names.iter().filter(|n|
n.contains(".log.")).collect();
+ assert_eq!(log_files.len(), 1, "the upsert went to a log");
+
+ let table = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ let reader = table
+ .create_file_group_reader_with_options(
+ Some(&ReadOptions::new()),
+ std::iter::empty::<(&str, &str)>(),
+ )
+ .await
+ .unwrap();
+ let volume = reader.storage.read_volume();
+ let options = ReadOptions::new()
+ .with_filters([("id", "=", "k05"), ("value", "=", "500")])
+ .unwrap();
+ let batch = reader
+ .read_file_slice_from_paths(base_file, log_files, &options)
+ .await
+ .unwrap();
+ assert_eq!(batch.num_rows(), 1);
+ let values = batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(values.value(0), 500);
+ assert_eq!(volume.file_row_groups.load(Relaxed), 5);
+ assert_eq!(volume.row_groups_read.load(Relaxed), 1);
+ }
+
const TEST_SAMPLE_BASE_FILE: &str =
"a079bdb3-731c-4894-b855-abfcd6921007-0_0-203-274_20240418173551906.parquet";
diff --git a/crates/core/src/file_group/reader_v2/adapter.rs
b/crates/core/src/file_group/reader_v2/adapter.rs
index 24c4d30e..c071524b 100644
--- a/crates/core/src/file_group/reader_v2/adapter.rs
+++ b/crates/core/src/file_group/reader_v2/adapter.rs
@@ -33,6 +33,7 @@ use crate::Result;
use crate::config::HudiConfigs;
use crate::config::read::HudiReadConfig;
use crate::error::CoreError;
+use crate::file_group::pushdown::FilterPushdown;
use crate::file_group::reader_v2::MAX_INSTANT_TIME;
use crate::file_group::reader_v2::engine::HoodieFileGroupReader;
use crate::file_group::reader_v2::input_split::InputSplit;
@@ -50,12 +51,14 @@ use std::sync::Arc;
/// to resolve partition-valued fields. `data_schema` is the table's schema;
the
/// merge-on-read reader requires it up front rather than deriving it from the
/// data as the existing reader does.
+#[allow(clippy::too_many_arguments)]
pub(crate) async fn read_file_slice(
hudi_configs: Arc<HudiConfigs>,
storage: Arc<Storage>,
base_file_path: &str,
log_file_paths: Vec<String>,
partition_path: String,
+ pushdown: Option<FilterPushdown>,
data_schema: Option<SchemaRef>,
completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
) -> Result<RecordBatch> {
@@ -65,6 +68,7 @@ pub(crate) async fn read_file_slice(
base_file_path,
log_file_paths,
partition_path,
+ pushdown,
data_schema,
completion_gate_inputs,
)?;
@@ -77,12 +81,14 @@ pub(crate) async fn read_file_slice(
/// Same merge as [`read_file_slice`], but the base file is decoded one row
group
/// at a time instead of whole, so peak memory tracks a row group rather than
the
/// file. Arguments carry the same meaning.
+#[allow(clippy::too_many_arguments)]
pub(crate) async fn read_file_slice_stream(
hudi_configs: Arc<HudiConfigs>,
storage: Arc<Storage>,
base_file_path: &str,
log_file_paths: Vec<String>,
partition_path: String,
+ pushdown: Option<FilterPushdown>,
data_schema: Option<SchemaRef>,
completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
) -> Result<futures::stream::BoxStream<'static, Result<RecordBatch>>> {
@@ -92,6 +98,7 @@ pub(crate) async fn read_file_slice_stream(
base_file_path,
log_file_paths,
partition_path,
+ pushdown,
data_schema,
completion_gate_inputs,
)?;
@@ -101,12 +108,14 @@ pub(crate) async fn read_file_slice_stream(
/// Assemble the reader both entry points use, so the eager and streaming reads
/// cannot resolve their context, split or parameters differently.
+#[allow(clippy::too_many_arguments)]
fn build_reader(
hudi_configs: Arc<HudiConfigs>,
storage: Arc<Storage>,
base_file_path: &str,
log_file_paths: Vec<String>,
partition_path: String,
+ pushdown: Option<FilterPushdown>,
data_schema: Option<SchemaRef>,
completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
) -> Result<HoodieFileGroupReader> {
@@ -121,6 +130,16 @@ fn build_reader(
// timeline; `None` leaves the gate a no-op, which is what a caller reading
// from paths alone (the cxx bridge) can offer.
context.completion_gate_inputs = completion_gate_inputs;
+ if let Some(pushdown) = pushdown {
+ // `repair_risk_columns` stays empty: the pushdown refuses any file
that
+ // stores a filter column in a type other than the table's, which
covers
+ // every file the logical timestamp repair would reinterpret.
+ context.row_filter_builder = Some(pushdown.row_filter);
+ context.row_group_selector = Some(pushdown.row_groups);
+ // A merge by row position pairs log records with base rows one to one,
+ // so nothing may drop records from either side before it.
+ context.mor_pk_safe = pushdown.mor_pk_safe &&
!context.should_merge_use_record_position;
+ }
context.rebuild_record_context(partition_path.clone());
let base_file_commit_time = base_file_path.and_then(base_file_commit_time);
diff --git a/crates/core/src/file_group/reader_v2/reader_context.rs
b/crates/core/src/file_group/reader_v2/reader_context.rs
index 51197f38..2d0baff4 100644
--- a/crates/core/src/file_group/reader_v2/reader_context.rs
+++ b/crates/core/src/file_group/reader_v2/reader_context.rs
@@ -127,8 +127,8 @@ pub struct ReaderContext {
/// appear in one of those two, or it reads as unset.
pub hoodie_reader_config: HashMap<String, String>,
/// Optional parquet `RowFilter` builder for predicate pushdown into base
- /// parquet files and parquet-format log blocks. Set by the FFI bridge from
- /// the decoded substrait predicate; `None` when no predicate was pushed.
+ /// parquet files and parquet-format log blocks; `None` when no predicate
was
+ /// pushed.
///
/// Whether it actually gets installed is gated by whether the read merges:
/// `HoodieFileGroupReader::base_read_pushdown_is_safe` for the base file
@@ -136,13 +136,10 @@ pub struct ReaderContext {
/// [`Self::mor_pk_safe`] alone for a parquet log block, which by
definition
/// only exists on a slice that does merge.
///
- /// **Not active for reads through this crate.**
///
[`resolve_reader_context`](crate::file_group::reader_v2::resolver::resolve_reader_context)
- /// leaves it `None`, so a `Table` or DataFusion read pushes no predicate
into
- /// either the base file or a log block, and the primary-key-safety gate
never
- /// runs outside the test harness. Filters are applied after the merge
instead.
- /// Groundwork for a caller that supplies one; no performance claim about
this
- /// crate's reads rests on it.
+ /// leaves it `None`; a slice read sets it from the caller's filters
+ /// (`file_group::pushdown`), and the caller still applies those filters
+ /// after the merge.
pub row_filter_builder: Option<RowFilterBuilder>,
/// Optional row-group selector for pruning base parquet reads from footer
/// statistics; `None` when no caller supplied one.
diff --git a/crates/core/tests/table_read_filter_pushdown_tests.rs
b/crates/core/tests/table_read_filter_pushdown_tests.rs
new file mode 100644
index 00000000..11683704
--- /dev/null
+++ b/crates/core/tests/table_read_filter_pushdown_tests.rs
@@ -0,0 +1,211 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+//! Filters pushed into the parquet reader must not change what a read returns,
+//! on copy-on-write and on merge-on-read tables whose log files update the
+//! filtered columns.
+
+use std::sync::Arc;
+
+use arrow_array::{Int64Array, RecordBatch, StringArray};
+use arrow_schema::{DataType, Field, Schema};
+use hudi_core::config::table::TableTypeValue;
+use hudi_core::table::{ReadOptions, Table};
+use tempfile::tempdir;
+
+fn batch(ids: &[&str], values: &[i64]) -> RecordBatch {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(ids.to_vec())),
+ Arc::new(Int64Array::from(values.to_vec())),
+ ],
+ )
+ .unwrap()
+}
+
+fn filters(filters: &[(&str, &str, &str)]) -> ReadOptions {
+ ReadOptions::new()
+ .with_filters(filters.iter().copied())
+ .unwrap()
+}
+
+async fn rows(table: &Table, options: &ReadOptions) -> Vec<(String, i64)> {
+ let mut out: Vec<(String, i64)> = table
+ .read(options)
+ .await
+ .unwrap()
+ .iter()
+ .flat_map(|b| {
+ let ids = b
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let values = b
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ ids.iter()
+ .zip(values.iter())
+ .map(|(i, v)| (i.unwrap().to_string(), v.unwrap()))
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ out.sort();
+ out
+}
+
+/// Two file groups per table, with a few rows per parquet row group so the
+/// row-group selection has something to drop.
+async fn table(dir: &std::path::Path, table_type: TableTypeValue) -> Table {
+ let mut table = Table::create(dir.to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .with_table_type(table_type)
+ .with_option("hoodie.parquet.small.file.limit", "0")
+ .with_option("hoodie.parquet.block.size", "2048")
+ .with_option("hoodie.copyonwrite.record.size.estimate", "1024")
+ .create()
+ .await
+ .unwrap();
+ table
+ .append([batch(&["a1", "a2", "a3", "a4"], &[1, 2, 3, 4])])
+ .await
+ .unwrap();
+ table
+ .append([batch(&["b1", "b2", "b3", "b4"], &[5, 6, 7, 8])])
+ .await
+ .unwrap();
+ Table::new(dir.to_str().unwrap()).await.unwrap()
+}
+
+#[tokio::test]
+async fn test_pushed_filters_return_the_same_rows_on_copy_on_write() {
+ let dir = tempdir().unwrap();
+ let table = table(dir.path(), TableTypeValue::CopyOnWrite).await;
+ assert_eq!(
+ rows(
+ &table,
+ &filters(&[("value", ">=", "3"), ("value", "<", "7")])
+ )
+ .await,
+ vec![
+ ("a3".to_string(), 3),
+ ("a4".to_string(), 4),
+ ("b1".to_string(), 5),
+ ("b2".to_string(), 6),
+ ]
+ );
+ assert_eq!(
+ rows(&table, &filters(&[("_hoodie_record_key", "IN", "a2,b4")])).await,
+ vec![("a2".to_string(), 2), ("b4".to_string(), 8)]
+ );
+ assert_eq!(rows(&table, &filters(&[("id", "=", "zz")])).await, vec![]);
+}
+
+/// A log file changes `value`; a filter on `value` must see the merged row,
+/// so it cannot drop base rows before the merge, while a filter on the key
+/// can.
+#[tokio::test]
+async fn test_pushed_filters_see_merged_rows_on_merge_on_read() {
+ let dir = tempdir().unwrap();
+ table(dir.path(), TableTypeValue::MergeOnRead).await;
+ let mut writer = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ writer
+ .upsert([batch(&["a2", "b3"], &[200, 300])])
+ .await
+ .unwrap();
+ let table = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+
+ // The old values are gone and the new ones match.
+ assert_eq!(rows(&table, &filters(&[("value", "=", "2")])).await, vec![]);
+ assert_eq!(
+ rows(&table, &filters(&[("value", ">", "100")])).await,
+ vec![("a2".to_string(), 200), ("b3".to_string(), 300)]
+ );
+ // Key filters are pushed into the base read and still merge the update.
+ assert_eq!(
+ rows(&table, &filters(&[("id", "IN", "a2,a3")])).await,
+ vec![("a2".to_string(), 200), ("a3".to_string(), 3)]
+ );
+ assert_eq!(
+ rows(
+ &table,
+ &filters(&[("id", "=", "b3"), ("value", "=", "300")])
+ )
+ .await,
+ vec![("b3".to_string(), 300)]
+ );
+}
+
+/// With an ordering field a log record can lose to the base row. A filter on
+/// the column it would have changed must then see the base value, which a
+/// filter pushed below the merge would have dropped.
+#[tokio::test]
+async fn test_a_losing_log_record_does_not_leak_through_a_pushed_filter() {
+ let dir = tempdir().unwrap();
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ Field::new("ts", DataType::Int64, false),
+ ]));
+ let batch = |value: i64, ts: i64| {
+ RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["a1"])),
+ Arc::new(Int64Array::from(vec![value])),
+ Arc::new(Int64Array::from(vec![ts])),
+ ],
+ )
+ .unwrap()
+ };
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .with_ordering_fields(["ts"])
+ .with_table_type(TableTypeValue::MergeOnRead)
+ .create()
+ .await
+ .unwrap();
+ table.append([batch(5, 10)]).await.unwrap();
+ table.upsert([batch(1, 5)]).await.unwrap();
+ let table = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ let slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
+ assert!(
+ slices.iter().any(|s| s.has_log_file()),
+ "the upsert went to a log"
+ );
+ assert_eq!(rows(&table, &filters(&[("value", "=", "1")])).await, vec![]);
+ assert_eq!(
+ rows(&table, &filters(&[("id", "=", "a1"), ("value", "=",
"1")])).await,
+ vec![]
+ );
+ assert_eq!(
+ rows(&table, &filters(&[("value", "=", "5")])).await,
+ vec![("a1".to_string(), 5)]
+ );
+}
