yihua commented on code in PR #704: URL: https://github.com/apache/hudi-rs/pull/704#discussion_r3911378693
########## crates/core/src/file_group/base_file/hfile.rs: ########## @@ -0,0 +1,1587 @@ +/* + * 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. + */ + +//! HFile implementation of [`BaseFileReader`]. + +use std::sync::Arc; + +use arrow::array::{RecordBatch, RecordBatchOptions}; +use arrow_schema::{Schema, SchemaRef}; +use futures::StreamExt; +use futures::future::BoxFuture; + +use super::reader::{BaseFileReadOptions, BaseFileReader, BaseFileStream, KeyPredicate}; +use crate::file_group::log_file::avro::{AvroBlockDecoder, RegisteredWriterSchema}; +use crate::hfile::HFileReader; +use crate::hfile::record_key::fill_empty_entry_keys; +use crate::statistics::{StatisticsContainer, StatsGranularity}; +use crate::storage::Storage; +use crate::storage::error::{Result, StorageError}; +use crate::storage::file_metadata::FileMetadata; +use crate::util::arrow::project_batch_by_names; + +/// Records per Arrow batch while decoding an HFile's values. +const DECODE_BATCH_SIZE: usize = 1024; + +/// Only reached if a reader reports no budget, which a ranged reader always does. +const DEFAULT_WINDOW_BUDGET_FALLBACK: u64 = 16 * 1024 * 1024; + +/// Reads HFile base files. +#[derive(Debug)] +pub struct HFileBaseFileReader { + storage: Arc<Storage>, +} + +impl HFileBaseFileReader { + pub fn new(storage: Arc<Storage>) -> Self { + Self { storage } + } + + /// The record schema an HFile was written with, as Avro JSON and as Arrow. + /// + /// An HFile stores each value Avro-encoded and carries the schema it used in + /// its own file info. Decoding against that is what makes a base file and a + /// log block of the same table yield the same columns, which is the whole + /// reason they can merge; handing the value on as bytes does not. + /// + /// The decoder and the registration come back rather than being dropped, because + /// building a decoder is the dominant cost of reading a small HFile: `arrow_avro` + /// re-parses the writer schema's JSON on every construction, which for the + /// metadata table's eight-kilobyte record schema is more than reading the file. + /// The decoder has decoded nothing yet, so the first window can decode through + /// it; the registration is immutable and serves every later window, which then + /// pays only to build. + fn decoded_schema( + reader: &HFileReader, + relative_path: &str, + ) -> Result<(SchemaRef, AvroBlockDecoder, RegisteredWriterSchema)> { + let json = reader + .avro_schema_json() + .map_err(|e| { + StorageError::Creation(format!( + "Failed to read the Avro schema of HFile {relative_path}: {e:?}" + )) + })? + .ok_or_else(|| { + StorageError::Creation(format!( + "HFile {relative_path} carries no Avro schema, so its values cannot be decoded" + )) + })? + .to_string(); + // The schema comes from the decoder, not from converting the Avro JSON: + // `avro_to_arrow` does not handle named-type references, and the metadata + // table's record schema uses them. + let registered = RegisteredWriterSchema::new(&json) + .map_err(|e| StorageError::Creation(format!("{e}")))?; + let decoder = + AvroBlockDecoder::try_new_with_registered(®istered, None, DECODE_BATCH_SIZE) Review Comment: Has the HFile base path been exercised against an evolved schema? The values decode at the writer schema with no Avro writer-to-reader resolution (Java's `GenericDatumReader(writer, reader)` resolves), so an int-to-long promotion or added column relies entirely on the shared intersection/batch-evolution machinery — which looks right but has no HFile test behind it, and the missing schema-resolver arm this stack tracks separately is in the same territory. ########## crates/core/src/metadata/table/mod.rs: ########## @@ -272,7 +284,17 @@ impl Table { )); let storage = Storage::new(Arc::new(self.storage_options()), configs.clone())?; - reader::MetadataTableFileGroupReader::new(configs, storage) + // About 2.1x the cost of the reader it replaces on this table's own `files` + // partition, measured by `v2_reader::tests::reader_cost_on_a_metadata_slice`, + // and accepted: roughly a quarter of that is `arrow_avro` re-parsing the + // writer schema's JSON on every decoder construction, which no change here + // can remove while a flushed decoder cannot be reused (arrow-rs#10876). + // + // The fixture is also this reader's worst case, holding 2.6 records per block, + // where per-block fixed cost cannot amortise; on larger blocks it overtakes + // the reader it replaces. So the ratio above is not the production ratio, and + // nothing measured here establishes what that is. + v2_reader::MetadataTableV2Reader::new(configs, storage) Review Comment: non-blocking: Now that the production files-partition read goes through this layer, could the missing valid-instant filter get a tracked follow-up? Java gates every MDT log block on an EXACT_MATCH set from `getValidInstantTimestamps` (completed data-table instants + completed MDT deltacommits not pending on the data table + rollback-derived instants), so a data-table commit that fails after its MDT deltacommit completes is excluded there but admitted here — commit visibility catches it for file listing today, but nothing will for col-stats or RLI once they ride this layer. ########## crates/core/src/file_group/reader_v2/engine.rs: ########## @@ -410,6 +415,13 @@ impl HoodieFileGroupReader { } else { BaseFileFormatValue::from_str(&self.reader_context.base_file_format)? }; + // The shared factory refuses HFile, which is what keeps the legacy Review Comment: non-blocking: this comment predates the factory change later in this stack — `create_base_file_reader` now serves HFile, so "the shared factory refuses HFile" is no longer true and this arm duplicates it. Could the special case just fall through to the factory, taking the stale comment with it? -- 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]
