blackmwk commented on code in PR #3091:
URL: https://github.com/apache/iceberg-rust/pull/3091#discussion_r3901399586
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -282,3 +243,140 @@ pub struct FileScanTaskDeleteFile {
#[builder(default)]
pub key_metadata: Option<Box<[u8]>>,
}
+
+mod _serde {
+ use serde_derive::{Deserialize as DeserializeDerive, Serialize as
SerializeDerive};
+
+ use super::*;
Review Comment:
Replaced the glob import with explicit named imports and now use the plain
Serialize and Deserialize derives.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -282,3 +243,140 @@ pub struct FileScanTaskDeleteFile {
#[builder(default)]
pub key_metadata: Option<Box<[u8]>>,
}
+
+mod _serde {
+ use serde_derive::{Deserialize as DeserializeDerive, Serialize as
SerializeDerive};
+
+ use super::*;
+ use crate::{Error, ErrorKind};
+
+ #[derive(SerializeDerive, DeserializeDerive)]
+ pub(super) struct FileScanTaskSerde {
+ file_size_in_bytes: u64,
+ start: u64,
+ length: u64,
+ record_count: Option<u64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ first_row_id: Option<i64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ data_sequence_number: Option<i64>,
+ data_file_path: String,
+ data_file_format: DataFileFormat,
+ schema: SchemaRef,
+ project_field_ids: Vec<i32>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ predicate: Option<BoundPredicate>,
+ deletes: Vec<FileScanTaskDeleteFile>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ partition: Option<RawLiteral>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ partition_spec: Option<Arc<PartitionSpec>>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ name_mapping: Option<Arc<NameMapping>>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ unified_partition_type: Option<Arc<StructType>>,
+ case_sensitive: bool,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ key_metadata: Option<Box<[u8]>>,
+ }
+
+ impl TryFrom<FileScanTask> for FileScanTaskSerde {
+ type Error = Error;
+
+ fn try_from(value: FileScanTask) -> Result<Self> {
+ let partition = value
+ .partition
+ .map(|partition| {
+ let partition_spec = value
+ .partition_spec
+ .clone()
+ .unwrap_or_else(||
Arc::new(PartitionSpec::unpartition_spec()));
+ let partition_type =
+
Type::Struct(partition_spec.partition_type(&value.schema)?);
Review Comment:
The serde path now detects partition-spec source IDs missing from the
current schema and omits the advisory partition data instead of failing
serialization. Added a dropped-source-column regression test that verifies the
remaining task fields round-trip.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -282,3 +243,140 @@ pub struct FileScanTaskDeleteFile {
#[builder(default)]
pub key_metadata: Option<Box<[u8]>>,
}
+
+mod _serde {
+ use serde_derive::{Deserialize as DeserializeDerive, Serialize as
SerializeDerive};
+
+ use super::*;
+ use crate::{Error, ErrorKind};
+
+ #[derive(SerializeDerive, DeserializeDerive)]
+ pub(super) struct FileScanTaskSerde {
+ file_size_in_bytes: u64,
+ start: u64,
+ length: u64,
+ record_count: Option<u64>,
Review Comment:
Added skip_serializing_if = "Option::is_none" for record_count on both serde
mirrors, with a test asserting that an absent record count is omitted.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -282,3 +243,140 @@ pub struct FileScanTaskDeleteFile {
#[builder(default)]
pub key_metadata: Option<Box<[u8]>>,
}
+
+mod _serde {
+ use serde_derive::{Deserialize as DeserializeDerive, Serialize as
SerializeDerive};
+
+ use super::*;
+ use crate::{Error, ErrorKind};
+
+ #[derive(SerializeDerive, DeserializeDerive)]
+ pub(super) struct FileScanTaskSerde {
+ file_size_in_bytes: u64,
+ start: u64,
+ length: u64,
+ record_count: Option<u64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ first_row_id: Option<i64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ data_sequence_number: Option<i64>,
+ data_file_path: String,
+ data_file_format: DataFileFormat,
+ schema: SchemaRef,
+ project_field_ids: Vec<i32>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ predicate: Option<BoundPredicate>,
+ deletes: Vec<FileScanTaskDeleteFile>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ partition: Option<RawLiteral>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ partition_spec: Option<Arc<PartitionSpec>>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ name_mapping: Option<Arc<NameMapping>>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ unified_partition_type: Option<Arc<StructType>>,
+ case_sensitive: bool,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ key_metadata: Option<Box<[u8]>>,
+ }
+
+ impl TryFrom<FileScanTask> for FileScanTaskSerde {
+ type Error = Error;
+
+ fn try_from(value: FileScanTask) -> Result<Self> {
+ let partition = value
+ .partition
+ .map(|partition| {
+ let partition_spec = value
+ .partition_spec
+ .clone()
+ .unwrap_or_else(||
Arc::new(PartitionSpec::unpartition_spec()));
Review Comment:
Serialization now rejects a non-empty partition without a partition spec
using a descriptive DataInvalid error. The empty unpartitioned case still
round-trips, and the invalid state has dedicated regression coverage.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -282,3 +243,140 @@ pub struct FileScanTaskDeleteFile {
#[builder(default)]
pub key_metadata: Option<Box<[u8]>>,
}
+
+mod _serde {
+ use serde_derive::{Deserialize as DeserializeDerive, Serialize as
SerializeDerive};
+
+ use super::*;
+ use crate::{Error, ErrorKind};
+
+ #[derive(SerializeDerive, DeserializeDerive)]
+ pub(super) struct FileScanTaskSerde {
+ file_size_in_bytes: u64,
+ start: u64,
+ length: u64,
+ record_count: Option<u64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ first_row_id: Option<i64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ data_sequence_number: Option<i64>,
+ data_file_path: String,
+ data_file_format: DataFileFormat,
+ schema: SchemaRef,
+ project_field_ids: Vec<i32>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ predicate: Option<BoundPredicate>,
+ deletes: Vec<FileScanTaskDeleteFile>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ partition: Option<RawLiteral>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ partition_spec: Option<Arc<PartitionSpec>>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ name_mapping: Option<Arc<NameMapping>>,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ unified_partition_type: Option<Arc<StructType>>,
+ case_sensitive: bool,
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ key_metadata: Option<Box<[u8]>>,
+ }
+
+ impl TryFrom<FileScanTask> for FileScanTaskSerde {
+ type Error = Error;
+
+ fn try_from(value: FileScanTask) -> Result<Self> {
+ let partition = value
+ .partition
+ .map(|partition| {
+ let partition_spec = value
+ .partition_spec
+ .clone()
+ .unwrap_or_else(||
Arc::new(PartitionSpec::unpartition_spec()));
+ let partition_type =
+
Type::Struct(partition_spec.partition_type(&value.schema)?);
+ RawLiteral::try_from(Literal::Struct(partition),
&partition_type)
+ })
+ .transpose()?;
+
+ Ok(Self {
+ file_size_in_bytes: value.file_size_in_bytes,
+ start: value.start,
+ length: value.length,
+ record_count: value.record_count,
+ first_row_id: value.first_row_id,
+ data_sequence_number: value.data_sequence_number,
+ data_file_path: value.data_file_path,
+ data_file_format: value.data_file_format,
+ schema: value.schema,
+ project_field_ids: value.project_field_ids,
+ predicate: value.predicate,
+ deletes: value.deletes,
+ partition,
+ partition_spec: value.partition_spec,
+ name_mapping: value.name_mapping,
+ unified_partition_type: value.unified_partition_type,
+ case_sensitive: value.case_sensitive,
+ key_metadata: value.key_metadata,
+ })
+ }
+ }
+
+ impl Serialize for FileScanTask {
+ fn serialize<__S>(&self, __serializer: __S) ->
std::result::Result<__S::Ok, __S::Error>
+ where __S: serde::Serializer {
Review Comment:
Serialization now builds a borrowed FileScanTaskRefSerde via
TryFrom<&FileScanTask>; schema, deletes, predicate, mappings, and key metadata
are borrowed, so the whole task is no longer cloned. I also changed the
hand-written serializer names to S and serializer and updated the public-API
baseline.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -2648,61 +2648,122 @@ pub mod tests {
assert_eq!(string_arr.value(0), "Apache");
}
- #[test]
- fn test_file_scan_task_serialize_deserialize() {
- let test_fn = |task: FileScanTask| {
- let serialized = serde_json::to_string(&task).unwrap();
- let deserialized: FileScanTask =
serde_json::from_str(&serialized).unwrap();
-
- assert_eq!(task.data_file_path, deserialized.data_file_path);
- assert_eq!(task.start, deserialized.start);
- assert_eq!(task.length, deserialized.length);
- assert_eq!(task.project_field_ids, deserialized.project_field_ids);
- assert_eq!(task.predicate, deserialized.predicate);
- assert_eq!(task.schema, deserialized.schema);
- assert_eq!(task.first_row_id, deserialized.first_row_id);
- assert_eq!(task.data_sequence_number,
deserialized.data_sequence_number);
- };
-
- // without predicate
- let schema = Arc::new(
+ fn file_scan_task_test_schema(primitive_type: PrimitiveType) ->
Arc<Schema> {
+ Arc::new(
Schema::builder()
.with_fields(vec![Arc::new(NestedField::required(
1,
"x",
- Type::Primitive(PrimitiveType::Binary),
+ Type::Primitive(primitive_type),
))])
.build()
.unwrap(),
- );
+ )
+ }
+
+ fn assert_file_scan_task_serde_round_trip(task: FileScanTask) {
+ // Regression test for
https://github.com/apache/iceberg-rust/issues/3089.
+ let serialized = serde_json::to_string(&task).unwrap();
+ let deserialized: FileScanTask =
serde_json::from_str(&serialized).unwrap();
+
+ assert_eq!(task, deserialized);
+ }
+
+ #[test]
+ fn test_file_scan_task_serde_without_predicate() {
let task = FileScanTask::builder()
.with_data_file_path("data_file_path".to_string())
.with_file_size_in_bytes(0)
.with_start(0)
.with_length(100)
.with_project_field_ids(vec![1, 2, 3])
- .with_schema(schema.clone())
+ .with_schema(file_scan_task_test_schema(PrimitiveType::Binary))
.with_record_count(Some(100))
.with_first_row_id(Some(1000))
.with_data_sequence_number(Some(5))
.with_data_file_format(DataFileFormat::Parquet)
.with_case_sensitive(false)
.build();
- test_fn(task);
+ assert_file_scan_task_serde_round_trip(task);
+ }
- // with predicate
+ #[test]
+ fn test_file_scan_task_serde_with_predicate() {
let task = FileScanTask::builder()
.with_data_file_path("data_file_path".to_string())
.with_file_size_in_bytes(0)
.with_start(0)
.with_length(100)
.with_project_field_ids(vec![1, 2, 3])
.with_predicate(Some(BoundPredicate::AlwaysTrue))
- .with_schema(schema)
+ .with_schema(file_scan_task_test_schema(PrimitiveType::Binary))
.with_data_file_format(DataFileFormat::Avro)
.with_case_sensitive(false)
.build();
- test_fn(task);
+ assert_file_scan_task_serde_round_trip(task);
+ }
+
+ #[test]
+ fn test_unpartitioned_file_scan_task_serde() {
+ let task = FileScanTask::builder()
+ .with_data_file_path("data_file_path".to_string())
+ .with_file_size_in_bytes(0)
+ .with_start(0)
+ .with_length(100)
+ .with_project_field_ids(vec![1, 2, 3])
+ .with_schema(file_scan_task_test_schema(PrimitiveType::Binary))
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_partition(Some(Struct::empty()))
+ .with_case_sensitive(false)
+ .build();
+ assert_file_scan_task_serde_round_trip(task);
+ }
+
+ #[test]
+ fn test_file_scan_task_serde_with_all_optional_fields() {
+ let schema = file_scan_task_test_schema(PrimitiveType::Long);
+ let partition_spec = Arc::new(
+ PartitionSpec::builder(schema.clone())
+ .add_partition_field("x", "x", Transform::Identity)
+ .unwrap()
+ .build()
+ .unwrap(),
+ );
+ let unified_partition_type =
Arc::new(partition_spec.partition_type(&schema).unwrap());
+ let task = FileScanTask::builder()
+ .with_data_file_path("data_file_path".to_string())
+ .with_file_size_in_bytes(123)
+ .with_start(10)
+ .with_length(100)
+ .with_project_field_ids(vec![1])
+ .with_schema(schema)
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_deletes(vec![
+ FileScanTaskDeleteFile::builder()
+ .with_file_path("delete_file_path".to_string())
+ .with_file_size_in_bytes(23)
+ .with_file_type(DataContentType::EqualityDeletes)
+ .with_partition_spec_id(0)
+ .with_equality_ids(Some(vec![1]))
+
.with_referenced_data_file(Some("data_file_path".to_string()))
+ .with_content_offset(Some(12))
+ .with_content_size_in_bytes(Some(34))
+ .with_record_count(Some(5))
+ .with_key_metadata(Some(vec![4, 5, 6].into_boxed_slice()))
+ .build(),
+ ])
+ .with_partition(Some(Struct::from_iter([Some(Literal::long(42))])))
Review Comment:
Added independent round-trip cases for Date, timestamp_ns, timestamptz_ns,
Decimal, UUID, Fixed, and Bucket(4) partition encodings. I also added focused
RawLiteral JSON regression coverage for both nanosecond timestamp types.
--
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]
