comphead commented on code in PR #7148: URL: https://github.com/apache/datafusion-comet/pull/7148#discussion_r4354855631
########## native/core/src/parquet/int96_coercion.rs: ########## @@ -0,0 +1,447 @@ +// 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. + +//! INT96 coercion for the leaves DataFusion's `Int96Coercer` does not find. +//! +//! The scan sets `coerce_int96`, so DataFusion's opener rewrites every INT96 column's Arrow type +//! from `Timestamp(Nanosecond)` to `Timestamp(Microsecond, UTC)` before arrow-rs decodes it. A +//! nanosecond count only spans 1677-09-21 to 2262-04-11, so a leaf the rewrite misses decodes to +//! a wrapped value outside that range, with no error. +//! +//! `Int96Coercer` finds a leaf by rebuilding its Parquet column path from the Arrow schema, and it +//! assumes every list is the standard `list` / `element` layout. A list in another layout has a +//! different path, so its leaf is never coerced: the `bag` / `array` form Spark writes under +//! `spark.sql.parquet.writeLegacyFormat=true`, the `bag` / `array_element` form Hive writes, and +//! the two-level `repeated int96` form. Tracked as apache/datafusion#26393 upstream and +//! apache/datafusion-comet#7029 here. +//! +//! The fix pairs each Arrow leaf with its Parquet column by ordinal, which holds for any layout, +//! and hands the opener the coerced types as the footer's `ARROW:schema` hint. arrow-rs applies the +//! hint when it derives the file's schema, so the opener sees the leaves already coerced and its +//! own coercion skips them. The hint is the schema the opener would have derived anyway, with only +//! those leaves' types changed. + +use arrow::datatypes::{DataType, FieldRef, Schema, TimeUnit}; +use datafusion::datasource::physical_plan::parquet::Int96Coercer; +use parquet::arrow::{encode_arrow_schema, parquet_to_arrow_schema, ARROW_SCHEMA_META_KEY}; +use parquet::basic::Type as PhysicalType; +use parquet::errors::{ParquetError, Result as ParquetResult}; +use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaData, ParquetMetaDataBuilder}; +use parquet::schema::types::ColumnDescPtr; +use std::sync::Arc; + +/// The `coerce_int96` resolution of every native scan, in DataFusion's option syntax. +pub(crate) const COERCE_INT96_UNIT: &str = "us"; +/// The `coerce_int96_tz` of every native scan. INT96 encodes a UTC-adjusted instant. +pub(crate) const COERCE_INT96_TZ: &str = "UTC"; + +/// Returns a copy of `metadata` whose `ARROW:schema` hint coerces every INT96 leaf that DataFusion's +/// `Int96Coercer` would leave as `Timestamp(Nanosecond)`, or `None` when it finds them all. Only the +/// file-level key-value list changes: row groups and page indexes are carried over as-is. +/// +/// A file with no repeated INT96 column returns `None` without building a schema, since only a +/// column inside a list can have a path the coercer gets wrong. The result already holds the +/// coerced types, so passing it back returns `None`. +/// +/// The parquet API cannot carry a file decryptor across this rebuild, so callers must not pass the +/// metadata of an open that supplied decryption properties. +pub(crate) fn coerce_missed_int96_leaves( + metadata: &ParquetMetaData, +) -> ParquetResult<Option<ParquetMetaData>> { + let file_metadata = metadata.file_metadata(); + let descriptor = file_metadata.schema_descr(); + let columns = descriptor.columns(); + if !columns + .iter() + .any(|column| column.physical_type() == PhysicalType::INT96 && column.max_rep_level() > 0) + { + return Ok(None); + } + + // The schema the opener starts from: the footer's own key-value metadata, hint included. + let schema = parquet_to_arrow_schema(descriptor, file_metadata.key_value_metadata())?; + let unit = TimeUnit::Microsecond; + let timezone: Arc<str> = Arc::from(COERCE_INT96_TZ); + // What the opener's coercion does on its own; this pass takes over only what it leaves. + let schema = Int96Coercer::new(descriptor, &schema, &unit) + .with_timezone(Some(Arc::clone(&timezone))) + .coerce() + .unwrap_or(schema); + + let mut leaf = 0; + let mut missed = false; + let fields = schema + .fields() + .iter() + .map(|field| coerce_field(field, columns, &mut leaf, &unit, &timezone, &mut missed)) + .collect::<ParquetResult<Vec<_>>>()?; + if leaf != columns.len() { + return Err(ParquetError::General( + "Parquet schema contains more leaves than the Arrow schema".to_string(), + )); + } + if !missed { + return Ok(None); + } + + let hint = Schema::new_with_metadata(fields, schema.metadata().clone()); + let mut key_values = file_metadata + .key_value_metadata() + .into_iter() + .flatten() + .filter(|key_value| key_value.key != ARROW_SCHEMA_META_KEY) + .cloned() + .collect::<Vec<_>>(); + key_values.push(KeyValue { + key: ARROW_SCHEMA_META_KEY.to_string(), + value: Some(encode_arrow_schema(&hint)), + }); + let file_metadata = FileMetaData::new( + file_metadata.version(), + file_metadata.num_rows(), + file_metadata.created_by().map(str::to_owned), + Some(key_values), + file_metadata.schema_descr_ptr(), + file_metadata.column_orders().cloned(), + ); + Ok(Some( + ParquetMetaDataBuilder::new(file_metadata) + .set_row_groups(metadata.row_groups().to_vec()) Review Comment: Each open that needs the hint deep-copies all row groups with `to_vec()` and clones the column and offset index, which are page-level `Vec<Vec<..>>` and loaded eagerly here, and `coerce_field` rebuilds the whole schema. Files with the standard `list/element` layout, Spark's default, also pay a `parquet_to_arrow_schema` and `Int96Coercer` pass on every open to learn there is nothing to do, because the gate at line 68 is any INT96 column with `max_rep_level() > 0`. Could the gate look at the column path so it only fires for an INT96 leaf under a repeated group not named `list`? Could the description also say what the per-open cost is on a wide, many-page file? The hint also copies `schema.metadata()`, which includes the Spark row metadata JSON that the footer already carries. I have not measured any of this. ########## spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala: ########## @@ -532,6 +532,48 @@ abstract class ParquetReadSuite extends CometTestBase { } } + // DataFusion's INT96 coercion finds a leaf by a Parquet path that assumes the standard + // `list` / `element` layout. A leaf in a legacy list layout (`bag` / `array`, as Spark writes + // with `writeLegacyFormat=true`) was never coerced, so a value outside the nanosecond range + // (1677 to 2262) wrapped to a wrong timestamp with no error + // (https://github.com/apache/datafusion-comet/issues/7029). + test("INT96 timestamps outside the nanosecond range inside lists in either layout") { Review Comment: This test is a good fit for a SQL file. A `-- ConfigMatrix: spark.sql.parquet.writeLegacyFormat=true,false` together with `-- Config: spark.sql.parquet.outputTimestampType=INT96` covers the same two layouts, `datetime/trunc_timestamp.sql` already combines `outputTimestampType` with `CREATE TABLE ... USING parquet` and `INSERT`, and comparing with Spark would make it cheap to add rows the test lacks. All four rows are non-null, so there is no NULL array, empty array, NULL element or NULL map value, and no `array<array<timestamp>>`. I have not run such a fixture. ########## native/core/src/parquet/int96_coercion.rs: ########## @@ -0,0 +1,447 @@ +// 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. + +//! INT96 coercion for the leaves DataFusion's `Int96Coercer` does not find. +//! +//! The scan sets `coerce_int96`, so DataFusion's opener rewrites every INT96 column's Arrow type +//! from `Timestamp(Nanosecond)` to `Timestamp(Microsecond, UTC)` before arrow-rs decodes it. A +//! nanosecond count only spans 1677-09-21 to 2262-04-11, so a leaf the rewrite misses decodes to +//! a wrapped value outside that range, with no error. +//! +//! `Int96Coercer` finds a leaf by rebuilding its Parquet column path from the Arrow schema, and it +//! assumes every list is the standard `list` / `element` layout. A list in another layout has a +//! different path, so its leaf is never coerced: the `bag` / `array` form Spark writes under +//! `spark.sql.parquet.writeLegacyFormat=true`, the `bag` / `array_element` form Hive writes, and +//! the two-level `repeated int96` form. Tracked as apache/datafusion#26393 upstream and +//! apache/datafusion-comet#7029 here. +//! +//! The fix pairs each Arrow leaf with its Parquet column by ordinal, which holds for any layout, +//! and hands the opener the coerced types as the footer's `ARROW:schema` hint. arrow-rs applies the +//! hint when it derives the file's schema, so the opener sees the leaves already coerced and its +//! own coercion skips them. The hint is the schema the opener would have derived anyway, with only +//! those leaves' types changed. + +use arrow::datatypes::{DataType, FieldRef, Schema, TimeUnit}; +use datafusion::datasource::physical_plan::parquet::Int96Coercer; +use parquet::arrow::{encode_arrow_schema, parquet_to_arrow_schema, ARROW_SCHEMA_META_KEY}; +use parquet::basic::Type as PhysicalType; +use parquet::errors::{ParquetError, Result as ParquetResult}; +use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaData, ParquetMetaDataBuilder}; +use parquet::schema::types::ColumnDescPtr; +use std::sync::Arc; + +/// The `coerce_int96` resolution of every native scan, in DataFusion's option syntax. +pub(crate) const COERCE_INT96_UNIT: &str = "us"; +/// The `coerce_int96_tz` of every native scan. INT96 encodes a UTC-adjusted instant. +pub(crate) const COERCE_INT96_TZ: &str = "UTC"; + +/// Returns a copy of `metadata` whose `ARROW:schema` hint coerces every INT96 leaf that DataFusion's +/// `Int96Coercer` would leave as `Timestamp(Nanosecond)`, or `None` when it finds them all. Only the +/// file-level key-value list changes: row groups and page indexes are carried over as-is. +/// +/// A file with no repeated INT96 column returns `None` without building a schema, since only a +/// column inside a list can have a path the coercer gets wrong. The result already holds the +/// coerced types, so passing it back returns `None`. +/// +/// The parquet API cannot carry a file decryptor across this rebuild, so callers must not pass the +/// metadata of an open that supplied decryption properties. +pub(crate) fn coerce_missed_int96_leaves( + metadata: &ParquetMetaData, +) -> ParquetResult<Option<ParquetMetaData>> { + let file_metadata = metadata.file_metadata(); + let descriptor = file_metadata.schema_descr(); + let columns = descriptor.columns(); + if !columns + .iter() + .any(|column| column.physical_type() == PhysicalType::INT96 && column.max_rep_level() > 0) + { + return Ok(None); + } + + // The schema the opener starts from: the footer's own key-value metadata, hint included. + let schema = parquet_to_arrow_schema(descriptor, file_metadata.key_value_metadata())?; + let unit = TimeUnit::Microsecond; + let timezone: Arc<str> = Arc::from(COERCE_INT96_TZ); + // What the opener's coercion does on its own; this pass takes over only what it leaves. + let schema = Int96Coercer::new(descriptor, &schema, &unit) + .with_timezone(Some(Arc::clone(&timezone))) + .coerce() + .unwrap_or(schema); + + let mut leaf = 0; + let mut missed = false; + let fields = schema + .fields() + .iter() + .map(|field| coerce_field(field, columns, &mut leaf, &unit, &timezone, &mut missed)) + .collect::<ParquetResult<Vec<_>>>()?; + if leaf != columns.len() { + return Err(ParquetError::General( + "Parquet schema contains more leaves than the Arrow schema".to_string(), + )); + } + if !missed { + return Ok(None); + } + + let hint = Schema::new_with_metadata(fields, schema.metadata().clone()); + let mut key_values = file_metadata + .key_value_metadata() + .into_iter() + .flatten() + .filter(|key_value| key_value.key != ARROW_SCHEMA_META_KEY) + .cloned() + .collect::<Vec<_>>(); + key_values.push(KeyValue { + key: ARROW_SCHEMA_META_KEY.to_string(), + value: Some(encode_arrow_schema(&hint)), + }); + let file_metadata = FileMetaData::new( + file_metadata.version(), + file_metadata.num_rows(), + file_metadata.created_by().map(str::to_owned), + Some(key_values), + file_metadata.schema_descr_ptr(), + file_metadata.column_orders().cloned(), + ); + Ok(Some( + ParquetMetaDataBuilder::new(file_metadata) + .set_row_groups(metadata.row_groups().to_vec()) + .set_column_index(metadata.column_index().cloned()) + .set_offset_index(metadata.offset_index().cloned()) + .build(), + )) +} + +/// `field` with each INT96 leaf under it that is still `Timestamp(Nanosecond)` coerced to +/// `Timestamp(unit, timezone)`, setting `missed` when it coerces one. `leaf` is the ordinal of the +/// next Parquet column and advances over every leaf under `field`, in the depth-first order both +/// schemas list them in. +fn coerce_field( + field: &FieldRef, + columns: &[ColumnDescPtr], + leaf: &mut usize, + unit: &TimeUnit, + timezone: &Arc<str>, + missed: &mut bool, +) -> ParquetResult<FieldRef> { + let mut child = |field: &FieldRef| coerce_field(field, columns, leaf, unit, timezone, missed); + let data_type = match field.data_type() { + DataType::Struct(fields) => DataType::Struct( + fields + .iter() + .map(&mut child) + .collect::<ParquetResult<Vec<_>>>()? + .into(), + ), + DataType::List(item) => DataType::List(child(item)?), + DataType::LargeList(item) => DataType::LargeList(child(item)?), + DataType::FixedSizeList(item, size) => DataType::FixedSizeList(child(item)?, *size), + DataType::ListView(item) => DataType::ListView(child(item)?), + DataType::LargeListView(item) => DataType::LargeListView(child(item)?), + DataType::Map(entries, sorted) => DataType::Map(child(entries)?, *sorted), + data_type => { + let column = columns.get(*leaf).ok_or_else(|| { + ParquetError::General( + "Arrow schema contains more leaves than the Parquet schema".to_string(), + ) + })?; + *leaf += 1; + return Ok( + if column.physical_type() == PhysicalType::INT96 + && *data_type == DataType::Timestamp(TimeUnit::Nanosecond, None) + { + *missed = true; + Arc::new( + field + .as_ref() + .clone() + .with_data_type(DataType::Timestamp(*unit, Some(Arc::clone(timezone)))), + ) + } else { + Arc::clone(field) + }, + ); + } + }; + Ok(Arc::new(field.as_ref().clone().with_data_type(data_type))) +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{Array, AsArray}; + use arrow::datatypes::TimestampMicrosecondType; + use bytes::Bytes; + use parquet::arrow::arrow_reader::{ + ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder, + }; + use parquet::data_type::{Int96, Int96Type}; + use parquet::file::metadata::ParquetMetaDataReader; + use parquet::file::properties::WriterProperties; + use parquet::file::writer::SerializedFileWriter; + use parquet::schema::parser::parse_message_type; + + const SECONDS: [i64; 4] = [ + 253_402_300_799, // 9999-12-31 23:59:59 + -11_676_096_000, // 1600-01-01 00:00:00 + 9_246_182_400, // 2263-01-01 00:00:00 + 1_577_836_800, // 2020-01-01 00:00:00 + ]; + + const STANDARD: &str = + "optional group at (LIST) { repeated group list { optional int96 element; } }"; + const SPARK_LEGACY: &str = + "optional group at (LIST) { repeated group bag { optional int96 array; } }"; + const HIVE: &str = + "optional group at (LIST) { repeated group bag { optional int96 array_element; } }"; + const TWO_LEVEL: &str = "optional group at (LIST) { repeated int96 array; }"; + const BARE_REPEATED: &str = "repeated int96 at;"; + + fn int96(unix_seconds: i64) -> Int96 { + let nanos_of_day = unix_seconds.rem_euclid(86_400) * 1_000_000_000; + let julian_day = unix_seconds.div_euclid(86_400) + 2_440_588; + let mut value = Int96::new(); + value.set_data( + nanos_of_day as u32, + (nanos_of_day >> 32) as u32, + julian_day as u32, + ); + value + } + + /// Writes a file of `message` whose leaf columns hold `SECONDS`, one single-element list (or + /// struct holding one) per row. `max_defs` is each leaf's maximum definition level, in order. + fn write_file(message: &str, max_defs: &[i16], key_values: Vec<KeyValue>) -> Bytes { + let schema = Arc::new(parse_message_type(message).unwrap()); + let properties = WriterProperties::builder() + .set_key_value_metadata(Some(key_values)) + .build(); + let mut buffer = Vec::new(); + let mut writer = + SerializedFileWriter::new(&mut buffer, schema, Arc::new(properties)).unwrap(); + let mut row_group = writer.next_row_group().unwrap(); + let values: Vec<Int96> = SECONDS.iter().map(|s| int96(*s)).collect(); + for max_def in max_defs { + let mut column = row_group.next_column().unwrap().unwrap(); + column + .typed::<Int96Type>() + .write_batch( + &values, + Some(&vec![*max_def; values.len()]), + Some(&vec![0; values.len()]), + ) + .unwrap(); + column.close().unwrap(); + } + assert!(row_group.next_column().unwrap().is_none()); + row_group.close().unwrap(); + writer.close().unwrap(); + Bytes::from(buffer) + } + + fn metadata_of(file: &Bytes) -> ParquetMetaData { + ParquetMetaDataReader::new().parse_and_finish(file).unwrap() + } + + /// The reader the opener builds for `metadata`: the schema arrow-rs derives from the footer, + /// then DataFusion's own coercion on top, as `ParquetOpener` applies it. + fn opener_reader_metadata(metadata: ParquetMetaData) -> ArrowReaderMetadata { Review Comment: `opener_reader_metadata` is a local copy of the `ParquetOpener` sequence, so it can drift from DataFusion, and nothing exercises `get_metadata` itself. The description says the rebuilt footer is never written to the shared cache and that an encrypted open is left as read, but no test covers either. `parquet_exec.rs` already has cold and warm factory tests. A test there could open a legacy-list INT96 file, assert that the returned footer has the hint while the cached entry does not, and assert that an open with decryption properties is returned unchanged. Scanning through `init_datasource_exec` would also let `scan_leaves` and `opener_reader_metadata` go. ########## native/core/src/parquet/int96_coercion.rs: ########## @@ -0,0 +1,447 @@ +// 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. + +//! INT96 coercion for the leaves DataFusion's `Int96Coercer` does not find. +//! +//! The scan sets `coerce_int96`, so DataFusion's opener rewrites every INT96 column's Arrow type +//! from `Timestamp(Nanosecond)` to `Timestamp(Microsecond, UTC)` before arrow-rs decodes it. A +//! nanosecond count only spans 1677-09-21 to 2262-04-11, so a leaf the rewrite misses decodes to +//! a wrapped value outside that range, with no error. +//! +//! `Int96Coercer` finds a leaf by rebuilding its Parquet column path from the Arrow schema, and it +//! assumes every list is the standard `list` / `element` layout. A list in another layout has a +//! different path, so its leaf is never coerced: the `bag` / `array` form Spark writes under +//! `spark.sql.parquet.writeLegacyFormat=true`, the `bag` / `array_element` form Hive writes, and +//! the two-level `repeated int96` form. Tracked as apache/datafusion#26393 upstream and +//! apache/datafusion-comet#7029 here. +//! +//! The fix pairs each Arrow leaf with its Parquet column by ordinal, which holds for any layout, +//! and hands the opener the coerced types as the footer's `ARROW:schema` hint. arrow-rs applies the +//! hint when it derives the file's schema, so the opener sees the leaves already coerced and its +//! own coercion skips them. The hint is the schema the opener would have derived anyway, with only +//! those leaves' types changed. + +use arrow::datatypes::{DataType, FieldRef, Schema, TimeUnit}; +use datafusion::datasource::physical_plan::parquet::Int96Coercer; +use parquet::arrow::{encode_arrow_schema, parquet_to_arrow_schema, ARROW_SCHEMA_META_KEY}; +use parquet::basic::Type as PhysicalType; +use parquet::errors::{ParquetError, Result as ParquetResult}; +use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaData, ParquetMetaDataBuilder}; +use parquet::schema::types::ColumnDescPtr; +use std::sync::Arc; + +/// The `coerce_int96` resolution of every native scan, in DataFusion's option syntax. +pub(crate) const COERCE_INT96_UNIT: &str = "us"; +/// The `coerce_int96_tz` of every native scan. INT96 encodes a UTC-adjusted instant. +pub(crate) const COERCE_INT96_TZ: &str = "UTC"; + +/// Returns a copy of `metadata` whose `ARROW:schema` hint coerces every INT96 leaf that DataFusion's +/// `Int96Coercer` would leave as `Timestamp(Nanosecond)`, or `None` when it finds them all. Only the +/// file-level key-value list changes: row groups and page indexes are carried over as-is. +/// +/// A file with no repeated INT96 column returns `None` without building a schema, since only a +/// column inside a list can have a path the coercer gets wrong. The result already holds the +/// coerced types, so passing it back returns `None`. +/// +/// The parquet API cannot carry a file decryptor across this rebuild, so callers must not pass the +/// metadata of an open that supplied decryption properties. +pub(crate) fn coerce_missed_int96_leaves( + metadata: &ParquetMetaData, +) -> ParquetResult<Option<ParquetMetaData>> { + let file_metadata = metadata.file_metadata(); + let descriptor = file_metadata.schema_descr(); + let columns = descriptor.columns(); + if !columns + .iter() + .any(|column| column.physical_type() == PhysicalType::INT96 && column.max_rep_level() > 0) + { + return Ok(None); + } + + // The schema the opener starts from: the footer's own key-value metadata, hint included. + let schema = parquet_to_arrow_schema(descriptor, file_metadata.key_value_metadata())?; + let unit = TimeUnit::Microsecond; + let timezone: Arc<str> = Arc::from(COERCE_INT96_TZ); + // What the opener's coercion does on its own; this pass takes over only what it leaves. + let schema = Int96Coercer::new(descriptor, &schema, &unit) + .with_timezone(Some(Arc::clone(&timezone))) + .coerce() + .unwrap_or(schema); + + let mut leaf = 0; + let mut missed = false; + let fields = schema + .fields() + .iter() + .map(|field| coerce_field(field, columns, &mut leaf, &unit, &timezone, &mut missed)) + .collect::<ParquetResult<Vec<_>>>()?; + if leaf != columns.len() { + return Err(ParquetError::General( + "Parquet schema contains more leaves than the Arrow schema".to_string(), + )); + } + if !missed { + return Ok(None); + } + + let hint = Schema::new_with_metadata(fields, schema.metadata().clone()); + let mut key_values = file_metadata + .key_value_metadata() + .into_iter() + .flatten() + .filter(|key_value| key_value.key != ARROW_SCHEMA_META_KEY) + .cloned() + .collect::<Vec<_>>(); + key_values.push(KeyValue { + key: ARROW_SCHEMA_META_KEY.to_string(), + value: Some(encode_arrow_schema(&hint)), + }); + let file_metadata = FileMetaData::new( + file_metadata.version(), + file_metadata.num_rows(), + file_metadata.created_by().map(str::to_owned), + Some(key_values), + file_metadata.schema_descr_ptr(), + file_metadata.column_orders().cloned(), + ); + Ok(Some( + ParquetMetaDataBuilder::new(file_metadata) + .set_row_groups(metadata.row_groups().to_vec()) + .set_column_index(metadata.column_index().cloned()) + .set_offset_index(metadata.offset_index().cloned()) + .build(), Review Comment: This is the third copy of rebuilding a `ParquetMetaData` with a new key-value list (next to `stamp_int96_leaves` in `datetime_rebase.rs` and `with_spark_arrow_schema` in `eager_page_index_reader_factory.rs`), and `coerce_field` is the third depth-first Arrow-leaf to Parquet-column walk (next to `mark_repeated_primitives` and `leaf_count` with `leaf_policies`). Could the rebuild be one helper such as `with_key_values(&ParquetMetaData, Vec<KeyValue>) -> ParquetMetaData` used by all three? The leaf walk could be a shared `map_leaves` with a per-leaf closure, or left for a follow-up. The INT96 stamp itself is not reusable here, since only rebase-aware scans enable 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
