anoopj commented on code in PR #2986: URL: https://github.com/apache/iceberg-rust/pull/2986#discussion_r3770088367
########## crates/iceberg/src/writer/base_writer/position_delete_writer.rs: ########## @@ -0,0 +1,841 @@ +// 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. + +//! This module provides `PositionDeleteFileWriter`. +//! +//! A position delete file has two required columns: `file_path` (`string`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_PATH`]) and `pos` (`long`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_POS`]). The writer takes batches already shaped as +//! those two columns (see [`position_delete_schema`]) and sets +//! [`DataContentType::PositionDeletes`] on the output. It does not sort its input; see +//! [`PositionDeleteFileWriter::write`]. Review Comment: Could we add a note that position delete files are a V2 construct? Per the spec they are deprecated in v3 and writers must not add new position delete files to v3 tables because v3s add DVs instead.. This base writer correctly has no format-version gate, so gating belongs at the transaction/commit layer. Consider adding something like "position delete files apply to v2 tables; v3 replaces them with deletion vectors, so callers must not route v3 writes here". ########## crates/iceberg/src/writer/base_writer/position_delete_writer.rs: ########## @@ -0,0 +1,841 @@ +// 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. + +//! This module provides `PositionDeleteFileWriter`. +//! +//! A position delete file has two required columns: `file_path` (`string`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_PATH`]) and `pos` (`long`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_POS`]). The writer takes batches already shaped as +//! those two columns (see [`position_delete_schema`]) and sets +//! [`DataContentType::PositionDeletes`] on the output. It does not sort its input; see +//! [`PositionDeleteFileWriter::write`]. + +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_schema::{DataType, Field, SchemaRef as ArrowSchemaRef}; +use once_cell::sync::Lazy; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + +use crate::arrow::schema_to_arrow_schema; +use crate::metadata_columns::{ + RESERVED_FIELD_ID_DELETE_FILE_PATH, RESERVED_FIELD_ID_DELETE_FILE_POS, delete_file_path_field, + delete_file_pos_field, +}; +use crate::spec::{DataContentType, DataFile, PartitionKey, Schema, SchemaRef}; +use crate::writer::file_writer::FileWriterBuilder; +use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator}; +use crate::writer::file_writer::rolling_writer::{RollingFileWriter, RollingFileWriterBuilder}; +use crate::writer::{IcebergWriter, IcebergWriterBuilder}; +use crate::{Error, ErrorKind, Result}; + +/// The canonical Iceberg schema of a position delete file: the required `file_path` +/// (`string`) and `pos` (`long`) columns with their reserved field ids. +static POSITION_DELETE_SCHEMA: Lazy<SchemaRef> = Lazy::new(|| { + Arc::new( + Schema::builder() + .with_fields(vec![ + delete_file_path_field().clone(), + delete_file_pos_field().clone(), + ]) + .build() + .expect("position delete schema is statically valid"), + ) +}); + +/// [`POSITION_DELETE_SCHEMA`] converted to Arrow, keeping the reserved field ids in +/// each field's Parquet field-id metadata. +static POSITION_DELETE_ARROW_SCHEMA: Lazy<ArrowSchemaRef> = Lazy::new(|| { + Arc::new( + schema_to_arrow_schema(&POSITION_DELETE_SCHEMA) + .expect("position delete arrow schema is statically valid"), + ) +}); + +/// Returns the canonical Iceberg schema of a position delete file. +/// +/// Use this to build the [`ParquetWriterBuilder`](crate::writer::file_writer::ParquetWriterBuilder) +/// that backs a [`PositionDeleteFileWriter`], so the written file matches the +/// spec exactly. +pub fn position_delete_schema() -> SchemaRef { + POSITION_DELETE_SCHEMA.clone() +} + +/// Returns the canonical Arrow schema of a position delete file. +pub fn position_delete_arrow_schema() -> ArrowSchemaRef { + POSITION_DELETE_ARROW_SCHEMA.clone() +} + +/// Reads a field's Iceberg field id from its Parquet field-id metadata. +fn field_id(field: &Field) -> Result<i32> { + field + .metadata() + .get(PARQUET_FIELD_ID_META_KEY) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Position delete column `{}` is missing its Iceberg field id metadata.", + field.name() + ), + ) + })? + .parse::<i32>() + .map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Position delete column `{}` has an invalid field id: {e}", + field.name() + ), + ) + }) +} + +/// Validates that a batch is a position delete file: the `file_path` (`Utf8`) and +/// `pos` (`Int64`) columns, in order, with the two reserved field ids. Checking it +/// here gives a clear error before the batch reaches the Parquet writer. +fn validate_position_delete_batch(batch: &RecordBatch) -> Result<()> { + let fields = batch.schema_ref().fields(); + if fields.len() != 2 { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "This writer supports only the two required position delete columns (`file_path`, `pos`); \ + batches with a different column count (e.g. including the optional `row` column) are not supported. Got {} columns.", + fields.len() + ), + )); + } + + let path = &fields[0]; + let path_id = field_id(path)?; + if path_id != RESERVED_FIELD_ID_DELETE_FILE_PATH { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "The first position delete column must be `file_path` (field id {RESERVED_FIELD_ID_DELETE_FILE_PATH}), but got field id {path_id}." + ), + )); + } + // The canonical schema maps Iceberg `string` to `Utf8` and the file writer is + // configured with it, so a `LargeUtf8` column has to be cast to `Utf8` first. + if path.data_type() != &DataType::Utf8 { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "The position delete `file_path` column must be Utf8 (cast it first); got {:?}.", + path.data_type() + ), + )); + } + // Required column: a nullable field could write nulls under a required schema. + if path.is_nullable() { + return Err(Error::new( + ErrorKind::DataInvalid, + "The position delete `file_path` column must be required (non-nullable).", + )); + } + + let pos = &fields[1]; + let pos_id = field_id(pos)?; + if pos_id != RESERVED_FIELD_ID_DELETE_FILE_POS { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "The second position delete column must be `pos` (field id {RESERVED_FIELD_ID_DELETE_FILE_POS}), but got field id {pos_id}." + ), + )); + } + if pos.data_type() != &DataType::Int64 { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "The position delete `pos` column must be Int64, but got {:?}.", + pos.data_type() + ), + )); + } + if pos.is_nullable() { + return Err(Error::new( + ErrorKind::DataInvalid, + "The position delete `pos` column must be required (non-nullable).", + )); + } + + Ok(()) +} + +/// Builder for [`PositionDeleteFileWriter`]. +#[derive(Debug)] +pub struct PositionDeleteFileWriterBuilder< + B: FileWriterBuilder, + L: LocationGenerator, + F: FileNameGenerator, +> { + inner: RollingFileWriterBuilder<B, L, F>, +} + +impl<B, L, F> PositionDeleteFileWriterBuilder<B, L, F> +where + B: FileWriterBuilder, + L: LocationGenerator, + F: FileNameGenerator, +{ + /// Create a new `PositionDeleteFileWriterBuilder` using a `RollingFileWriterBuilder`. + /// + /// The `RollingFileWriterBuilder` must be backed by a file writer configured + /// with the [`position_delete_schema`]; the per-batch validation in + /// [`PositionDeleteFileWriter::write`] guards against a mismatched batch, but + /// the caller is responsible for wiring the same schema into the file writer. + pub fn new(inner: RollingFileWriterBuilder<B, L, F>) -> Self { + Self { inner } + } +} + +#[async_trait::async_trait] +impl<B, L, F> IcebergWriterBuilder for PositionDeleteFileWriterBuilder<B, L, F> +where + B: FileWriterBuilder, + L: LocationGenerator, + F: FileNameGenerator, +{ + type R = PositionDeleteFileWriter<B, L, F>; + + async fn build(&self, partition_key: Option<PartitionKey>) -> Result<Self::R> { + Ok(PositionDeleteFileWriter { + inner: Some(self.inner.build()), + partition_key, + }) + } +} + +/// Writer used to write position delete files within one spec/partition. +#[derive(Debug)] +pub struct PositionDeleteFileWriter< + B: FileWriterBuilder, + L: LocationGenerator, + F: FileNameGenerator, +> { + inner: Option<RollingFileWriter<B, L, F>>, + partition_key: Option<PartitionKey>, +} + +#[async_trait::async_trait] +impl<B, L, F> IcebergWriter for PositionDeleteFileWriter<B, L, F> +where + B: FileWriterBuilder, + L: LocationGenerator, + F: FileNameGenerator, +{ + /// Writes a batch of `(file_path, pos)` records; the shape is validated on every + /// call. + /// + /// The writer does not sort its input. Position delete files must be sorted by + /// `file_path` then `pos`, so the caller must supply rows in that order across all + /// `write` calls; a sorting writer will remove this requirement. + async fn write(&mut self, batch: RecordBatch) -> Result<()> { + // Reject a closed writer before validating the batch. + let Some(writer) = self.inner.as_mut() else { + return Err(Error::new( + ErrorKind::Unexpected, + "Position delete writer is already closed; cannot write.", + )); + }; + validate_position_delete_batch(&batch)?; + writer.write(&self.partition_key, &batch).await + } + + async fn close(&mut self) -> Result<Vec<DataFile>> { + if let Some(writer) = self.inner.take() { + writer + .close() + .await? + .into_iter() + .map(|mut res| { + res.content(DataContentType::PositionDeletes); + // sort_order_id stays null, as the spec requires for position deletes. + if let Some(pk) = self.partition_key.as_ref() { + res.partition(pk.data().clone()); + res.partition_spec_id(pk.spec().spec_id()); + } + res.build().map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Failed to build position delete file: {e}"), + ) + }) + }) + .collect() + } else { + Err(Error::new( + ErrorKind::Unexpected, + "Position delete writer is already closed.", + )) + } + } +} + +#[cfg(test)] +mod test { + use std::collections::HashMap; + use std::sync::Arc; + + use arrow_array::{Int32Array, Int64Array, LargeStringArray, RecordBatch, StringArray}; + use arrow_schema::{DataType, Field}; + use arrow_select::concat::concat_batches; + use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + use parquet::file::properties::WriterProperties; + use tempfile::TempDir; + + use super::*; + use crate::io::FileIO; + use crate::metadata_columns::{ + RESERVED_COL_NAME_DELETE_FILE_PATH, RESERVED_COL_NAME_DELETE_FILE_POS, + }; + use crate::spec::{ + DataFileFormat, Literal, NestedField, PartitionSpec, PrimitiveType, Struct, Transform, Type, + }; + use crate::writer::file_writer::ParquetWriterBuilder; + use crate::writer::file_writer::location_generator::{ + DefaultFileNameGenerator, DefaultLocationGenerator, + }; + use crate::writer::file_writer::rolling_writer::RollingFileWriterBuilder; + + #[test] + fn test_position_delete_schema_shape() { + let schema = position_delete_schema(); + let fields = schema.as_struct().fields(); + assert_eq!(fields.len(), 2); + + assert_eq!(fields[0].id, RESERVED_FIELD_ID_DELETE_FILE_PATH); + assert_eq!(fields[0].name, RESERVED_COL_NAME_DELETE_FILE_PATH); + assert!(fields[0].required); + assert_eq!( + fields[0].field_type.as_ref(), + &Type::Primitive(PrimitiveType::String) + ); + + assert_eq!(fields[1].id, RESERVED_FIELD_ID_DELETE_FILE_POS); + assert_eq!(fields[1].name, RESERVED_COL_NAME_DELETE_FILE_POS); + assert!(fields[1].required); + assert_eq!( + fields[1].field_type.as_ref(), + &Type::Primitive(PrimitiveType::Long) + ); + + // The Arrow projection carries the reserved field ids and non-null flags. + let arrow_schema = position_delete_arrow_schema(); + assert_eq!(arrow_schema.fields().len(), 2); + assert_eq!(arrow_schema.field(0).data_type(), &DataType::Utf8); + assert_eq!(arrow_schema.field(1).data_type(), &DataType::Int64); + assert!(!arrow_schema.field(0).is_nullable()); + assert!(!arrow_schema.field(1).is_nullable()); + } + + fn position_delete_batch(paths: Vec<&str>, positions: Vec<i64>) -> RecordBatch { + RecordBatch::try_new(position_delete_arrow_schema(), vec![ + Arc::new(StringArray::from(paths)), + Arc::new(Int64Array::from(positions)), + ]) + .unwrap() + } + + /// A field carrying an explicit Iceberg field-id metadata entry. + fn field_with_id(name: &str, data_type: DataType, field_id: i32) -> Field { + Field::new(name, data_type, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + field_id.to_string(), + )])) + } + + fn writer_setup( + temp_dir: &TempDir, + ) -> ( + FileIO, + PositionDeleteFileWriterBuilder< + ParquetWriterBuilder, + DefaultLocationGenerator, + DefaultFileNameGenerator, + >, + ) { + let file_io = FileIO::new_with_fs(); + let location_gen = DefaultLocationGenerator::with_data_location( + temp_dir.path().to_str().unwrap().to_string(), + ); + let file_name_gen = + DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet); + + let parquet_writer_builder = ParquetWriterBuilder::new( + WriterProperties::builder().build(), + position_delete_schema(), + ); + let rolling_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + file_io.clone(), + location_gen, + file_name_gen, + ); + ( + file_io, + PositionDeleteFileWriterBuilder::new(rolling_writer_builder), + ) + } + + #[tokio::test] + async fn test_position_delete_writer_round_trip() -> Result<()> { + let temp_dir = TempDir::new().unwrap(); + let (file_io, builder) = writer_setup(&temp_dir); + let mut writer = builder.build(None).await?; + + // Sorted by (file_path, pos): f0/1, f0/4, then f1/2. + let batch = position_delete_batch( + vec![ + "s3://bucket/data/f0.parquet", + "s3://bucket/data/f0.parquet", + "s3://bucket/data/f1.parquet", + ], + vec![1, 4, 2], + ); + writer.write(batch.clone()).await?; + let data_files = writer.close().await?; + + assert_eq!(data_files.len(), 1); + let data_file = &data_files[0]; + assert_eq!(data_file.content_type(), DataContentType::PositionDeletes); + assert_eq!(data_file.file_format, DataFileFormat::Parquet); + assert_eq!(data_file.record_count, 3); + // Unpartitioned writer leaves the default (empty) partition / spec id. + assert_eq!(data_file.partition, Struct::empty()); + assert_eq!(data_file.partition_spec_id, 0); + // The rolling writer fills in file statistics. + assert!(data_file.file_size_in_bytes > 0); + + // The written Parquet file round-trips back to the exact input rows. Review Comment: Can we add a test/assertion that reads the written file's parquet schema and asserts the two column ids? ########## crates/iceberg/src/writer/base_writer/position_delete_writer.rs: ########## @@ -0,0 +1,841 @@ +// 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. + +//! This module provides `PositionDeleteFileWriter`. +//! +//! A position delete file has two required columns: `file_path` (`string`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_PATH`]) and `pos` (`long`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_POS`]). The writer takes batches already shaped as +//! those two columns (see [`position_delete_schema`]) and sets +//! [`DataContentType::PositionDeletes`] on the output. It does not sort its input; see +//! [`PositionDeleteFileWriter::write`]. + +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_schema::{DataType, Field, SchemaRef as ArrowSchemaRef}; +use once_cell::sync::Lazy; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + +use crate::arrow::schema_to_arrow_schema; +use crate::metadata_columns::{ + RESERVED_FIELD_ID_DELETE_FILE_PATH, RESERVED_FIELD_ID_DELETE_FILE_POS, delete_file_path_field, + delete_file_pos_field, +}; +use crate::spec::{DataContentType, DataFile, PartitionKey, Schema, SchemaRef}; +use crate::writer::file_writer::FileWriterBuilder; +use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator}; +use crate::writer::file_writer::rolling_writer::{RollingFileWriter, RollingFileWriterBuilder}; +use crate::writer::{IcebergWriter, IcebergWriterBuilder}; +use crate::{Error, ErrorKind, Result}; + +/// The canonical Iceberg schema of a position delete file: the required `file_path` +/// (`string`) and `pos` (`long`) columns with their reserved field ids. +static POSITION_DELETE_SCHEMA: Lazy<SchemaRef> = Lazy::new(|| { + Arc::new( + Schema::builder() + .with_fields(vec![ + delete_file_path_field().clone(), + delete_file_pos_field().clone(), + ]) + .build() + .expect("position delete schema is statically valid"), + ) +}); + +/// [`POSITION_DELETE_SCHEMA`] converted to Arrow, keeping the reserved field ids in +/// each field's Parquet field-id metadata. +static POSITION_DELETE_ARROW_SCHEMA: Lazy<ArrowSchemaRef> = Lazy::new(|| { + Arc::new( + schema_to_arrow_schema(&POSITION_DELETE_SCHEMA) + .expect("position delete arrow schema is statically valid"), + ) +}); + +/// Returns the canonical Iceberg schema of a position delete file. +/// +/// Use this to build the [`ParquetWriterBuilder`](crate::writer::file_writer::ParquetWriterBuilder) +/// that backs a [`PositionDeleteFileWriter`], so the written file matches the +/// spec exactly. +pub fn position_delete_schema() -> SchemaRef { + POSITION_DELETE_SCHEMA.clone() +} + +/// Returns the canonical Arrow schema of a position delete file. +pub fn position_delete_arrow_schema() -> ArrowSchemaRef { Review Comment: Can this be `pub crate` instead? `position_delete_schema` needs to be public though ########## crates/iceberg/src/writer/base_writer/position_delete_writer.rs: ########## @@ -0,0 +1,841 @@ +// 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. + +//! This module provides `PositionDeleteFileWriter`. +//! +//! A position delete file has two required columns: `file_path` (`string`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_PATH`]) and `pos` (`long`, field id +//! [`RESERVED_FIELD_ID_DELETE_FILE_POS`]). The writer takes batches already shaped as +//! those two columns (see [`position_delete_schema`]) and sets +//! [`DataContentType::PositionDeletes`] on the output. It does not sort its input; see +//! [`PositionDeleteFileWriter::write`]. + +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_schema::{DataType, Field, SchemaRef as ArrowSchemaRef}; +use once_cell::sync::Lazy; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + +use crate::arrow::schema_to_arrow_schema; +use crate::metadata_columns::{ + RESERVED_FIELD_ID_DELETE_FILE_PATH, RESERVED_FIELD_ID_DELETE_FILE_POS, delete_file_path_field, + delete_file_pos_field, +}; +use crate::spec::{DataContentType, DataFile, PartitionKey, Schema, SchemaRef}; +use crate::writer::file_writer::FileWriterBuilder; +use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator}; +use crate::writer::file_writer::rolling_writer::{RollingFileWriter, RollingFileWriterBuilder}; +use crate::writer::{IcebergWriter, IcebergWriterBuilder}; +use crate::{Error, ErrorKind, Result}; + +/// The canonical Iceberg schema of a position delete file: the required `file_path` +/// (`string`) and `pos` (`long`) columns with their reserved field ids. +static POSITION_DELETE_SCHEMA: Lazy<SchemaRef> = Lazy::new(|| { + Arc::new( + Schema::builder() + .with_fields(vec![ + delete_file_path_field().clone(), + delete_file_pos_field().clone(), + ]) + .build() + .expect("position delete schema is statically valid"), + ) +}); + +/// [`POSITION_DELETE_SCHEMA`] converted to Arrow, keeping the reserved field ids in +/// each field's Parquet field-id metadata. +static POSITION_DELETE_ARROW_SCHEMA: Lazy<ArrowSchemaRef> = Lazy::new(|| { + Arc::new( + schema_to_arrow_schema(&POSITION_DELETE_SCHEMA) + .expect("position delete arrow schema is statically valid"), + ) +}); + +/// Returns the canonical Iceberg schema of a position delete file. +/// +/// Use this to build the [`ParquetWriterBuilder`](crate::writer::file_writer::ParquetWriterBuilder) +/// that backs a [`PositionDeleteFileWriter`], so the written file matches the +/// spec exactly. +pub fn position_delete_schema() -> SchemaRef { + POSITION_DELETE_SCHEMA.clone() +} + +/// Returns the canonical Arrow schema of a position delete file. +pub fn position_delete_arrow_schema() -> ArrowSchemaRef { + POSITION_DELETE_ARROW_SCHEMA.clone() +} + +/// Reads a field's Iceberg field id from its Parquet field-id metadata. +fn field_id(field: &Field) -> Result<i32> { + field + .metadata() + .get(PARQUET_FIELD_ID_META_KEY) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Position delete column `{}` is missing its Iceberg field id metadata.", + field.name() + ), + ) + })? + .parse::<i32>() + .map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Position delete column `{}` has an invalid field id: {e}", Review Comment: Consider adding test coverage for this? -- 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]
