yihua commented on code in PR #708:
URL: https://github.com/apache/hudi-rs/pull/708#discussion_r3919813301
##########
crates/core/src/file_group/base_file/parquet.rs:
##########
@@ -229,22 +328,39 @@ impl BaseFileReader for ParquetBaseFileReader {
let builder = self
.open_builder(relative_path,
options.row_index_column.as_deref())
.await?;
- let builder = Self::apply_options(builder, &options)?;
+ let builder = self.apply_options(builder, &options)?;
let full_schema = builder.schema().clone();
let stream = builder.build()?;
let schema = Self::schema_with_row_index(
stream.schema(),
&full_schema,
options.row_index_column.as_deref(),
)?;
+ // Rows the stream actually yields — after any row filter. Against
+ // `file_rows` this is the read's selectivity; against
`bytes_read`,
+ // what that selectivity cost.
+ let volume = self.storage.read_volume.clone();
let mapped_stream = stream
- .map(|result| result.map_err(StorageError::from))
+ .map(move |result| {
+ let batch = result.map_err(StorageError::from)?;
+ volume.add_rows_out(batch.num_rows() as u64);
+ Ok(batch)
+ })
.boxed();
Ok(BaseFileStream::new(schema, mapped_stream))
})
}
+ /// Answered from the footer alone: no stream is built, and no read-volume
+ /// counter moves for a call that reads no data.
+ fn read_schema<'a>(
Review Comment:
non-blocking: get_schema converts with `None` key-value metadata, so a file
carrying ARROW:schema footer metadata reports different types than the stream
the default impl consults (the current caller only uses field names, so nothing
breaks today). Deriving the answer from ArrowReaderMetadata::try_new on the
already-fetched footer would make the override exactly the stream schema at the
same IO cost.
##########
crates/core/src/file_group/reader_v2/engine.rs:
##########
@@ -1629,18 +1761,235 @@ mod tests {
let reader = HoodieFileGroupReader::builder()
.with_reader_context(dummy_reader_context("MERGE_ON_READ"))
.with_storage(storage)
- .with_input_split(dummy_input_split())
+ .with_input_split(merging_input_split())
.with_row_filter_builder(make_row_filter_builder())
// mor_pk_safe defaults to false
.build()
.unwrap();
assert!(!reader.reader_context.mor_pk_safe);
assert!(
- !reader.reader_context.can_push_row_filter(),
+ !reader.base_read_pushdown_is_safe(),
"MOR without PK-safety must NOT push (mirrors Java's morFilters
gate)"
);
}
+ /// A MOR slice with no log files does not merge, so the predicate is safe
to
+ /// push whatever `mor_pk_safe` says. Parameterized over both values so the
+ /// "does it merge" rule is shown to be independent of PK safety.
+ #[test]
+ fn base_only_mor_slice_allows_pushdown_regardless_of_pk_safety() {
+ for mor_pk_safe in [false, true] {
+ let storage =
Storage::new_with_base_url(parse_uri("file:///tmp").unwrap()).unwrap();
+ let reader = HoodieFileGroupReader::builder()
+ .with_reader_context(dummy_reader_context("MERGE_ON_READ"))
+ .with_storage(storage)
+ .with_input_split(InputSplit::new(
+ Some("base.parquet".to_string()),
+ None,
+ // No log files => no merge => nothing can flip the
predicate.
+ vec![],
+ "p1".to_string(),
+ ))
+ .with_row_filter_builder(make_row_filter_builder())
+ .with_mor_pk_safe(mor_pk_safe)
+ .build()
+ .unwrap();
+ assert!(
+ reader.base_read_pushdown_is_safe(),
+ "base-only slice must push regardless of mor_pk_safe
({mor_pk_safe})"
+ );
+ }
+ }
+
+ /// The split rule must not weaken the real MOR case: with log files
present
+ /// the merge can supersede or delete a base row, so a non-PK-safe
predicate
+ /// still may not be pushed.
+ #[test]
+ fn mor_slice_with_log_files_still_blocks_pushdown_when_not_pk_safe() {
+ let storage =
Storage::new_with_base_url(parse_uri("file:///tmp").unwrap()).unwrap();
+ let reader = HoodieFileGroupReader::builder()
+ .with_reader_context(dummy_reader_context("MERGE_ON_READ"))
+ .with_storage(storage)
+ .with_input_split(InputSplit::new(
+ Some("base.parquet".to_string()),
+ None,
+ vec![".log.1".to_string()],
+ "p1".to_string(),
+ ))
+ .with_row_filter_builder(make_row_filter_builder())
+ // mor_pk_safe defaults to false
+ .build()
+ .unwrap();
+ assert!(reader.input_split.has_log_files());
+ assert!(
+ !reader.base_read_pushdown_is_safe(),
+ "MOR with log files and no PK safety must NOT push"
+ );
+ }
+
+ #[test]
+ fn builder_routes_row_group_selector_into_reader_context() {
+ let storage =
Storage::new_with_base_url(parse_uri("file:///tmp").unwrap()).unwrap();
+ let reader = HoodieFileGroupReader::builder()
+ .with_reader_context(dummy_reader_context("MERGE_ON_READ"))
+ .with_storage(storage)
+ .with_input_split(dummy_input_split())
+ .with_row_group_selector(std::sync::Arc::new(|_| None))
+ .build()
+ .unwrap();
+ assert!(
+ reader.reader_context.row_group_selector.is_some(),
+ "with_row_group_selector should land on reader_context"
+ );
+ assert!(
+ reader.reader_context.row_filter_builder.is_none(),
+ "the two mechanisms are independent: one may be set without the
other"
+ );
+ }
+
+ /// Three rows, one per row group. A selector keeping only the first must
+ /// leave the read with that row group's row and no other -- and the volume
+ /// counters must show that the other two were never scanned, which is the
+ /// difference between pruning and filtering.
+ #[tokio::test]
+ async fn a_selector_prunes_row_groups_when_the_read_does_not_merge() {
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let (tmp, base_name, schema) = three_row_groups();
+ let mut reader = test_file_group_reader_for_base_file(tmp.path(),
&base_name, schema).await;
+ let volume = reader.storage.read_volume();
+ install_selector(&mut reader, |_| Some(vec![0]), false);
+
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(out.num_rows(), 1, "only the kept row group was read");
+ assert_eq!(volume.row_group_selector_calls.load(Relaxed), 1);
+ assert_eq!(volume.row_group_selector_suppressed.load(Relaxed), 0);
+ assert_eq!(volume.file_row_groups.load(Relaxed), 3);
+ assert_eq!(
+ volume.row_groups_read.load(Relaxed),
+ 1,
+ "the other two row groups were never fetched"
+ );
+ }
+
+ /// The same selector on a slice that merges, with a predicate that is not
+ /// primary-key-safe. Pruning would drop base rows before the merge could
+ /// update them into a match, so the gate refuses it -- and counts the
+ /// refusal, because a suppressed selector otherwise reads as "no caller
ever
+ /// installed one": both are zero calls.
+ #[tokio::test]
+ async fn a_selector_the_gate_refuses_is_counted_not_silently_dropped() {
+ use std::sync::atomic::Ordering::Relaxed;
+
+ let (tmp, base_name, schema) = three_row_groups();
+ let mut reader = test_file_group_reader_for_base_file(tmp.path(),
&base_name, schema).await;
+ let volume = reader.storage.read_volume();
+ reader.input_split = InputSplit::new(
+ Some(base_name.clone()),
+ Some("20240101120000000".to_string()),
+ vec![".f1-0_20240101130000000.log.1_0-1-1".to_string()],
+ String::new(),
+ );
+ install_selector(&mut reader, |_| Some(vec![0]), false);
+
+ let out =
drain_base_source(reader.base_file_source().await.unwrap()).await;
+
+ assert_eq!(out.num_rows(), 3, "every base row still reaches the
merge");
+ assert_eq!(
+ volume.row_group_selector_calls.load(Relaxed),
+ 0,
+ "the selector never ran"
+ );
+ assert_eq!(
+ volume.row_group_selector_suppressed.load(Relaxed),
+ 1,
+ "and the reason it never ran is on the record"
+ );
+ assert_eq!(volume.row_groups_read.load(Relaxed), 3);
+ }
+
+ /// The same merging slice with a primary-key-safe predicate: the gate
opens,
+ /// so the selector runs. Pairs with the case above -- same file, same
+ /// selector, opposite outcome from `mor_pk_safe` alone.
+ #[tokio::test]
+ async fn a_pk_safe_predicate_lets_the_selector_run_on_a_merging_slice() {
Review Comment:
non-blocking: these gate tests stop at base_file_source, and the one
position-merge harness case has no selector, so the claim that RowNumber stays
absolute under with_row_groups (which the pruning-under-merge safety rests on)
is only guaranteed by parquet-rs internals today. An end-to-end case with a
selector, log files, and use_record_position asserting exact merged rows would
pin that invariant against future parquet upgrades.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
