laskoviymishka commented on code in PR #3091:
URL: https://github.com/apache/iceberg-rust/pull/3091#discussion_r3893135266
##########
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:
The round-trip assertion here is the right shape (full `assert_eq!`, not
just "doesn't throw"), but the partition path only ever sees a single Long
identity field across the whole suite. That's the part of this PR most likely
to have encoding bugs, and it's the least exercised.
I'd add a case per wire-encoding category — Date/Timestamp, Decimal (the
16-byte ByteBuf path), UUID/Fixed — and at least one non-identity transform
like `Bucket(4)`. A `timestamp_ns` identity partition in particular would
currently fail the round-trip (see the top-level note), so that one doubles as
a regression test for the missing branch.
Without those, a broken arm in `RawLiteral` for any of these types ships
green. wdyt?
##########
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:
`record_count` is the one optional field here without `skip_serializing_if`,
so it serializes as `"record_count": null` when it's `None` while every other
optional gets omitted. Since `FileScanTaskSerde` is a clean-slate struct, I'd
just add `#[serde(skip_serializing_if = "Option::is_none")]` to match the rest.
##########
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:
The `use super::*` here is what forces the `Serialize as SerializeDerive`
aliasing dance — the glob pulls the traits in, so the derive macros have to be
renamed to avoid the clash. Every other `_serde` module in the crate
(`manifest/_serde.rs`, `manifest_list/_serde.rs`, `values/serde.rs`) uses
explicit named imports and a plain `#[derive(Serialize, Deserialize)]`.
I'd switch to explicit imports here to match — it drops the alias and lines
this module up with the rest.
##########
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:
`partition_type(&value.schema)?` looks up each partition field's source
column in the current schema, so a table that partitioned on a column, evolved
its spec, then dropped that column will hit the "source column not found" error
and fail to serialize the task. That's a legal partition-evolution +
column-drop sequence —
`test_filtered_scan_with_dropped_partition_source_column` in this same file
already sets it up.
The partition data is only advisory for pruning, so I'd catch the
missing-source-column case and fall back to `partition = None` rather than
failing the whole serialize. Worth a comment on the fallback either way.
##########
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:
`self.clone()` here deep-copies the whole task — schema, deletes vec,
predicate, key metadata — on every serialize, purely because
`TryFrom<FileScanTask>` takes ownership. `FileScanTask` is the unit a planner
ships to workers, so this doubles allocation on the hot path. `impl
TryFrom<&FileScanTask> for FileScanTaskSerde` borrowing the `Arc` fields would
let `serialize` call it without the clone.
While we're here, the `__S` / `__serializer` names are proc-macro derive
output — in a hand-written impl I'd just use `S` / `serializer`.
##########
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:
I'm not sure what actually happens when `partition = Some(non_empty)` but
`partition_spec = None` — and that ambiguity is the problem. The fallback here
builds an `unpartition_spec()`, whose `partition_type` is an empty struct, and
then encodes a non-empty partition against a zero-field type.
Depending on how `RawLiteral::try_from` handles the length mismatch, that
either silently drops every partition value (the zip truncates and you
serialize `{}`, so the round-trip "succeeds" with lost data) or it errors out
with a type mismatch. Two reviewers read it both ways, which tells me the
behavior isn't pinned anywhere.
I'd add an explicit guard — a non-empty partition with no spec is a
programming error, so return a descriptive `DataInvalid` rather than falling
through — plus a test that asserts whichever behavior we land on. Right now
`test_unpartitioned_file_scan_task_serde` only uses `Struct::empty()`, so this
path is completely untested.
--
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]
