You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "linliu-code (via GitHub)" <gi...@apache.org> on 2026/08/07 08:34:21 UTC
[PR] refactor(core): decode delete blocks with arrow-avro, deleting the vendored converter [hudi-rs]
linliu-code opened a new pull request, #671:
URL: https://github.com/apache/hudi-rs/pull/671
**Stacked on #639–#670** — review only the last commit.
## What
Delete blocks were the **last** caller of the vendored Avro→Arrow array builder. Moving them to `arrow-avro` lets that builder go: **−1,073 lines**.
## Why now
That converter produced two silent-wrong-answer bugs while it was in use:
- an Avro map became `Dictionary(Utf8, V)` — not a valid Arrow type (#664)
- a decimal column read as **entirely NULL**, because `resolve_decimal128` missed `Value::Decimal` (#668)
Neither was on the delete path — delete records are `Utf8`, `Utf8`, and a narrowed primitive, which it handled correctly. But keeping a second decoder alive for one call site was not worth the risk of a third.
## The two shapes
**A delete block is one Avro datum**, not a run of framed records, so it decodes as a single row whose only column holds the list. The records come out of that list as a struct array, and the merge's batch is its columns lifted out.
**The ordering value needs unwrapping.** Hudi writes it as a union of per-type wrapper records, so decoding against that schema yields a union of one-field structs:
```
orderingVal: Union([(0,"null"), (1,"BooleanWrapper"), (2,"IntWrapper"), (3,"LongWrapper"), ...])
```
A block writes one ordering type, so exactly one branch is populated and its `value` child is the column.
**A block mixing branches is now rejected** rather than quietly reduced to one of them. The previous code read the type from the *first* record and narrowed the schema to it, applying that to every record in the block — so a mixed block was mis-decoded silently. This errors instead.
## What stays
`avro_to_arrow::schema` (378 lines). The schema resolver converts Avro schemas to Arrow well beyond log files, so only the row-oriented **array builder** is deleted, not the schema conversion.
Also unchanged: `schema::delete`'s `unwrap_ordering_value`, `avro_schema_for_delete_record` and `avro_schema_for_delete_record_list` are now unused by production code. They are `pub` in a `pub mod`, so removing them is a public-API change that deserves its own PR and a maintainer's call rather than being folded in here.
## Tests
No behavior should change, and the existing coverage is what says so: **11 delete-ordering harness cases** (long, decimal, timestamp, int, double, string), the delete gold fixtures, and the `table_read_tests` delete paths on the legacy engine — all unchanged and passing.
Two new unit tests on the genuinely new logic — union extraction: the populated branch is unwrapped to its values, and a mixed-branch block errors with a message naming the conflict.
Full workspace green: 1202 lib + 79 table-read + 39 datafusion + 21 + 12. Clippy clean in the changed files.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): delete the vendored Avro converter, and read columns promoted to string [hudi-rs]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #671:
URL: https://github.com/apache/hudi-rs/pull/671#discussion_r3779926958
##########
crates/core/src/file_group/reader.rs:
##########
@@ -1516,55 +1504,217 @@ mod tests {
Ok(())
}
+ /// The default must keep using the reader that has always served reads. If
+ /// this ever flips by accident, every other test still passes — they do not
+ /// say which engine produced their result — so it is asserted directly.
#[tokio::test]
- async fn test_read_metadata_table_files_partition() -> Result<()> {
- use crate::metadata::table_record::{FilesPartitionRecord, MetadataRecordType};
+ async fn merge_engine_defaults_to_legacy() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(base_url.as_ref(), empty_options()).await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
- let reader = create_metadata_table_reader()?;
- let file_slice = create_test_file_slice()?;
+ /// A metadata table ignores the setting. Its base files and log blocks are
+ /// HFile, which the merge-on-read reader cannot read, so honoring `v2`
+ /// there would fail somewhere deeper and less clearly.
+ #[tokio::test]
+ async fn metadata_table_stays_on_the_legacy_engine() -> Result<()> {
+ let configs = Arc::new(HudiConfigs::new([
+ (
+ HudiTableConfig::BasePath.as_ref(),
+ get_metadata_table_base_uri(),
+ ),
+ (HudiReadConfig::MergeEngine.as_ref(), "v2".to_string()),
+ ]));
+ let reader = FileGroupReader::new_with_overrides(configs, HashMap::new(), HashMap::new())?;
+ assert!(reader.is_metadata_table());
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
+
+ /// Asking for the merge-on-read engine routes to it. Without this the
+ /// switch could be wired to nothing and every test would still pass.
+ #[tokio::test]
+ async fn v2_engine_is_selected_when_asked_for() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(
+ base_url.as_ref(),
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::V2);
+ Ok(())
+ }
+
+ /// Selecting the merge-on-read engine must not change what a base-file-only
+ /// read returns.
+ ///
+ /// Both engines now serve this read — the merge-on-read one reduces to a
+ /// base file read — so this compares their output directly. It is the one
+ /// comparison they can be held to without settling the merge-mode question
+ /// first: with no log files there is no merge to disagree about.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_change_a_base_file_only_read() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
- // Test 1: Read all records (empty keys)
- let all_records = reader
- .read_metadata_table_files_partition(&file_slice, &[])
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
.await?;
- // Should have 4 keys after merging
- assert_eq!(
- all_records.len(),
- 4,
- "Should have 4 partition keys after merge"
- );
+ let v2_reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(v2_reader.merge_engine()?, MergeEngineValue::V2);
+ let v2 = v2_reader
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
+ .await?;
- // Validate all partition keys have correct record types
- for (key, record) in &all_records {
- if key == FilesPartitionRecord::ALL_PARTITIONS_KEY {
- assert_eq!(record.record_type, MetadataRecordType::AllPartitions);
- } else {
- assert_eq!(record.record_type, MetadataRecordType::Files);
- }
- }
+ assert_eq!(v2.num_rows(), legacy.num_rows());
+ assert_eq!(v2.schema(), legacy.schema());
+ assert_eq!(v2, legacy);
+ Ok(())
+ }
- // Validate chennai partition has files
- let chennai = all_records.get("city=chennai").unwrap();
- assert!(
- chennai.active_file_names().len() >= 2,
- "Chennai should have at least 2 active files"
- );
- assert!(chennai.total_size() > 0, "Total size should be > 0");
+ /// Read-optimized mode means the log files are not read, and asking for the
+ /// merge-on-read engine must not resurrect them.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_defeat_read_optimized_mode() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [
+ (HudiReadConfig::MergeEngine.as_ref(), "v2"),
+ (HudiReadConfig::UseReadOptimizedMode.as_ref(), "true"),
+ ],
+ )
+ .await?;
- // Test 2: Read specific keys
- let keys = vec![FilesPartitionRecord::ALL_PARTITIONS_KEY, "city=chennai"];
- let filtered_records = reader
- .read_metadata_table_files_partition(&file_slice, &keys)
+ // A path that cannot be read: the call only succeeds if the log files
+ // were dropped rather than handed to the engine.
+ let bogus_log = vec![".does-not-exist.log.1_0-0-0".to_string()];
+ let batch = reader
+ .read_file_slice_from_paths(&base_file_name, bogus_log, &ReadOptions::new())
.await?;
+ assert!(batch.num_rows() > 0);
+ Ok(())
+ }
- // Should only contain the requested keys
- assert_eq!(filtered_records.len(), 2);
- assert!(filtered_records.contains_key(FilesPartitionRecord::ALL_PARTITIONS_KEY));
- assert!(filtered_records.contains_key("city=chennai"));
- assert!(!filtered_records.contains_key("city=san_francisco"));
- assert!(!filtered_records.contains_key("city=sao_paulo"));
+ /// A slice with log files, read with the merge-on-read engine, actually
+ /// reads.
+ ///
+ /// Selecting an engine and reaching it are different things: every other
+ /// test here asserts the selection, which a dispatch wired to nothing would
+ /// still satisfy. This one goes through the dispatch to the engine and back
+ /// out through `apply_eager_options`.
+ #[tokio::test]
+ async fn v2_reads_a_slice_with_log_files_through_the_dispatch() -> Result<()> {
+ let (base_uri, partition, base_file_name, log_file_name) = v8_trips_mor_first_slice();
+ let base_path = format!("{partition}/{base_file_name}");
+ let log_path = format!("{partition}/{log_file_name}");
+
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ let batch = reader
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+
+ assert!(batch.num_rows() > 0, "expected a merged result");
+
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+ assert_eq!(batch.num_rows(), legacy.num_rows(), "row count differs");
+ Ok(())
+ }
+
+ /// No row an incremental read returns may sit outside its window.
+ ///
+ /// The slice is a compacted base file with a later log file. Compaction
+ /// merges records from many commits into one file while keeping each
+ /// record's own commit time, so a window that admits the *file* still has to
+ /// exclude most of its rows — the merge-on-read engine decides per file and
+ /// would otherwise return them all.
+ ///
+ /// This asserts the property rather than proving the difference: the
+ /// fixture's compacted slice holds a single record, and the log record in
+ /// the window replaces it, so the stale row is collapsed by the merge either
+ /// way. No fixture here has a compacted base row that survives unmatched,
+ /// which is what would make the two engines visibly disagree.
+ #[tokio::test]
+ async fn incremental_reads_return_nothing_outside_the_window() -> Result<()> {
Review Comment:
You were right, and this is the most useful comment on the stack — I verified it by mutation rather than by reading. Deleting the `apply_commit_time_filter` call from `read_via_v2` and rerunning this test: **it still passed.** The doc comment's own admission was accurate; the test guarded nothing.
Rewritten against the fixture you asked for. `V9MorCompactedIncremental` has a compacted base holding `c` and `d` at the same pre-window commit, with an in-window log record updating only `c` — so `d` survives the merge unmatched, carrying its out-of-window commit time, and only the filter can exclude it. The test is now `test_read_file_slice_from_paths_incremental_excludes_out_of_window_rows` and asserts the returned uuids are exactly `[a, b, c]` before checking the commit-time property.
Mutation-checked the new one: with the filter removed it fails with `left: ["a", "b", "c", "d"], right: ["a", "b", "c"]`. The doc comment now records why the fixture was swapped, so the vacuity can't quietly return.
##########
crates/core/src/file_group/log_file/avro.rs:
##########
@@ -16,77 +16,253 @@
* specific language governing permissions and limitations
* under the License.
*/
-use apache_avro::AvroResult;
-use apache_avro::types::Value as AvroValue;
-use apache_avro::{Schema as AvroSchema, from_avro_datum};
-use std::io::Read;
-
-pub struct AvroDataBlockContentReader<R: Read> {
- reader: R,
- writer_schema: AvroSchema,
- remaining_records: u32,
+use crate::Result;
+use crate::error::CoreError;
+use arrow_array::{ArrayRef, RecordBatch};
+use arrow_avro::reader::{Decoder as ArrowAvroDecoder, ReaderBuilder};
+use arrow_avro::schema::{AvroSchema as ArrowAvroSchema, SINGLE_OBJECT_MAGIC, SchemaStore};
+use arrow_cast::cast;
+use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use std::sync::Arc;
+
+/// Decodes the bare Avro record bodies in a Hudi log block straight into Arrow.
+///
+/// A Hudi data block frames each datum with a four-byte length and no Avro
+/// framing of its own, while `arrow-avro`'s decoder expects each record to be
+/// introduced by a Single Object Encoding prefix — the two-byte marker and the
+/// writer schema's fingerprint. The two are reconciled by synthesizing that
+/// prefix: the schema is registered once, and the ten bytes it yields are
+/// written ahead of every body.
+///
+/// The alternative is `arrow-avro`'s `AvroBodyDecoder`, which takes bare bodies
+/// directly but is not released yet.
+pub struct AvroBlockDecoder {
+ decoder: ArrowAvroDecoder,
+ /// Marker plus fingerprint, identical for every record in the block.
+ prefix: [u8; 10],
+ /// Reused across records so framing costs one copy, not one allocation.
+ framed: Vec<u8>,
+ /// When set, each batch is projected to this schema after decoding. See
+ /// [`AvroBlockDecoder::with_rewrite_to`].
+ rewrite_to: Option<SchemaRef>,
}
-impl<R: Read> AvroDataBlockContentReader<R> {
- pub fn new(reader: R, writer_schema: &AvroSchema, num_records: u32) -> Self {
- Self {
- reader,
- writer_schema: writer_schema.clone(),
- remaining_records: num_records,
+impl AvroBlockDecoder {
+ /// Build a decoder that resolves the block up to `reader_schema_json`.
+ ///
+ /// A log block records the schema it was written with, which may predate a
+ /// schema change on the table — an added column, or a column promoted from
+ /// `int` to `long`. Handing the reader schema to the decoder resolves the
+ /// block as it is read: absent columns are filled from their defaults and
+ /// promoted columns arrive in the promoted type.
+ ///
+ /// Only promotions Avro itself defines are handled this way. A block whose
+ /// writer schema differs in a way Avro does not allow is rejected here
+ /// rather than silently mis-read.
+ pub fn try_new_with_reader(
+ writer_schema_json: &str,
+ reader_schema_json: Option<&str>,
+ batch_size: usize,
+ ) -> Result<Self> {
+ let mut store = SchemaStore::new();
+ let fingerprint = store
+ .register(ArrowAvroSchema::new(writer_schema_json.to_string()))
+ .map_err(|e| {
+ CoreError::LogBlockError(format!("Failed to register block writer schema: {e}"))
+ })?;
+
+ let arrow_avro::schema::Fingerprint::Rabin(rabin) = fingerprint else {
+ return Err(CoreError::LogBlockError(format!(
+ "Expected a Rabin fingerprint for the block writer schema, got {fingerprint:?}"
+ )));
+ };
+
+ let mut builder = ReaderBuilder::new()
+ .with_writer_schema_store(store)
+ .with_active_fingerprint(fingerprint)
+ .with_batch_size(batch_size);
+ if let Some(reader_schema_json) = reader_schema_json {
+ builder =
+ builder.with_reader_schema(ArrowAvroSchema::new(reader_schema_json.to_string()));
}
+ let decoder = builder.build_decoder().map_err(|e| {
+ CoreError::LogBlockError(format!("Failed to build the Avro decoder: {e}"))
+ })?;
+
+ let mut prefix = [0u8; 10];
+ prefix[..2].copy_from_slice(&SINGLE_OBJECT_MAGIC);
+ prefix[2..].copy_from_slice(&rabin.to_le_bytes());
+
+ Ok(Self {
+ decoder,
+ prefix,
+ framed: Vec::new(),
+ rewrite_to: None,
+ })
+ }
+
+ /// Project each decoded batch to `schema` instead of resolving during the
+ /// read.
+ ///
+ /// Hudi permits promotions Avro does not — a number, or anything with a
+ /// logical type, to string. Avro refuses to build a reader for those, so a
+ /// block carrying one is decoded at the schema it was written with and
+ /// converted afterwards. Mirrors what the Java reader does when
+ /// `recordNeedsRewriteForExtendedAvroTypePromotion` says so: read
+ /// writer-to-writer, then promote.
+ pub fn with_rewrite_to(mut self, schema: Option<SchemaRef>) -> Self {
Review Comment:
Fixed. `with_rewrite_to(schema: SchemaRef)` now takes the concrete value, and the one caller in `content.rs` skips the call when there's nothing to rewrite to:
```rust
let mut decoder = AvroBlockDecoder::try_new_with_reader(...)?;
if let Some(rewrite_to) = rewrite_to {
decoder = decoder.with_rewrite_to(rewrite_to);
}
```
##########
crates/core/src/file_group/reader.rs:
##########
@@ -1516,55 +1504,217 @@ mod tests {
Ok(())
}
+ /// The default must keep using the reader that has always served reads. If
+ /// this ever flips by accident, every other test still passes — they do not
+ /// say which engine produced their result — so it is asserted directly.
#[tokio::test]
- async fn test_read_metadata_table_files_partition() -> Result<()> {
- use crate::metadata::table_record::{FilesPartitionRecord, MetadataRecordType};
+ async fn merge_engine_defaults_to_legacy() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(base_url.as_ref(), empty_options()).await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
- let reader = create_metadata_table_reader()?;
- let file_slice = create_test_file_slice()?;
+ /// A metadata table ignores the setting. Its base files and log blocks are
+ /// HFile, which the merge-on-read reader cannot read, so honoring `v2`
+ /// there would fail somewhere deeper and less clearly.
+ #[tokio::test]
+ async fn metadata_table_stays_on_the_legacy_engine() -> Result<()> {
+ let configs = Arc::new(HudiConfigs::new([
+ (
+ HudiTableConfig::BasePath.as_ref(),
+ get_metadata_table_base_uri(),
+ ),
+ (HudiReadConfig::MergeEngine.as_ref(), "v2".to_string()),
+ ]));
+ let reader = FileGroupReader::new_with_overrides(configs, HashMap::new(), HashMap::new())?;
+ assert!(reader.is_metadata_table());
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
+
+ /// Asking for the merge-on-read engine routes to it. Without this the
+ /// switch could be wired to nothing and every test would still pass.
+ #[tokio::test]
+ async fn v2_engine_is_selected_when_asked_for() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(
+ base_url.as_ref(),
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::V2);
+ Ok(())
+ }
+
+ /// Selecting the merge-on-read engine must not change what a base-file-only
+ /// read returns.
+ ///
+ /// Both engines now serve this read — the merge-on-read one reduces to a
+ /// base file read — so this compares their output directly. It is the one
+ /// comparison they can be held to without settling the merge-mode question
+ /// first: with no log files there is no merge to disagree about.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_change_a_base_file_only_read() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
- // Test 1: Read all records (empty keys)
- let all_records = reader
- .read_metadata_table_files_partition(&file_slice, &[])
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
.await?;
- // Should have 4 keys after merging
- assert_eq!(
- all_records.len(),
- 4,
- "Should have 4 partition keys after merge"
- );
+ let v2_reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(v2_reader.merge_engine()?, MergeEngineValue::V2);
+ let v2 = v2_reader
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
+ .await?;
- // Validate all partition keys have correct record types
- for (key, record) in &all_records {
- if key == FilesPartitionRecord::ALL_PARTITIONS_KEY {
- assert_eq!(record.record_type, MetadataRecordType::AllPartitions);
- } else {
- assert_eq!(record.record_type, MetadataRecordType::Files);
- }
- }
+ assert_eq!(v2.num_rows(), legacy.num_rows());
+ assert_eq!(v2.schema(), legacy.schema());
+ assert_eq!(v2, legacy);
+ Ok(())
+ }
- // Validate chennai partition has files
- let chennai = all_records.get("city=chennai").unwrap();
- assert!(
- chennai.active_file_names().len() >= 2,
- "Chennai should have at least 2 active files"
- );
- assert!(chennai.total_size() > 0, "Total size should be > 0");
+ /// Read-optimized mode means the log files are not read, and asking for the
+ /// merge-on-read engine must not resurrect them.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_defeat_read_optimized_mode() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [
+ (HudiReadConfig::MergeEngine.as_ref(), "v2"),
+ (HudiReadConfig::UseReadOptimizedMode.as_ref(), "true"),
+ ],
+ )
+ .await?;
- // Test 2: Read specific keys
- let keys = vec![FilesPartitionRecord::ALL_PARTITIONS_KEY, "city=chennai"];
- let filtered_records = reader
- .read_metadata_table_files_partition(&file_slice, &keys)
+ // A path that cannot be read: the call only succeeds if the log files
+ // were dropped rather than handed to the engine.
+ let bogus_log = vec![".does-not-exist.log.1_0-0-0".to_string()];
+ let batch = reader
+ .read_file_slice_from_paths(&base_file_name, bogus_log, &ReadOptions::new())
.await?;
+ assert!(batch.num_rows() > 0);
+ Ok(())
+ }
- // Should only contain the requested keys
- assert_eq!(filtered_records.len(), 2);
- assert!(filtered_records.contains_key(FilesPartitionRecord::ALL_PARTITIONS_KEY));
- assert!(filtered_records.contains_key("city=chennai"));
- assert!(!filtered_records.contains_key("city=san_francisco"));
- assert!(!filtered_records.contains_key("city=sao_paulo"));
+ /// A slice with log files, read with the merge-on-read engine, actually
+ /// reads.
+ ///
+ /// Selecting an engine and reaching it are different things: every other
+ /// test here asserts the selection, which a dispatch wired to nothing would
+ /// still satisfy. This one goes through the dispatch to the engine and back
+ /// out through `apply_eager_options`.
+ #[tokio::test]
+ async fn v2_reads_a_slice_with_log_files_through_the_dispatch() -> Result<()> {
+ let (base_uri, partition, base_file_name, log_file_name) = v8_trips_mor_first_slice();
+ let base_path = format!("{partition}/{base_file_name}");
+ let log_path = format!("{partition}/{log_file_name}");
+
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ let batch = reader
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+
+ assert!(batch.num_rows() > 0, "expected a merged result");
+
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+ assert_eq!(batch.num_rows(), legacy.num_rows(), "row count differs");
+ Ok(())
+ }
+
+ /// No row an incremental read returns may sit outside its window.
+ ///
+ /// The slice is a compacted base file with a later log file. Compaction
+ /// merges records from many commits into one file while keeping each
+ /// record's own commit time, so a window that admits the *file* still has to
+ /// exclude most of its rows — the merge-on-read engine decides per file and
+ /// would otherwise return them all.
+ ///
+ /// This asserts the property rather than proving the difference: the
+ /// fixture's compacted slice holds a single record, and the log record in
+ /// the window replaces it, so the stale row is collapsed by the merge either
+ /// way. No fixture here has a compacted base row that survives unmatched,
+ /// which is what would make the two engines visibly disagree.
+ #[tokio::test]
Review Comment:
Fixed — it's now `test_read_file_slice_from_paths_incremental_excludes_out_of_window_rows`, which follows the `test_<function>_<scenario>_<expected>` convention and is close to what you suggested. (The test body was rewritten too, for the reason in your other comment on this file.)
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): delete the vendored Avro converter, and read columns promoted to string [hudi-rs]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on PR #671:
URL: https://github.com/apache/hudi-rs/pull/671#issuecomment-5302610960
Superseded by #660, which merged as `75b188e`. This PR's content is in that commit — the ten PRs in this series were the reviewable decomposition of the port, and shipping went through the squashed #660 instead.
Every review thread here was answered before that merge; the replies stay on the thread for the record. Closing to keep the queue honest about what is still in flight.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): delete the vendored Avro converter, and read columns promoted to string [hudi-rs]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #671:
URL: https://github.com/apache/hudi-rs/pull/671#discussion_r3773237520
##########
crates/core/src/file_group/reader.rs:
##########
@@ -1516,55 +1504,217 @@ mod tests {
Ok(())
}
+ /// The default must keep using the reader that has always served reads. If
+ /// this ever flips by accident, every other test still passes — they do not
+ /// say which engine produced their result — so it is asserted directly.
#[tokio::test]
- async fn test_read_metadata_table_files_partition() -> Result<()> {
- use crate::metadata::table_record::{FilesPartitionRecord, MetadataRecordType};
+ async fn merge_engine_defaults_to_legacy() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(base_url.as_ref(), empty_options()).await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
- let reader = create_metadata_table_reader()?;
- let file_slice = create_test_file_slice()?;
+ /// A metadata table ignores the setting. Its base files and log blocks are
+ /// HFile, which the merge-on-read reader cannot read, so honoring `v2`
+ /// there would fail somewhere deeper and less clearly.
+ #[tokio::test]
+ async fn metadata_table_stays_on_the_legacy_engine() -> Result<()> {
+ let configs = Arc::new(HudiConfigs::new([
+ (
+ HudiTableConfig::BasePath.as_ref(),
+ get_metadata_table_base_uri(),
+ ),
+ (HudiReadConfig::MergeEngine.as_ref(), "v2".to_string()),
+ ]));
+ let reader = FileGroupReader::new_with_overrides(configs, HashMap::new(), HashMap::new())?;
+ assert!(reader.is_metadata_table());
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
+
+ /// Asking for the merge-on-read engine routes to it. Without this the
+ /// switch could be wired to nothing and every test would still pass.
+ #[tokio::test]
+ async fn v2_engine_is_selected_when_asked_for() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(
+ base_url.as_ref(),
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::V2);
+ Ok(())
+ }
+
+ /// Selecting the merge-on-read engine must not change what a base-file-only
+ /// read returns.
+ ///
+ /// Both engines now serve this read — the merge-on-read one reduces to a
+ /// base file read — so this compares their output directly. It is the one
+ /// comparison they can be held to without settling the merge-mode question
+ /// first: with no log files there is no merge to disagree about.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_change_a_base_file_only_read() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
- // Test 1: Read all records (empty keys)
- let all_records = reader
- .read_metadata_table_files_partition(&file_slice, &[])
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
.await?;
- // Should have 4 keys after merging
- assert_eq!(
- all_records.len(),
- 4,
- "Should have 4 partition keys after merge"
- );
+ let v2_reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(v2_reader.merge_engine()?, MergeEngineValue::V2);
+ let v2 = v2_reader
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
+ .await?;
- // Validate all partition keys have correct record types
- for (key, record) in &all_records {
- if key == FilesPartitionRecord::ALL_PARTITIONS_KEY {
- assert_eq!(record.record_type, MetadataRecordType::AllPartitions);
- } else {
- assert_eq!(record.record_type, MetadataRecordType::Files);
- }
- }
+ assert_eq!(v2.num_rows(), legacy.num_rows());
+ assert_eq!(v2.schema(), legacy.schema());
+ assert_eq!(v2, legacy);
+ Ok(())
+ }
- // Validate chennai partition has files
- let chennai = all_records.get("city=chennai").unwrap();
- assert!(
- chennai.active_file_names().len() >= 2,
- "Chennai should have at least 2 active files"
- );
- assert!(chennai.total_size() > 0, "Total size should be > 0");
+ /// Read-optimized mode means the log files are not read, and asking for the
+ /// merge-on-read engine must not resurrect them.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_defeat_read_optimized_mode() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [
+ (HudiReadConfig::MergeEngine.as_ref(), "v2"),
+ (HudiReadConfig::UseReadOptimizedMode.as_ref(), "true"),
+ ],
+ )
+ .await?;
- // Test 2: Read specific keys
- let keys = vec![FilesPartitionRecord::ALL_PARTITIONS_KEY, "city=chennai"];
- let filtered_records = reader
- .read_metadata_table_files_partition(&file_slice, &keys)
+ // A path that cannot be read: the call only succeeds if the log files
+ // were dropped rather than handed to the engine.
+ let bogus_log = vec![".does-not-exist.log.1_0-0-0".to_string()];
+ let batch = reader
+ .read_file_slice_from_paths(&base_file_name, bogus_log, &ReadOptions::new())
.await?;
+ assert!(batch.num_rows() > 0);
+ Ok(())
+ }
- // Should only contain the requested keys
- assert_eq!(filtered_records.len(), 2);
- assert!(filtered_records.contains_key(FilesPartitionRecord::ALL_PARTITIONS_KEY));
- assert!(filtered_records.contains_key("city=chennai"));
- assert!(!filtered_records.contains_key("city=san_francisco"));
- assert!(!filtered_records.contains_key("city=sao_paulo"));
+ /// A slice with log files, read with the merge-on-read engine, actually
+ /// reads.
+ ///
+ /// Selecting an engine and reaching it are different things: every other
+ /// test here asserts the selection, which a dispatch wired to nothing would
+ /// still satisfy. This one goes through the dispatch to the engine and back
+ /// out through `apply_eager_options`.
+ #[tokio::test]
+ async fn v2_reads_a_slice_with_log_files_through_the_dispatch() -> Result<()> {
+ let (base_uri, partition, base_file_name, log_file_name) = v8_trips_mor_first_slice();
+ let base_path = format!("{partition}/{base_file_name}");
+ let log_path = format!("{partition}/{log_file_name}");
+
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ let batch = reader
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+
+ assert!(batch.num_rows() > 0, "expected a merged result");
+
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+ assert_eq!(batch.num_rows(), legacy.num_rows(), "row count differs");
+ Ok(())
+ }
+
+ /// No row an incremental read returns may sit outside its window.
+ ///
+ /// The slice is a compacted base file with a later log file. Compaction
+ /// merges records from many commits into one file while keeping each
+ /// record's own commit time, so a window that admits the *file* still has to
+ /// exclude most of its rows — the merge-on-read engine decides per file and
+ /// would otherwise return them all.
+ ///
+ /// This asserts the property rather than proving the difference: the
+ /// fixture's compacted slice holds a single record, and the log record in
+ /// the window replaces it, so the stale row is collapsed by the merge either
+ /// way. No fixture here has a compacted base row that survives unmatched,
+ /// which is what would make the two engines visibly disagree.
+ #[tokio::test]
+ async fn incremental_reads_return_nothing_outside_the_window() -> Result<()> {
Review Comment:
🤖 As the doc comment candidly notes, the fixture's compacted base holds a single record that the in-window log record replaces — so this test would still pass with the `apply_commit_time_filter` call removed and doesn't actually guard the fix. Would it be feasible to use (or synthesize) a slice where a compacted base row with an out-of-window commit time survives unmatched, so the assertion fails without the filter?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
crates/core/src/file_group/reader.rs:
##########
@@ -1516,55 +1504,217 @@ mod tests {
Ok(())
}
+ /// The default must keep using the reader that has always served reads. If
+ /// this ever flips by accident, every other test still passes — they do not
+ /// say which engine produced their result — so it is asserted directly.
#[tokio::test]
- async fn test_read_metadata_table_files_partition() -> Result<()> {
- use crate::metadata::table_record::{FilesPartitionRecord, MetadataRecordType};
+ async fn merge_engine_defaults_to_legacy() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(base_url.as_ref(), empty_options()).await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
- let reader = create_metadata_table_reader()?;
- let file_slice = create_test_file_slice()?;
+ /// A metadata table ignores the setting. Its base files and log blocks are
+ /// HFile, which the merge-on-read reader cannot read, so honoring `v2`
+ /// there would fail somewhere deeper and less clearly.
+ #[tokio::test]
+ async fn metadata_table_stays_on_the_legacy_engine() -> Result<()> {
+ let configs = Arc::new(HudiConfigs::new([
+ (
+ HudiTableConfig::BasePath.as_ref(),
+ get_metadata_table_base_uri(),
+ ),
+ (HudiReadConfig::MergeEngine.as_ref(), "v2".to_string()),
+ ]));
+ let reader = FileGroupReader::new_with_overrides(configs, HashMap::new(), HashMap::new())?;
+ assert!(reader.is_metadata_table());
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::Legacy);
+ Ok(())
+ }
+
+ /// Asking for the merge-on-read engine routes to it. Without this the
+ /// switch could be wired to nothing and every test would still pass.
+ #[tokio::test]
+ async fn v2_engine_is_selected_when_asked_for() -> Result<()> {
+ use hudi_test::SampleTable;
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_mor_avro();
+ let reader = FileGroupReader::new_with_options(
+ base_url.as_ref(),
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(reader.merge_engine()?, MergeEngineValue::V2);
+ Ok(())
+ }
+
+ /// Selecting the merge-on-read engine must not change what a base-file-only
+ /// read returns.
+ ///
+ /// Both engines now serve this read — the merge-on-read one reduces to a
+ /// base file read — so this compares their output directly. It is the one
+ /// comparison they can be held to without settling the merge-mode question
+ /// first: with no log files there is no merge to disagree about.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_change_a_base_file_only_read() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
- // Test 1: Read all records (empty keys)
- let all_records = reader
- .read_metadata_table_files_partition(&file_slice, &[])
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
.await?;
- // Should have 4 keys after merging
- assert_eq!(
- all_records.len(),
- 4,
- "Should have 4 partition keys after merge"
- );
+ let v2_reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ assert_eq!(v2_reader.merge_engine()?, MergeEngineValue::V2);
+ let v2 = v2_reader
+ .read_file_slice_from_paths(&base_file_name, Vec::<&str>::new(), &ReadOptions::new())
+ .await?;
- // Validate all partition keys have correct record types
- for (key, record) in &all_records {
- if key == FilesPartitionRecord::ALL_PARTITIONS_KEY {
- assert_eq!(record.record_type, MetadataRecordType::AllPartitions);
- } else {
- assert_eq!(record.record_type, MetadataRecordType::Files);
- }
- }
+ assert_eq!(v2.num_rows(), legacy.num_rows());
+ assert_eq!(v2.schema(), legacy.schema());
+ assert_eq!(v2, legacy);
+ Ok(())
+ }
- // Validate chennai partition has files
- let chennai = all_records.get("city=chennai").unwrap();
- assert!(
- chennai.active_file_names().len() >= 2,
- "Chennai should have at least 2 active files"
- );
- assert!(chennai.total_size() > 0, "Total size should be > 0");
+ /// Read-optimized mode means the log files are not read, and asking for the
+ /// merge-on-read engine must not resurrect them.
+ #[tokio::test]
+ async fn asking_for_v2_does_not_defeat_read_optimized_mode() -> Result<()> {
+ let (base_uri, base_file_name) = v8np_base_uri_and_first_parquet();
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [
+ (HudiReadConfig::MergeEngine.as_ref(), "v2"),
+ (HudiReadConfig::UseReadOptimizedMode.as_ref(), "true"),
+ ],
+ )
+ .await?;
- // Test 2: Read specific keys
- let keys = vec![FilesPartitionRecord::ALL_PARTITIONS_KEY, "city=chennai"];
- let filtered_records = reader
- .read_metadata_table_files_partition(&file_slice, &keys)
+ // A path that cannot be read: the call only succeeds if the log files
+ // were dropped rather than handed to the engine.
+ let bogus_log = vec![".does-not-exist.log.1_0-0-0".to_string()];
+ let batch = reader
+ .read_file_slice_from_paths(&base_file_name, bogus_log, &ReadOptions::new())
.await?;
+ assert!(batch.num_rows() > 0);
+ Ok(())
+ }
- // Should only contain the requested keys
- assert_eq!(filtered_records.len(), 2);
- assert!(filtered_records.contains_key(FilesPartitionRecord::ALL_PARTITIONS_KEY));
- assert!(filtered_records.contains_key("city=chennai"));
- assert!(!filtered_records.contains_key("city=san_francisco"));
- assert!(!filtered_records.contains_key("city=sao_paulo"));
+ /// A slice with log files, read with the merge-on-read engine, actually
+ /// reads.
+ ///
+ /// Selecting an engine and reaching it are different things: every other
+ /// test here asserts the selection, which a dispatch wired to nothing would
+ /// still satisfy. This one goes through the dispatch to the engine and back
+ /// out through `apply_eager_options`.
+ #[tokio::test]
+ async fn v2_reads_a_slice_with_log_files_through_the_dispatch() -> Result<()> {
+ let (base_uri, partition, base_file_name, log_file_name) = v8_trips_mor_first_slice();
+ let base_path = format!("{partition}/{base_file_name}");
+ let log_path = format!("{partition}/{log_file_name}");
+
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ [(HudiReadConfig::MergeEngine.as_ref(), "v2")],
+ )
+ .await?;
+ let batch = reader
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+
+ assert!(batch.num_rows() > 0, "expected a merged result");
+
+ let legacy = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await?
+ .read_file_slice_from_paths(&base_path, vec![log_path.clone()], &ReadOptions::new())
+ .await?;
+ assert_eq!(batch.num_rows(), legacy.num_rows(), "row count differs");
+ Ok(())
+ }
+
+ /// No row an incremental read returns may sit outside its window.
+ ///
+ /// The slice is a compacted base file with a later log file. Compaction
+ /// merges records from many commits into one file while keeping each
+ /// record's own commit time, so a window that admits the *file* still has to
+ /// exclude most of its rows — the merge-on-read engine decides per file and
+ /// would otherwise return them all.
+ ///
+ /// This asserts the property rather than proving the difference: the
+ /// fixture's compacted slice holds a single record, and the log record in
+ /// the window replaces it, so the stale row is collapsed by the merge either
+ /// way. No fixture here has a compacted base row that survives unmatched,
+ /// which is what would make the two engines visibly disagree.
+ #[tokio::test]
Review Comment:
🤖 nit: the test name reads like a property statement rather than a test scenario — something like `test_read_file_slice_incremental_window_excludes_out_of_range_rows` would better follow the `test_<function>_<scenario>_<expected>` convention used elsewhere in the file.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
crates/core/src/file_group/log_file/avro.rs:
##########
@@ -16,77 +16,253 @@
* specific language governing permissions and limitations
* under the License.
*/
-use apache_avro::AvroResult;
-use apache_avro::types::Value as AvroValue;
-use apache_avro::{Schema as AvroSchema, from_avro_datum};
-use std::io::Read;
-
-pub struct AvroDataBlockContentReader<R: Read> {
- reader: R,
- writer_schema: AvroSchema,
- remaining_records: u32,
+use crate::Result;
+use crate::error::CoreError;
+use arrow_array::{ArrayRef, RecordBatch};
+use arrow_avro::reader::{Decoder as ArrowAvroDecoder, ReaderBuilder};
+use arrow_avro::schema::{AvroSchema as ArrowAvroSchema, SINGLE_OBJECT_MAGIC, SchemaStore};
+use arrow_cast::cast;
+use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use std::sync::Arc;
+
+/// Decodes the bare Avro record bodies in a Hudi log block straight into Arrow.
+///
+/// A Hudi data block frames each datum with a four-byte length and no Avro
+/// framing of its own, while `arrow-avro`'s decoder expects each record to be
+/// introduced by a Single Object Encoding prefix — the two-byte marker and the
+/// writer schema's fingerprint. The two are reconciled by synthesizing that
+/// prefix: the schema is registered once, and the ten bytes it yields are
+/// written ahead of every body.
+///
+/// The alternative is `arrow-avro`'s `AvroBodyDecoder`, which takes bare bodies
+/// directly but is not released yet.
+pub struct AvroBlockDecoder {
+ decoder: ArrowAvroDecoder,
+ /// Marker plus fingerprint, identical for every record in the block.
+ prefix: [u8; 10],
+ /// Reused across records so framing costs one copy, not one allocation.
+ framed: Vec<u8>,
+ /// When set, each batch is projected to this schema after decoding. See
+ /// [`AvroBlockDecoder::with_rewrite_to`].
+ rewrite_to: Option<SchemaRef>,
}
-impl<R: Read> AvroDataBlockContentReader<R> {
- pub fn new(reader: R, writer_schema: &AvroSchema, num_records: u32) -> Self {
- Self {
- reader,
- writer_schema: writer_schema.clone(),
- remaining_records: num_records,
+impl AvroBlockDecoder {
+ /// Build a decoder that resolves the block up to `reader_schema_json`.
+ ///
+ /// A log block records the schema it was written with, which may predate a
+ /// schema change on the table — an added column, or a column promoted from
+ /// `int` to `long`. Handing the reader schema to the decoder resolves the
+ /// block as it is read: absent columns are filled from their defaults and
+ /// promoted columns arrive in the promoted type.
+ ///
+ /// Only promotions Avro itself defines are handled this way. A block whose
+ /// writer schema differs in a way Avro does not allow is rejected here
+ /// rather than silently mis-read.
+ pub fn try_new_with_reader(
+ writer_schema_json: &str,
+ reader_schema_json: Option<&str>,
+ batch_size: usize,
+ ) -> Result<Self> {
+ let mut store = SchemaStore::new();
+ let fingerprint = store
+ .register(ArrowAvroSchema::new(writer_schema_json.to_string()))
+ .map_err(|e| {
+ CoreError::LogBlockError(format!("Failed to register block writer schema: {e}"))
+ })?;
+
+ let arrow_avro::schema::Fingerprint::Rabin(rabin) = fingerprint else {
+ return Err(CoreError::LogBlockError(format!(
+ "Expected a Rabin fingerprint for the block writer schema, got {fingerprint:?}"
+ )));
+ };
+
+ let mut builder = ReaderBuilder::new()
+ .with_writer_schema_store(store)
+ .with_active_fingerprint(fingerprint)
+ .with_batch_size(batch_size);
+ if let Some(reader_schema_json) = reader_schema_json {
+ builder =
+ builder.with_reader_schema(ArrowAvroSchema::new(reader_schema_json.to_string()));
}
+ let decoder = builder.build_decoder().map_err(|e| {
+ CoreError::LogBlockError(format!("Failed to build the Avro decoder: {e}"))
+ })?;
+
+ let mut prefix = [0u8; 10];
+ prefix[..2].copy_from_slice(&SINGLE_OBJECT_MAGIC);
+ prefix[2..].copy_from_slice(&rabin.to_le_bytes());
+
+ Ok(Self {
+ decoder,
+ prefix,
+ framed: Vec::new(),
+ rewrite_to: None,
+ })
+ }
+
+ /// Project each decoded batch to `schema` instead of resolving during the
+ /// read.
+ ///
+ /// Hudi permits promotions Avro does not — a number, or anything with a
+ /// logical type, to string. Avro refuses to build a reader for those, so a
+ /// block carrying one is decoded at the schema it was written with and
+ /// converted afterwards. Mirrors what the Java reader does when
+ /// `recordNeedsRewriteForExtendedAvroTypePromotion` says so: read
+ /// writer-to-writer, then promote.
+ pub fn with_rewrite_to(mut self, schema: Option<SchemaRef>) -> Self {
Review Comment:
🤖 nit: builder methods usually take the concrete value, not an `Option` — could you make this `with_rewrite_to(schema: SchemaRef) -> Self` and let callers simply skip the call when there's nothing to rewrite to? The current signature makes `.with_rewrite_to(None)` valid but semantically meaningless.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): delete the vendored Avro converter, and read columns promoted to string [hudi-rs]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code closed pull request #671: refactor(core): delete the vendored Avro converter, and read columns promoted to string
URL: https://github.com/apache/hudi-rs/pull/671
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org