Lin/eng38913 mor prep#610
Open
linliu-code wants to merge 17 commits into
Open
Conversation
Replace the top-level-only `RecordBatch::project(&indices)` with a recursive `narrow_array_to_field` walk that descends into `StructArray` children when the target field is a narrower `Struct`. Examples that now work end-to-end (where they previously emitted the wider struct): - `fare<amount, currency>` -> `fare<currency>` (subset) - `fare<amount, currency>` -> `fare<currency, amount>` (reorder) Build the result via `RecordBatch::try_new_with_options` carrying an explicit row count, so an entirely-pruned data schema (e.g. a query that selects only partition columns) doesn't trip Arrow's "must specify a row count or at least one column" invariant. Includes column-name-aware error messages when a requested top-level column or struct subfield isn't present in the source. Tests: - `test_output_converter_narrows_nested_struct` - `test_output_converter_reorders_struct_subfields` - `test_output_converter_errors_on_missing_subfield` - `test_output_converter_empty_target_preserves_row_count` 7/7 existing+new tests pass. Verified end-to-end via gluten/velox TestMORDataSource#testPrunedFiltered[AVRO,9]: hudi-rs now ships `fare:ROW<currency:VARCHAR>` directly when only `fare.currency` is asked for (vs always the full struct pre-fix). Engages once gluten-internal's Hudi1xScanSplitInfoAdapter actually populates the requested_schema_json over FFI. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
… perf regressed) Add `ParquetReadOptions::nested_projection: Option<SchemaRef>` so callers can hand a (potentially nested-narrowed) Arrow schema to the parquet reader, which we translate into `ProjectionMask::leaves(...)` to skip unused leaf columns at decode time. Wired into `Storage::get_parquet_file_data_projected` so the v9 file-group reader automatically gets leaf-level pruning when `required_schema` is set. KNOWN PERF REGRESSION (~6x wall-clock vs. top-level `roots` projection) on TestMORDataSource at N=50000. The functional path is correct — test passes 1/1 — but enabling B2 by default would make every query ~6x slower. Pending investigation, this commit ships the code disabled- by-default behind `HUDI_B2_ENABLED` env var; set to `1`/`true` to opt in. Default is `false` so existing callers behave identically to the pre-B2 top-level path. A/B (2026-05-10): - B2 OFF: 199.7s - B2 ON : 1192.9s Spark-job time is identical (118s vs 122s). The 6x slowdown is in a single Spark task that takes 974s with B2 on vs <1m off. Suspect the List<Struct<>> handling or per-batch reorder cost. Investigation captured under the ENG-40168 follow-up subtask. Tests: - `arrow_includes_parquet_leaf_*` (5 cases — empty path, struct subfield present/pruned, recursive struct, list conservative) - `parquet_read_options_accessors` 16/16 storage tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
…tion
Add structured per-call timing logs in get_parquet_file_data_projected
and the get_parquet_file_stream B2 leaf-index walk. All emit at
log::debug! level so they are silent at the default RUST_LOG=info and
only fire when an investigator explicitly opts in:
RUST_LOG=info,hudi_core::storage=debug
Logged metrics (one block per parquet read call):
- get_parquet_file_data_projected START + path + top_cols
- B2 leaf-index walk: total_parquet_leaves, selected, walk_us
- stream open time + post-projection col / leaf counts
- per-batch read time (min, max), batch count, total rows
- concat time and TOTAL elapsed ms
Helper: count_leaves(DataType) walks Struct / List / LargeList /
FixedSizeList / Map element types.
Useful for ENG-41499 (B2 perf investigation, currently deferred): when
a real-scale workload — TPC-DS or Lake Loader 100 GB — surfaces a B2
perf signal and someone reopens the investigation, this instrumentation
is in place. Removing it would force re-authoring identical code.
Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Replaces the Velox-side Option A workaround (unconditional remainingFilter
recompile in HudiSplitReader) with end-to-end filter pushdown:
Phase 1 — FfiColumnFilter shared struct in cpp/src/lib.rs cxx-bridge.
Custom union (kind-string + parallel scalar fields) covers IsNull,
IsNotNull, BoolValue, BigintRange, FloatingPointRange (f32/f64 widened
to f64), BigintValuesUsingHashTable/Bitmask, BytesRange/Values for UTF-8.
Phase 3 — Post-merge filter (semantically safe for all MOR shapes).
cpp/src/predicate.rs decodes FfiColumnFilter into ColumnPredicate,
evaluates against the merged RecordBatch using arrow-rs typed-array
accessors. HoodieFileGroupReader::read_record_batch applies the filter
after the base+log merge, so log updates are resolved before filtering.
Phase 4 — Parquet RowFilter hooks for safe cases (no log files).
- predicate::build_row_filter builds a parquet RowFilter from
ColumnPredicate slice
- predicate::is_parquet_pushdown_safe gates the call site
- storage/mod.rs gains a RowFilterBuilder type, a row_filter_builder
field on ParquetReadOptions, and get_parquet_file_data_projected_with_options
- storage's stream builder installs with_row_filter when the option
is set
Phase 5 — Rust unit tests in predicate.rs cover boundary semantics,
NULL handling, missing-column passthrough, AND combination, and type
mismatch error.
Unrecognised filter kinds are skipped with a log line, NOT errored —
Velox's existing post-scan filter still evaluates them, so correctness
is preserved while we iterate on IR coverage.
Velox-side wiring (appendScanSpecFilters in HudiSplitReader) is in the
companion velox-internal commit. Option A is intentionally left in
place for one release; revert is tracked as a follow-up.
Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
… format Replaces the custom FfiColumnFilter IR with substrait::ExtendedExpression bytes so the filter Velox already received from Gluten can be evaluated inside hudi-rs without re-encoding through a per-column tagged union. * FFI: column_filters: Vec<FfiColumnFilter> → substrait_filter_bytes: Vec<u8>. * predicate.rs rewritten as a narrow Substrait evaluator: ScalarFunction (equal/not_equal/lt/lte/gt/gte/is_null/is_not_null/and/or/not), FieldReference, Literal over i32/i64/f32/f64/bool/Utf8/LargeUtf8. Unsupported shapes drop the predicate (Velox's post-scan filter is the safety net) — no correctness regression. * substrait crate pinned to =0.59: 0.60+ renamed extension_uris → extension_urns, breaking wire compatibility with Gluten's bundled substrait protos (which still use extension_uris). * 13 new unit tests round-trip ExtendedExpression → arrow filter: decode/compare/and/or/not/is_null/null-literal/reversed-operands/ missing-column-error/nested-compound. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
…+ graceful fallback Extends the Substrait predicate evaluator added in f18245b to cover every column type TestMORFileSliceLayouts exercises (and that MOR queries typically see). Also adds a graceful fallback so any shape the evaluator doesn't recognise drops the pushed filter and falls back to Velox's post-scan filter — preserving correctness regardless of which Substrait expression Gluten emits. predicate.rs: * ScalarValue gains I8, I16, Decimal128(value,precision,scale), Date, TimestampMicros, Binary variants. * decode_literal handles: Boolean, I8/I16/I32/I64, Fp32/Fp64, String, FixedChar, VarChar, Binary, FixedBinary, Uuid, Decimal (16-byte little-endian two's-complement + precision/scale), Date (i32 days since epoch), Timestamp (i64 µs since epoch — deprecated form Gluten still emits), PrecisionTimestamp/PrecisionTimestampTz (rescaled to µs using the literal's precision field), and typed Null. * compare_column_scalar gains Int8/Int16, Date32, Timestamp (with per-TimeUnit rescaling between µs scalar and second/ms/µs/ns column), Decimal128 (with per-scale rescaling between scalar and column), Binary/LargeBinary (byte-lex order; String scalar coerces to bytes). * 18 new unit tests covering each new type, mixed-type compound predicates, IS NOT NULL, and a coverage-marker that lists every ScalarValue variant (compilation-checked). lib.rs (read_record_batch): * If predicate::filter_batch errors (unsupported nested struct/list/map reference, unusual column type, type mismatch), log a warning and return the unfiltered batch instead of aborting the scan. Velox's Option-A post-scan filter still evaluates the original predicate, so correctness is preserved — only the early-filter perf benefit is lost for that shape. End-to-end verification: TestMORFileSliceLayouts on the lin/eng38913-mor-prep branch of hudi-internal — the same test the original ENG-40156 Option A fix used as its verification gate — now passes with the new Choice A wire format + extended evaluator ("All tests passed", 1m39s, 02:24 UTC). Logs confirm pushdown fires across every column type ([ENG-40156] post-merge filter: 32 -> 1 rows per assertion). 31 hudi-cpp unit tests passing locally. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
…bility The CI vcpkg builder image used by adhoc_gluten_jar.yml doesn't have a system protoc available, so substrait 0.59's build script fails with "Could not find protoc". Enabling the crate's protoc feature pulls in protobuf-src, which builds protoc from source as part of cargo build. Adds ~30-60s to fresh hudi-cpp builds (protobuf-src compiles a small C++ toolchain once and caches it). Local builds still work; previously they relied on the dev container having /usr/local/bin/protoc on PATH. Verified locally: cargo build --lib finishes in 1m34s with the feature on, vs 20s with system protoc. Both produce identical libhudi.so. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
object_store::parse_url_opts only consults the options HashMap when building an AmazonS3 client — it never reads env vars. With no `region` key in options it falls back to the hardcoded `"us-east-1"` default at object_store-0.12.5/src/aws/builder.rs:1003, producing `https://s3.us-east-1.amazonaws.com/<bucket>/...` URLs that hit BareRedirect for buckets in any other region. In Quanton/Velox MOR reads on EKS, Spark sets AWS_REGION on the executor JVM via `spark.executorEnv.AWS_REGION=...` and the JNI'd Rust code inherits the same process env. But hudi-rs's call site at storage/mod.rs:Storage::new doesn't bridge env to options, so the env var is set yet useless. This patch adds a tiny `with_region_fallback` helper invoked before parse_url_opts. For s3:// / s3a:// URLs without a `region`/`aws_region` in options, it reads AWS_REGION (then AWS_DEFAULT_REGION) and injects the value as the `region` key. Non-S3 URLs and explicit-region callers are untouched. Tested: - 9 unit tests in storage::tests::test_region_fallback_*, all pass: - non-S3 scheme → passthrough - explicit `region` in options → preserved - explicit `aws_region` alias → preserved - AWS_REGION env → injected as `region` - AWS_DEFAULT_REGION env (no AWS_REGION) → injected - both env vars set → AWS_REGION wins - no env → passthrough - empty AWS_REGION → passthrough (don't propagate empty string) - s3a:// scheme → same injection path - Env-touching tests use `#[serial(env_vars)]` per existing convention (see crates/core/src/table/builder.rs). Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
How are the changes test-covered