This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 0aa3b0f7 fix(reader): apply the logical timestamp repair to parquet
log block records (#815)
0aa3b0f7 is described below
commit 0aa3b0f78200166ec05e1e590e52e73f3324d12f
Author: Y Ethan Guo <[email protected]>
AuthorDate: Wed Oct 28 15:51:26 2026 -0700
fix(reader): apply the logical timestamp repair to parquet log block
records (#815)
A parquet log block reaches the merge in its file schema, and the drain
reconcile refused the pairings Java repairs with SchemaRepair: tz-aware
micros labelling millis values, and a plain long the table declares as a
local timestamp. MOR reads with such log blocks failed with "reconcile:
column 'ts_millis' type Timestamp(us, "UTC") incompatible with target
Timestamp(ms, "UTC")".
Reconcile now relabels exactly those leaf pairings, at any depth, keeping
the stored i64, which matches the base file path and Java. Every other
leaf difference is still refused.
A PK-safe row filter pushed into such a block would be evaluated on the
mislabelled values and drop matching rows before the merge. The decoder
now withholds the row filter from a parquet block whose footer needs the
repair against the table schema, and from every parquet block when the
table schema is unknown, as the base file read withholds it for a
mislabelled base file.
---
crates/core/src/file_group/log_file/content.rs | 181 +++++++++++-
.../file_group/reader_v2/buffer/row_extraction.rs | 309 ++++++++++++++++++++-
.../src/file_group/reader_v2/log_record_reader.rs | 84 ++++++
crates/core/src/schema/batch_evolution.rs | 212 ++++++++++++++
4 files changed, 776 insertions(+), 10 deletions(-)
diff --git a/crates/core/src/file_group/log_file/content.rs
b/crates/core/src/file_group/log_file/content.rs
index 553e18a7..25618585 100644
--- a/crates/core/src/file_group/log_file/content.rs
+++ b/crates/core/src/file_group/log_file/content.rs
@@ -138,6 +138,9 @@ pub struct Decoder {
/// it is safe to evaluate before the merge. See
/// [`Decoder::with_row_filter`].
row_filter: Option<RowFilterBuilder>,
+ /// Schema a parquet block's columns are checked against before the row
+ /// filter is pushed. See [`Decoder::with_repair_table_schema`].
+ repair_table_schema: Option<arrow_schema::SchemaRef>,
/// Schema an Avro block is resolved up to, as Avro JSON. See
/// [`Decoder::with_reader_schema`].
reader_schema_json: Option<String>,
@@ -174,6 +177,30 @@ impl Decoder {
self
}
+ /// The schema a parquet block is read as, for deciding whether the row
filter
+ /// may be pushed into it.
+ ///
+ /// A pushed predicate sees the block's physical values. A block whose
footer
+ /// needs the logical timestamp repair against this schema stores values
its
+ /// labels misdescribe, so the predicate would drop rows that match, and
the
+ /// row filter is withheld for that block. With no schema set that cannot
be
+ /// ruled out, so the row filter is withheld from every parquet block.
+ pub fn with_repair_table_schema(
+ mut self,
+ repair_table_schema: Option<arrow_schema::SchemaRef>,
+ ) -> Self {
+ self.repair_table_schema = repair_table_schema;
+ self
+ }
+
+ /// The schema installed by [`Self::with_repair_table_schema`]. Test-only,
+ /// because the decision to install one is made a layer up, in the log
record
+ /// reader, and that is what its test asserts.
+ #[cfg(test)]
+ pub(crate) fn repair_table_schema(&self) ->
Option<&arrow_schema::SchemaRef> {
+ self.repair_table_schema.as_ref()
+ }
+
/// Resolve Avro blocks up to this schema as they are read.
///
/// A block records the schema it was written with, which may predate a
@@ -223,6 +250,7 @@ impl Decoder {
batch_size: 1024,
hudi_configs,
row_filter: None,
+ repair_table_schema: None,
reader_schema_json: None,
hfile_as_records: false,
key_predicate: None,
@@ -502,10 +530,31 @@ impl Decoder {
// Resolved here rather than by the caller because the predicate has to
// be matched against this block's own schema, which does not exist
until
// its footer is read. A builder that declines reads every row.
- let row_filter = self
- .row_filter
- .as_ref()
- .and_then(|build| build(builder.parquet_schema(),
builder.schema().as_ref()));
+ // Withheld when the block needs the logical timestamp repair, as the
base
+ // file read does (see `with_repair_table_schema`). That costs only the
+ // pushdown, the same as a predicate that is not `mor_pk_safe`.
+ let row_filter = match (&self.row_filter, &self.repair_table_schema) {
+ (Some(build), Some(table))
+ if
!crate::schema::batch_evolution::schema_needs_logical_type_repair(
+ builder.schema(),
+ table,
+ ) =>
+ {
+ build(builder.parquet_schema(), builder.schema().as_ref())
+ }
+ (Some(_), table) => {
+ log::debug!(
+ "parquet log block {}; not pushing its row filter",
+ if table.is_some() {
+ "needs the logical timestamp repair"
+ } else {
+ "has no table schema to check the logical timestamp
repair against"
+ }
+ );
+ None
+ }
+ (None, _) => None,
+ };
if let Some(row_filter) = row_filter {
builder = builder.with_row_filter(row_filter);
}
@@ -656,7 +705,7 @@ mod tests {
use apache_avro::types::Record as AvroRecord;
use apache_avro::types::Value as AvroValue;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch, StringArray};
- use arrow_schema::{DataType, Field, Schema};
+ use arrow_schema::{DataType, Field, Schema, TimeUnit};
use parquet::arrow::ArrowWriter;
use std::io::{BufReader, Cursor};
use std::sync::Arc;
@@ -1363,4 +1412,126 @@ mod tests {
.unwrap_err();
assert!(err.to_string().contains("mixes"), "got: {err}");
}
+
+ /// A parquet log block holding one row: `key` = "k" and `ts` of
`file_type`
+ /// storing `stored`.
+ fn one_ts_row_parquet_block(file_type: DataType, stored: i64) -> Bytes {
+ use arrow_array::TimestampMicrosecondArray;
+ let ts: ArrayRef = match &file_type {
+ DataType::Int64 => Arc::new(Int64Array::from(vec![stored])),
+ DataType::Timestamp(TimeUnit::Microsecond, tz) => Arc::new(
+
TimestampMicrosecondArray::from(vec![stored]).with_timezone_opt(tz.clone()),
+ ),
+ other => panic!("unsupported type {other:?}"),
+ };
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("ts", file_type, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![Arc::new(StringArray::from(vec!["k"])) as ArrayRef, ts],
+ )
+ .unwrap();
+ let mut buf = Vec::new();
+ let mut writer = ArrowWriter::try_new(&mut buf, schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ Bytes::from(buf)
+ }
+
+ /// Keeps rows whose `ts` (column 1), read as the physical i64 the block's
+ /// footer describes, is at least `min`. This is how a pushed predicate
sees
+ /// a column: on the block's values, before any repair relabels them.
+ struct TsAtLeast {
+ projection: parquet::arrow::ProjectionMask,
+ min: i64,
+ }
+
+ impl parquet::arrow::arrow_reader::ArrowPredicate for TsAtLeast {
+ fn projection(&self) -> &parquet::arrow::ProjectionMask {
+ &self.projection
+ }
+ fn evaluate(
+ &mut self,
+ batch: RecordBatch,
+ ) -> std::result::Result<arrow_array::BooleanArray,
arrow_schema::ArrowError> {
+ let raw = arrow_cast::cast(batch.column(0), &DataType::Int64)?;
+ let raw = raw.as_any().downcast_ref::<Int64Array>().unwrap();
+ Ok((0..raw.len())
+ .map(|i| Some(raw.value(i) >= self.min))
+ .collect())
+ }
+ }
+
+ fn ts_at_least(min: i64) -> RowFilterBuilder {
+ Arc::new(move |parquet_schema, _projected| {
+ Some(parquet::arrow::arrow_reader::RowFilter::new(vec![
+ Box::new(TsAtLeast {
+ projection:
parquet::arrow::ProjectionMask::roots(parquet_schema, vec![1]),
+ min,
+ }),
+ ]))
+ })
+ }
+
+ fn table_with_ts(ts_type: DataType) -> arrow_schema::SchemaRef {
+ Arc::new(Schema::new(vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("ts", ts_type, false),
+ ]))
+ }
+
+ /// A predicate pushed into a parquet log block that needs the logical
+ /// timestamp repair is evaluated on mislabelled values and drops a row
that
+ /// matches. The gate must withhold it for that block, and only that block.
+ #[test]
+ fn
parquet_log_block_withholds_the_row_filter_when_the_block_needs_timestamp_repair()
{
+ // 2020-01-01T00:00:00.001Z as millis. The predicate asks for
+ // ts >= 2020-01-01T00:00:00Z in the unit the footer claims.
+ const STORED_MS: i64 = 1_577_836_800_001;
+ let cases = [
+ // Labelled tz-aware micros: the literal is 1_577_836_800_000_000.
+ (
+ DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
+ DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
+ 1_577_836_800_000_000,
+ ),
+ // A plain long the table reads as a local timestamp. Compared as a
+ // bare long against a micros-scaled literal it drops the row too.
+ (
+ DataType::Int64,
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ 1_577_836_800_000_000,
+ ),
+ ];
+ for (file_type, table_type, literal) in cases {
+ let content = one_ts_row_parquet_block(file_type.clone(),
STORED_MS);
+ let decode = |table: Option<arrow_schema::SchemaRef>| {
+ Decoder::new(Arc::new(HudiConfigs::empty()))
+ .with_row_filter(Some(ts_at_least(literal)))
+ .with_repair_table_schema(table)
+ .decode_parquet_record_content(content.as_ref())
+ .unwrap()
+ .num_data_rows()
+ };
+
+ // Pushed, the predicate drops the matching row: what the gate
prevents.
+ assert_eq!(
+ decode(Some(table_with_ts(file_type.clone()))),
+ 0,
+ "{file_type}: a table that agrees with the footer keeps the
filter"
+ );
+ assert_eq!(
+ decode(Some(table_with_ts(table_type.clone()))),
+ 1,
+ "{file_type} -> {table_type}: the filter must be withheld, not
drop the row"
+ );
+ assert_eq!(
+ decode(None),
+ 1,
+ "{file_type}: with no table schema the filter must be withheld"
+ );
+ }
+ }
}
diff --git a/crates/core/src/file_group/reader_v2/buffer/row_extraction.rs
b/crates/core/src/file_group/reader_v2/buffer/row_extraction.rs
index 78d57018..e4cd8e01 100644
--- a/crates/core/src/file_group/reader_v2/buffer/row_extraction.rs
+++ b/crates/core/src/file_group/reader_v2/buffer/row_extraction.rs
@@ -117,11 +117,11 @@ fn reconcile_to_schema(batch: &RecordBatch, schema:
&SchemaRef) -> Result<Record
}
}
-/// Reconcile a RecordBatch to a target schema (name-metadata only).
+/// Reconcile a RecordBatch to a target schema: name metadata, allowlisted
+/// promotions, and the logical timestamp repair.
///
-/// Schema evolution runs upstream, so every batch reaching the
-/// merge is already REQUIRED-SHAPED: it has exactly the target's columns with
the
-/// target's types. The only remaining job of this function is field-NAME
+/// Schema evolution runs upstream for most batches, so a batch reaching the
+/// merge has the target's columns. The main job of this function is field-NAME
/// metadata reconciliation between Avro-derived schemas (from log files) and
/// Parquet-derived schemas (from base files). For example:
/// - List child field: arrow-avro uses "item", Parquet uses "element"
@@ -132,6 +132,10 @@ fn reconcile_to_schema(batch: &RecordBatch, schema:
&SchemaRef) -> Result<Record
/// by rebuilding the column's `ArrayData` under the target `DataType` — the
child
/// `ArrayData` is carried over unchanged.
///
+/// A parquet log block reaches the merge in its file schema, so the rebuild
also
+/// applies the logical timestamp repair the base file path applies (see
+/// [`rebuild_array_data_to_type`]).
+///
/// Anything beyond that — a missing column, or a genuine type mismatch that
the
/// ArrayData rebuild rejects — is an upstream evolution bug and is surfaced
as a
/// loud error rather than silently papered over.
@@ -202,7 +206,8 @@ pub(crate) fn reconcile_batch_to_schema(
/// own metadata-laden type, so validation rejects it. Descending into
/// `child_data` and rewriting each level's `DataType` resolves this.
///
-/// A leaf's type is never rewritten — see the refusal below.
+/// A leaf's type is never rewritten, except for the logical timestamp repair.
+/// See the refusal below.
fn rebuild_array_data_to_type(
data: arrow::array::ArrayData,
target_type: &DataType,
@@ -229,6 +234,16 @@ fn rebuild_array_data_to_type(
// An i64 buffer tagged i32 reads the low word of each value —
// 5000000000 comes back as 705032704. Refuse instead, and let
// `reconcile_batch_to_schema`'s promotion arm do the conversion.
+ //
+ // The exception is Java's logical-type repair: a plain long read as a
+ // timezone-less timestamp, or millis values a writer labelled tz-aware
+ // micros. Java relabels the file schema (`SchemaRepair`) for parquet
log
+ // blocks as it does for base files, and keeps the i64. Re-tagging
does the
+ // same and agrees with `project_batch_to_schema` on a base file.
+ if
crate::schema::batch_evolution::needs_logical_type_repair(data.data_type(),
target_type)
+ {
+ return data.into_builder().data_type(target_type.clone()).build();
+ }
if data.data_type() != target_type {
return Err(arrow_schema::ArrowError::InvalidArgumentError(format!(
"reconcile: leaf type {} cannot be re-tagged as {target_type};
a leaf \
@@ -719,4 +734,288 @@ mod tests {
&[7i64]
);
}
+
+ /// A one-row batch of a record key and a `ts` column of `data_type`
holding
+ /// `value`. `data_type` is `Int64` or a millis/micros/nanos `Timestamp`.
+ fn ts_batch(data_type: DataType, value: i64) -> RecordBatch {
+ use arrow_array::{Int64Array, TimestampMicrosecondArray,
TimestampMillisecondArray};
+ use arrow_schema::TimeUnit;
+
+ let col: ArrayRef = match &data_type {
+ DataType::Int64 => Arc::new(Int64Array::from(vec![value])),
+ DataType::Timestamp(TimeUnit::Microsecond, tz) => {
+
Arc::new(TimestampMicrosecondArray::from(vec![value]).with_timezone_opt(tz.clone()))
+ }
+ DataType::Timestamp(TimeUnit::Millisecond, tz) => {
+
Arc::new(TimestampMillisecondArray::from(vec![value]).with_timezone_opt(tz.clone()))
+ }
+ DataType::Timestamp(TimeUnit::Nanosecond, tz) => Arc::new(
+ arrow_array::TimestampNanosecondArray::from(vec![value])
+ .with_timezone_opt(tz.clone()),
+ ),
+ other => panic!("unsupported type {other:?}"),
+ };
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("_hoodie_record_key", DataType::Utf8, false),
+ Field::new("ts", data_type, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![Arc::new(StringArray::from(vec!["k"])) as ArrayRef, col],
+ )
+ .unwrap()
+ }
+
+ fn ts(unit: arrow_schema::TimeUnit, tz: Option<&str>) -> DataType {
+ DataType::Timestamp(unit, tz.map(Into::into))
+ }
+
+ /// A parquet log block reaches the drain in its file schema. Each pairing
Java
+ /// repairs (`SchemaRepair.needsLogicalTypeRepair`) must be relabelled
with the
+ /// stored i64 kept, matching what the base file path produces for the same
+ /// file schema, rather than failing reconcile.
+ #[test]
+ fn test_records_to_batch_applies_logical_timestamp_repair() {
+ use arrow_schema::TimeUnit::{Microsecond, Millisecond};
+
+ // 2020-01-01T00:00:00.001Z in the target unit.
+ const STORED: i64 = 1_577_836_800_001;
+ let cases = [
+ // Millis values a writer labelled tz-aware micros.
+ (ts(Microsecond, Some("UTC")), ts(Millisecond, Some("UTC"))),
+ // A plain long the table declares as a local timestamp.
+ (DataType::Int64, ts(Millisecond, None)),
+ (DataType::Int64, ts(Microsecond, None)),
+ ];
+ for (file_type, table_type) in cases {
+ let log_batch = Arc::new(ts_batch(file_type.clone(), STORED));
+ let target = ts_batch(table_type.clone(), 0).schema();
+ let records = vec![BufferedRecord::new_batch_ref(
+ "k".into(),
+ log_batch.clone(),
+ 0,
+ None,
+ )];
+ let out = records_to_batch(records,
target.clone()).unwrap_or_else(|e| {
+ panic!("{file_type} -> {table_type} must be repaired, got:
{e}")
+ });
+ assert_eq!(out.schema(), target);
+ assert_eq!(
+ out.column(1).as_ref(),
+ ts_batch(table_type.clone(), STORED).column(1).as_ref(),
+ "{file_type} -> {table_type} must keep the stored i64"
+ );
+
+ let base =
crate::schema::batch_evolution::project_batch_to_schema(&log_batch, &target)
+ .unwrap();
+ assert_eq!(
+ base.column(1).as_ref(),
+ out.column(1).as_ref(),
+ "{file_type} -> {table_type} must match the base file path"
+ );
+ }
+ }
+
+ /// The repair applies inside a struct too, as Java's `SchemaRepair`
recurses
+ /// into groups.
+ #[test]
+ fn test_reconcile_repairs_nested_tz_aware_timestamp_mislabeled_as_micros()
{
+ use arrow_array::{
+ Array, StructArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
+ };
+ use arrow_schema::{Fields, TimeUnit};
+
+ const STORED_MS: i64 = 1_577_836_799_999;
+ let micros = Field::new("ts", ts(TimeUnit::Microsecond, Some("UTC")),
true);
+ let millis = Field::new("ts", ts(TimeUnit::Millisecond, Some("UTC")),
true);
+ let inner = StructArray::new(
+ Fields::from(vec![micros]),
+ vec![
+
Arc::new(TimestampMicrosecondArray::from(vec![STORED_MS]).with_timezone("UTC"))
+ as ArrayRef,
+ ],
+ None,
+ );
+ let src = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![Field::new(
+ "s",
+ inner.data_type().clone(),
+ true,
+ )])),
+ vec![Arc::new(inner) as ArrayRef],
+ )
+ .unwrap();
+ let target: SchemaRef = Arc::new(Schema::new(vec![Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![millis])),
+ true,
+ )]));
+
+ let out = reconcile_batch_to_schema(&src, &target).unwrap();
+ assert_eq!(out.schema(), target);
+ let got = out
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap()
+ .column(0)
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .unwrap()
+ .value(0);
+ assert_eq!(got, STORED_MS);
+ }
+
+ /// Pairings outside Java's repair are still refused rather than
re-tagged: the
+ /// timezone-less micros -> millis pair is a real unit conversion in Java,
the
+ /// reverse direction is not a mislabel, and a plain long is only
relabelled as
+ /// a local timestamp.
+ #[test]
+ fn test_reconcile_does_not_repair_other_timestamp_pairs() {
+ use arrow_schema::TimeUnit::{Microsecond, Millisecond, Nanosecond};
+
+ let cases = [
+ (ts(Microsecond, None), ts(Millisecond, None)),
+ (ts(Millisecond, Some("UTC")), ts(Microsecond, Some("UTC"))),
+ (ts(Microsecond, None), ts(Millisecond, Some("UTC"))),
+ (ts(Microsecond, Some("UTC")), ts(Millisecond, None)),
+ (DataType::Int64, ts(Millisecond, Some("UTC"))),
+ // Java only relabels a plain long as local-timestamp millis or
micros.
+ (DataType::Int64, ts(Nanosecond, None)),
+ ];
+ for (file_type, table_type) in cases {
+ let src = ts_batch(file_type.clone(), 1_000);
+ let target = ts_batch(table_type.clone(), 0).schema();
+ let err = reconcile_batch_to_schema(&src,
&target).expect_err(&format!(
+ "{file_type} -> {table_type} must not be re-tagged"
+ ));
+ assert!(
+ err.to_string().contains("cannot be re-tagged"),
+ "unexpected error: {err}"
+ );
+ }
+ }
+
+ /// The re-tag keeps nulls and a slice's offset: the rebuilt column is the
+ /// same rows of the same buffer, relabelled.
+ #[test]
+ fn test_reconcile_repair_keeps_nulls_and_slice_offset() {
+ use arrow_array::{Int64Array, TimestampMicrosecondArray,
TimestampMillisecondArray};
+ use arrow_schema::TimeUnit;
+
+ let values = [Some(1_i64), None, Some(1_577_836_800_001), Some(-1),
None];
+ let cases: [(ArrayRef, DataType, ArrayRef); 2] = [
+ (
+
Arc::new(TimestampMicrosecondArray::from(values.to_vec()).with_timezone("UTC")),
+ ts(TimeUnit::Millisecond, Some("UTC")),
+
Arc::new(TimestampMillisecondArray::from(values.to_vec()).with_timezone("UTC")),
+ ),
+ (
+ Arc::new(Int64Array::from(values.to_vec())),
+ ts(TimeUnit::Microsecond, None),
+ Arc::new(TimestampMicrosecondArray::from(values.to_vec())),
+ ),
+ ];
+ for (src, table_type, expected) in cases {
+ let src_type = src.data_type().clone();
+ let batch = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![Field::new("ts", src_type.clone(),
true)])),
+ vec![src],
+ )
+ .unwrap()
+ .slice(1, 3);
+ let target: SchemaRef = Arc::new(Schema::new(vec![Field::new(
+ "ts",
+ table_type.clone(),
+ true,
+ )]));
+ let out = reconcile_batch_to_schema(&batch, &target).unwrap();
+ assert_eq!(
+ out.column(0).as_ref(),
+ expected.slice(1, 3).as_ref(),
+ "{src_type} -> {table_type}"
+ );
+ assert_eq!(out.column(0).null_count(), 1);
+ }
+ }
+
+ /// The repair applies to a list element and a map value, together with the
+ /// arrow-avro vs parquet child name differences reconcile already handles.
+ #[test]
+ fn test_reconcile_repairs_list_element_and_map_value_timestamps() {
+ use arrow_array::builder::{
+ Int64Builder, ListBuilder, MapBuilder, MapFieldNames,
StringBuilder,
+ TimestampMicrosecondBuilder,
+ };
+ use arrow_array::{Array, ListArray, MapArray,
TimestampMillisecondArray};
+ use arrow_schema::{Fields, TimeUnit};
+
+ const A: i64 = 1_577_836_800_001;
+ const B: i64 = 1_577_836_799_999;
+
+ // List<ts micros UTC>, element field "item" (arrow-avro spelling).
+ let mut lb =
ListBuilder::new(TimestampMicrosecondBuilder::new().with_timezone("UTC"));
+ lb.values().append_value(A);
+ lb.values().append_null();
+ lb.values().append_value(B);
+ lb.append(true);
+ let list = lb.finish();
+
+ // Map<string, long>, entries named "entries" (arrow-avro spelling).
+ let names = MapFieldNames {
+ entry: "entries".to_string(),
+ key: "key".to_string(),
+ value: "value".to_string(),
+ };
+ let mut mb = MapBuilder::new(Some(names), StringBuilder::new(),
Int64Builder::new());
+ mb.keys().append_value("a");
+ mb.values().append_value(A);
+ mb.keys().append_value("b");
+ mb.values().append_null();
+ mb.append(true).unwrap();
+ let map = mb.finish();
+
+ let src = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![
+ Field::new("l", list.data_type().clone(), true),
+ Field::new("m", map.data_type().clone(), true),
+ ])),
+ vec![Arc::new(list) as ArrayRef, Arc::new(map) as ArrayRef],
+ )
+ .unwrap();
+
+ // Parquet spelling ("element", "key_value") and the table's types.
+ let element = Field::new("element", ts(TimeUnit::Millisecond,
Some("UTC")), true);
+ let entries = Field::new(
+ "key_value",
+ DataType::Struct(Fields::from(vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", ts(TimeUnit::Millisecond, None), true),
+ ])),
+ false,
+ );
+ let target: SchemaRef = Arc::new(Schema::new(vec![
+ Field::new("l", DataType::List(Arc::new(element)), true),
+ Field::new("m", DataType::Map(Arc::new(entries), false), true),
+ ]));
+
+ let out = reconcile_batch_to_schema(&src, &target).unwrap();
+ assert_eq!(out.schema(), target);
+
+ let l = out.column(0).as_any().downcast_ref::<ListArray>().unwrap();
+ let l = l.value(0);
+ let l = l
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .unwrap();
+ assert_eq!(l.iter().collect::<Vec<_>>(), vec![Some(A), None, Some(B)]);
+
+ let m = out.column(1).as_any().downcast_ref::<MapArray>().unwrap();
+ let v = m
+ .values()
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .unwrap();
+ assert_eq!(v.iter().collect::<Vec<_>>(), vec![Some(A), None]);
+ }
}
diff --git a/crates/core/src/file_group/reader_v2/log_record_reader.rs
b/crates/core/src/file_group/reader_v2/log_record_reader.rs
index e398041d..3281b30a 100644
--- a/crates/core/src/file_group/reader_v2/log_record_reader.rs
+++ b/crates/core/src/file_group/reader_v2/log_record_reader.rs
@@ -885,6 +885,11 @@ impl BaseHoodieLogRecordReader {
} else {
None
})
+ // What a parquet block is checked against before that filter is pushed
+ // into it. Without a table schema the filter is withheld from parquet
+ // blocks, as the base file read withholds it when the repair cannot be
+ // ruled out.
+
.with_repair_table_schema(self.reader_context.schema_handler.table_schema.clone())
.with_reader_schema(
self.reader_context
.schema_handler
@@ -2381,4 +2386,83 @@ mod tests {
"the committed instant must still be merged"
);
}
+
+ /// A reader whose context carries `row_filter` under `mor_pk_safe`, for
+ /// asserting what `block_decoder` installs. The gate lives here rather
than
+ /// in `Decoder`, so it is tested here.
+ fn reader_with(
+ mor_pk_safe: bool,
+ row_filter: Option<crate::storage::RowFilterBuilder>,
+ ) -> BaseHoodieLogRecordReader {
+ let mut ctx = ReaderContext::empty();
+ ctx.table_config.insert(
+ crate::config::table::HudiTableConfig::OrderingFields
+ .as_ref()
+ .to_string(),
+ "ts".to_string(),
+ );
+ ctx.rebuild_record_context(String::new());
+ ctx.mor_pk_safe = mor_pk_safe;
+ ctx.row_filter_builder = row_filter;
+ BaseHoodieLogRecordReader {
+ reader_context: Arc::new(ctx),
+ storage: Storage::new_with_base_url(
+ crate::storage::util::parse_uri("file:///tmp").unwrap(),
+ )
+ .unwrap(),
+ log_file_paths: vec![],
+ latest_instant_time: "20260101000000000".to_string(),
+ instant_range: None,
+ force_full_scan: false,
+ record_buffer: make_test_buffer(),
+ allow_inflight_instants: false,
+ completion_gate_inputs: None,
+ valid_block_instants: vec![],
+ total_log_files: 0,
+ total_log_blocks: 0,
+ total_log_records: 0,
+ total_corrupt_blocks: 0,
+ total_rollbacks: 0,
+ progress: 0.0,
+ log_block_read_us: 0,
+ log_block_fetch_us: 0,
+ log_block_decode_us: 0,
+ merge_upsert_us: 0,
+ merge_insert_us: 0,
+ }
+ }
+
+ fn a_row_filter_builder() -> crate::storage::RowFilterBuilder {
+ Arc::new(|_, _| None)
+ }
+
+ /// The decoder checks each parquet block against the table schema before
+ /// pushing the filter. The required schema is not a substitute: with no
table
+ /// schema the decoder gets nothing, and so withholds the filter.
+ #[test]
+ fn log_block_decoder_checks_parquet_blocks_against_the_table_schema() {
+ let schema = |name: &str| {
+ Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+ name,
+ arrow_schema::DataType::Utf8,
+ true,
+ )]))
+ };
+ let with_schemas = |table: Option<arrow_schema::SchemaRef>,
+ required: Option<arrow_schema::SchemaRef>| {
+ let mut reader = reader_with(true, Some(a_row_filter_builder()));
+ let mut ctx = (*reader.reader_context).clone();
+ ctx.schema_handler.table_schema = table;
+ ctx.schema_handler.required_schema = required;
+ reader.reader_context = Arc::new(ctx);
+ reader.block_decoder().repair_table_schema().cloned()
+ };
+
+ assert_eq!(
+ with_schemas(Some(schema("table")), Some(schema("required"))),
+ Some(schema("table"))
+ );
+ assert_eq!(with_schemas(None, Some(schema("required"))), None);
+ assert_eq!(with_schemas(None, None), None);
+ }
}
diff --git a/crates/core/src/schema/batch_evolution.rs
b/crates/core/src/schema/batch_evolution.rs
index aaae693c..0bd8978a 100644
--- a/crates/core/src/schema/batch_evolution.rs
+++ b/crates/core/src/schema/batch_evolution.rs
@@ -114,6 +114,81 @@ fn is_value_reinterpreting(file: &DataType, table:
&DataType) -> bool {
)
}
+/// Java's `needsLogicalTypeRepair` (`AvroSchemaRepair`, and `SchemaRepair` for
+/// parquet) in Arrow terms: a file leaf type that Java relabels as the table's
+/// type, keeping the stored i64.
+///
+/// * A plain long the table declares as a timezone-less timestamp, as a writer
+/// without local-timestamp support stores it.
+/// * A tz-aware micros label on millis values ([`is_value_reinterpreting`]).
+///
+/// [`evolve_array`] reaches the same values for a base file, through
+/// `arrow_cast` and the tz-aware reinterpret arm respectively.
+pub(crate) fn needs_logical_type_repair(file: &DataType, table: &DataType) ->
bool {
+ matches!(
+ (file, table),
+ (
+ DataType::Int64,
+ DataType::Timestamp(TimeUnit::Millisecond | TimeUnit::Microsecond,
None),
+ )
+ ) || is_value_reinterpreting(file, table)
+}
+
+/// [`needs_logical_type_repair`] lifted through the container arms, as
+/// [`pair_is_value_reinterpreting`] lifts [`is_value_reinterpreting`].
+fn pair_needs_logical_type_repair(file: &DataType, table: &DataType) -> bool {
+ if needs_logical_type_repair(file, table) {
+ return true;
+ }
+ match (file, table) {
+ // Each file child against its table child, matched as the top level
+ // matches columns: exact name first, then case-insensitively, with an
+ // ambiguous match counting as needing the repair.
+ (DataType::Struct(ff), DataType::Struct(tf)) => ff.iter().any(|f| {
+ if let Some(t) = tf.iter().find(|t| t.name() == f.name()) {
+ return pair_needs_logical_type_repair(f.data_type(),
t.data_type());
+ }
+ let mut matches = tf
+ .iter()
+ .filter(|t| t.name().eq_ignore_ascii_case(f.name()));
+ match (matches.next(), matches.next()) {
+ (Some(t), None) =>
pair_needs_logical_type_repair(f.data_type(), t.data_type()),
+ (Some(_), Some(_)) => true,
+ (None, _) => false,
+ }
+ }),
+ (DataType::List(f), DataType::List(t))
+ | (DataType::LargeList(f), DataType::LargeList(t))
+ | (DataType::Map(f, _), DataType::Map(t, _)) => {
+ pair_needs_logical_type_repair(f.data_type(), t.data_type())
+ }
+ _ => false,
+ }
+}
+
+/// Whether any column of `file_schema`, at any depth, needs the logical-type
+/// repair to read as `table_schema`.
+///
+/// This decides whether a predicate may be evaluated on the file's physical
+/// values, so a column that matches the table ambiguously counts as needing
it:
+/// over-reporting costs pushdown, under-reporting drops rows.
+pub(crate) fn schema_needs_logical_type_repair(
+ file_schema: &arrow_schema::Schema,
+ table_schema: &arrow_schema::Schema,
+) -> bool {
+ file_schema
+ .fields()
+ .iter()
+ .any(|f| match index_of_ci(table_schema, f.name()) {
+ Ok(Some(idx)) => pair_needs_logical_type_repair(
+ f.data_type(),
+ table_schema.fields()[idx].data_type(),
+ ),
+ Ok(None) => false,
+ Err(_) => true,
+ })
+}
+
/// [`is_value_reinterpreting`] lifted through the container arms, mirroring
the
/// recursion in [`evolve_array`] so a nested affected field is not missed.
/// Container drift returns `false`; the read itself rejects that pairing.
@@ -1736,4 +1811,141 @@ mod tests {
"and so must one inside a list"
);
}
+
+ /// `schema_needs_logical_type_repair` and the container arms of
+ /// `pair_needs_logical_type_repair`, which gate row filter pushdown into a
+ /// parquet log block.
+ mod logical_type_repair_gate {
+ use super::super::schema_needs_logical_type_repair;
+ use arrow_schema::{DataType, Field, Fields, Schema, TimeUnit};
+ use std::sync::Arc;
+
+ fn micros_utc() -> DataType {
+ DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()))
+ }
+ fn millis_utc() -> DataType {
+ DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into()))
+ }
+ fn millis_local() -> DataType {
+ DataType::Timestamp(TimeUnit::Millisecond, None)
+ }
+ fn schema(fields: Vec<Field>) -> Schema {
+ Schema::new(fields)
+ }
+ fn col(name: &str, data_type: DataType) -> Field {
+ Field::new(name, data_type, true)
+ }
+ fn needs(file: Vec<Field>, table: Vec<Field>) -> bool {
+ schema_needs_logical_type_repair(&schema(file), &schema(table))
+ }
+ fn strukt(fields: Vec<Field>) -> DataType {
+ DataType::Struct(Fields::from(fields))
+ }
+
+ #[test]
+ fn top_level_leaf_pairs() {
+ assert!(needs(
+ vec![col("ts", micros_utc())],
+ vec![col("ts", millis_utc())]
+ ));
+ assert!(needs(
+ vec![col("ts", DataType::Int64)],
+ vec![col("ts", millis_local())]
+ ));
+ assert!(!needs(
+ vec![col("ts", micros_utc())],
+ vec![col("ts", micros_utc())]
+ ));
+ assert!(!needs(
+ vec![col("ts", DataType::Timestamp(TimeUnit::Microsecond,
None))],
+ vec![col("ts", millis_local())]
+ ));
+ // A column the table does not have cannot be misread against it.
+ assert!(!needs(
+ vec![col("ts", micros_utc())],
+ vec![col("other", millis_utc())]
+ ));
+ // Matched case-insensitively when there is no exact match.
+ assert!(needs(
+ vec![col("TS", micros_utc())],
+ vec![col("ts", millis_utc())]
+ ));
+ }
+
+ #[test]
+ fn top_level_ambiguous_match_counts_as_needing_repair() {
+ let table = vec![col("ts", DataType::Utf8), col("TS",
DataType::Utf8)];
+ assert!(needs(vec![col("Ts", DataType::Utf8)], table.clone()));
+ // An exact name is not ambiguous.
+ assert!(!needs(vec![col("ts", DataType::Utf8)], table));
+ }
+
+ #[test]
+ fn struct_child() {
+ let file = |name: &str| vec![col("s", strukt(vec![col(name,
micros_utc())]))];
+ let table = vec![col("s", strukt(vec![col("ts", millis_utc())]))];
+ assert!(needs(file("ts"), table.clone()));
+ assert!(needs(file("TS"), table.clone()));
+ assert!(!needs(file("other"), table));
+ assert!(!needs(
+ file("ts"),
+ vec![col("s", strukt(vec![col("ts", micros_utc())]))]
+ ));
+ }
+
+ #[test]
+ fn struct_child_ambiguous_match_counts_as_needing_repair() {
+ let table = vec![col(
+ "s",
+ strukt(vec![col("ts", DataType::Utf8), col("TS",
DataType::Utf8)]),
+ )];
+ let file = |name: &str| vec![col("s", strukt(vec![col(name,
DataType::Utf8)]))];
+ assert!(needs(file("Ts"), table.clone()));
+ // An exact name is preferred, so it is not ambiguous.
+ assert!(!needs(file("ts"), table));
+ }
+
+ #[test]
+ fn list_element() {
+ let list = |t: DataType|
DataType::List(Arc::new(Field::new("element", t, true)));
+ let large = |t: DataType|
DataType::LargeList(Arc::new(Field::new("item", t, true)));
+ assert!(needs(
+ vec![col("l", list(micros_utc()))],
+ vec![col("l", list(millis_utc()))]
+ ));
+ assert!(needs(
+ vec![col("l", large(micros_utc()))],
+ vec![col("l", large(millis_utc()))]
+ ));
+ assert!(!needs(
+ vec![col("l", list(micros_utc()))],
+ vec![col("l", list(micros_utc()))]
+ ));
+ }
+
+ #[test]
+ fn map_value() {
+ let map = |v: DataType| {
+ DataType::Map(
+ Arc::new(Field::new(
+ "key_value",
+ strukt(vec![
+ Field::new("key", DataType::Utf8, false),
+ col("value", v),
+ ]),
+ false,
+ )),
+ false,
+ )
+ };
+ assert!(needs(
+ vec![col("m", map(DataType::Int64))],
+ vec![col("m", map(millis_local()))]
+ ));
+ assert!(!needs(
+ vec![col("m", map(DataType::Int64))],
+ vec![col("m", map(DataType::Int64))]
+ ));
+ }
+ }
}
