linliu-code commented on code in PR #669:
URL: https://github.com/apache/hudi-rs/pull/669#discussion_r3779924109
##########
crates/core/src/file_group/log_file/content.rs:
##########
@@ -110,23 +152,48 @@ impl Decoder {
) -> Result<RecordBatches> {
Decoder::validate_log_block_version(&mut reader)?;
- let writer_schema =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
+ let writer_schema_json =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
CoreError::LogBlockError("Schema not found in block
header".to_string())
})?;
- let writer_schema = Arc::new(AvroSchema::parse_str(writer_schema)?);
let mut record_count_buf = [0u8; 4];
reader.read_exact(&mut record_count_buf)?;
let record_count = u32::from_be_bytes(record_count_buf);
- let record_content_reader =
- AvroDataBlockContentReader::new(reader, writer_schema.as_ref(),
record_count);
- let mut avro_arrow_array_reader =
- AvroArrowArrayReader::try_new(record_content_reader,
writer_schema.as_ref())?;
+ // A partial-update block carries only the columns that were written,
and
+ // the merge needs to know which those are. Resolving it up to the
table
+ // schema would fabricate the rest, so it decodes against its own
schema.
+ let is_partial = header.contains_key(&BlockMetadataKey::IsPartial);
Review Comment:
Agreed and fixed — this now parses the value rather than testing for the key:
```rust
let is_partial = header
.get(&BlockMetadataKey::IsPartial)
.is_some_and(|v| v.eq_ignore_ascii_case("true"));
```
with a comment naming Java's `Boolean.parseBoolean(getOrDefault(IS_PARTIAL,
"false"))` as the reason. Your failure mode is exactly right: a writer emitting
`IS_PARTIAL=false` on a full block would have had it decoded writer-only,
skipping the resolution up to the reader schema and reintroducing the int→long
mismatch at merge time on an evolved table.
Added `test_decode_avro_is_partial_false_resolves_to_the_reader_schema` — a
block written narrow with `IS_PARTIAL=false`, whose reader schema promotes `id`
to `long` and adds a column. It asserts both arrive, so it fails if the flag is
read by presence again. Mutation-checked: reverting to `contains_key` makes it
fail.
##########
crates/core/src/file_group/log_file/content.rs:
##########
@@ -39,13 +46,48 @@ use std::sync::Arc;
pub struct Decoder {
batch_size: usize,
hudi_configs: Arc<HudiConfigs>,
+ /// Predicate to push into a parquet log block, when the caller has decided
+ /// it is safe to evaluate before the merge. See
+ /// [`Decoder::with_row_filter`].
+ row_filter: Option<RowFilterBuilder>,
+ /// Schema an Avro block is resolved up to, as Avro JSON. See
Review Comment:
Fixed. `required_schema_json` → `reader_schema_json` and
`with_required_schema` → `with_reader_schema`, applied consistently across
`Decoder` and `LogFileReader` so both layers use Avro's own term for the
resolving schema (see also your other comment on the same rename).
##########
crates/core/src/file_group/log_file/content.rs:
##########
@@ -110,23 +152,48 @@ impl Decoder {
) -> Result<RecordBatches> {
Decoder::validate_log_block_version(&mut reader)?;
- let writer_schema =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
+ let writer_schema_json =
header.get(&BlockMetadataKey::Schema).ok_or_else(|| {
CoreError::LogBlockError("Schema not found in block
header".to_string())
})?;
- let writer_schema = Arc::new(AvroSchema::parse_str(writer_schema)?);
let mut record_count_buf = [0u8; 4];
reader.read_exact(&mut record_count_buf)?;
let record_count = u32::from_be_bytes(record_count_buf);
- let record_content_reader =
- AvroDataBlockContentReader::new(reader, writer_schema.as_ref(),
record_count);
- let mut avro_arrow_array_reader =
- AvroArrowArrayReader::try_new(record_content_reader,
writer_schema.as_ref())?;
+ // A partial-update block carries only the columns that were written,
and
+ // the merge needs to know which those are. Resolving it up to the
table
+ // schema would fabricate the rest, so it decodes against its own
schema.
+ let is_partial = header.contains_key(&BlockMetadataKey::IsPartial);
+ let reader_schema_json = if is_partial {
+ None
+ } else {
+ self.required_schema_json.as_deref()
+ };
+ let mut decoder = AvroBlockDecoder::try_new_with_reader(
Review Comment:
Checked this one against the fixture that has exactly the promotion you
describe, and I don't think base rows and log rows diverge.
`table_evo_promotion [MorAvro]` promotes both an `int`→`long` column (`num`)
and a `float`→`double` column (`fnum`), and it genuinely mixes the two sources:
the compacted base holds pre-promotion rows and a log record supplies
`5000000000`. It ships a `gold_options` manifest and is compared against
Spark's own snapshot cell-by-cell across every option case — and its only entry
on the known-disagreements list is for reader **version 1** ("version 1 cannot
widen a promoted column"). Under version 2 it matches Hudi on every case, which
it could not do if the Avro-resolved log doubles and the string-mediated base
doubles disagreed for `fnum`.
You're right that the int→long promotion in the unit test is exact and so
wouldn't surface this on its own; the gold fixture is what covers it. No change
made.
--
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]
