Skip to content

Preserve V2 partition bounds after schema promotion #108

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
osipovartem merged 1 commit into from
Oct 15, 2026
Merged

Preserve V2 partition bounds after schema promotion #108

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 47 additions & 0 deletions iceberg-rust-spec/src/spec/manifest.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -707,6 +707,22 @@ pub(crate) fn decode_bound(bytes: &[u8], data_type: &Type) -> Result<Value, Erro
}
}

/// Decodes the bound at its original width when copying an existing manifest-list entry.
pub(crate) fn decode_bound_for_copy(bytes: &[u8], data_type: &Type) -> Result<Value, Error> {
match (data_type, bytes.len()) {
(Type::Primitive(PrimitiveType::Long), 4) => {
Value::try_from_bytes(bytes, &Type::Primitive(PrimitiveType::Int))
}
(Type::Primitive(PrimitiveType::Double), 4) => {
Value::try_from_bytes(bytes, &Type::Primitive(PrimitiveType::Float))
}
(Type::Primitive(PrimitiveType::Timestamp), 4) => {
Value::try_from_bytes(bytes, &Type::Primitive(PrimitiveType::Date))
}
_ => Value::try_from_bytes(bytes, data_type),
}
}

impl From<HashMap<i32, Value>> for AvroMap<ByteBuf> {
fn from(value: HashMap<i32, Value>) -> Self {
AvroMap(HashMap::from_iter(
Expand Down Expand Up @@ -776,6 +792,37 @@ impl DataFile {
DataFileBuilder::default()
}

/// Promotes partition values when moving a file entry into a manifest written with a newer schema.
pub fn promote_partition_to_types(
&mut self,
partition_spec: &PartitionSpec,
target_types: &[Type],
) -> Result<(), Error> {
if partition_spec.fields().len() != target_types.len() {
return Err(Error::InvalidFormat(
"Partition type count mismatch".to_owned(),
));
}
for (field, target_type) in partition_spec.fields().iter().zip(target_types.iter()) {
let Some(value) = self.partition.get_mut(field.name()) else {
return Err(Error::NotFound(format!("Partition field {}", field.name())));
};
let Some(value) = value else {
continue;
};
let source_type = value.datatype();
let decimal_unchanged = matches!(
(&*value, target_type),
(Value::Decimal(decimal), Type::Primitive(PrimitiveType::Decimal { scale, .. }))
if decimal_scale(decimal) == *scale
);
if source_type != *target_type && !decimal_unchanged {
*value = value.clone().promote_iceberg(&source_type, target_type)?;
}
}
Ok(())
}

/// Promote numeric bounds to the current schema before rewriting this file
/// into a manifest whose embedded schema has evolved.
pub fn promote_bounds_to_schema(&mut self, schema: &Schema) {
Expand Down
47 changes: 33 additions & 14 deletions iceberg-rust-spec/src/spec/manifest_list.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,12 +394,13 @@ impl ManifestListEntry {
schema: &Schema,
) -> Result<ManifestListEntry, Error> {
let types = partition_field_types(table_metadata, entry.partition_spec_id, schema)?;
Self::try_from_v3_with_types(entry, &types)
Self::try_from_v3_with_types(entry, &types, true)
}

fn try_from_v3_with_types(
entry: _serde::ManifestListEntryV3,
types: &[Option<Type>],
promote: bool,
) -> Result<ManifestListEntry, Error> {
Ok(ManifestListEntry {
format_version: FormatVersion::V3,
Expand All @@ -416,7 +417,7 @@ impl ManifestListEntry {
added_rows_count: Some(entry.added_rows_count),
existing_rows_count: Some(entry.existing_rows_count),
deleted_rows_count: Some(entry.deleted_rows_count),
partitions: decode_partition_summaries(entry.partitions, types)?,
partitions: decode_partition_summaries(entry.partitions, types, promote)?,
key_metadata: entry.key_metadata,
first_row_id: entry.first_row_id,
})
Expand All @@ -440,12 +441,13 @@ impl ManifestListEntry {
schema: &Schema,
) -> Result<ManifestListEntry, Error> {
let types = partition_field_types(table_metadata, entry.partition_spec_id, schema)?;
Self::try_from_v2_with_types(entry, &types)
Self::try_from_v2_with_types(entry, &types, true)
}

fn try_from_v2_with_types(
entry: _serde::ManifestListEntryV2,
types: &[Option<Type>],
promote: bool,
) -> Result<ManifestListEntry, Error> {
Ok(ManifestListEntry {
format_version: FormatVersion::V2,
Expand All @@ -462,7 +464,7 @@ impl ManifestListEntry {
added_rows_count: Some(entry.added_rows_count),
existing_rows_count: Some(entry.existing_rows_count),
deleted_rows_count: Some(entry.deleted_rows_count),
partitions: decode_partition_summaries(entry.partitions, types)?,
partitions: decode_partition_summaries(entry.partitions, types, promote)?,
key_metadata: entry.key_metadata,
first_row_id: None,
})
Expand All @@ -486,12 +488,13 @@ impl ManifestListEntry {
schema: &Schema,
) -> Result<ManifestListEntry, Error> {
let types = partition_field_types(table_metadata, entry.partition_spec_id, schema)?;
Self::try_from_v1_with_types(entry, &types)
Self::try_from_v1_with_types(entry, &types, true)
}

fn try_from_v1_with_types(
entry: _serde::ManifestListEntryV1,
types: &[Option<Type>],
promote: bool,
) -> Result<ManifestListEntry, Error> {
Ok(ManifestListEntry {
format_version: FormatVersion::V1,
Expand All @@ -508,7 +511,7 @@ impl ManifestListEntry {
added_rows_count: entry.added_rows_count,
existing_rows_count: entry.existing_rows_count,
deleted_rows_count: entry.deleted_rows_count,
partitions: decode_partition_summaries(entry.partitions, types)?,
partitions: decode_partition_summaries(entry.partitions, types, promote)?,
key_metadata: entry.key_metadata,
first_row_id: None,
})
Expand Down Expand Up @@ -602,29 +605,32 @@ impl<'a> ManifestListEntryDecoder<'a> {
match format_version {
FormatVersion::V1 => {
let entry = apache_avro::from_value::<ManifestListEntryV1>(&entry)?;
let promote = self.schema.is_some();
let (types, has_partition_types) =
self.partition_types(entry.partition_spec_id, entry.added_snapshot_id)?;
let mut decoded = ManifestListEntry::try_from_v1_with_types(entry, types)?;
let mut decoded = ManifestListEntry::try_from_v1_with_types(entry, types, promote)?;
if !has_partition_types {
decoded.partitions = None;
}
Ok(decoded)
}
FormatVersion::V2 => {
let entry = apache_avro::from_value::<ManifestListEntryV2>(&entry)?;
let promote = self.schema.is_some();
let (types, has_partition_types) =
self.partition_types(entry.partition_spec_id, entry.added_snapshot_id)?;
let mut decoded = ManifestListEntry::try_from_v2_with_types(entry, types)?;
let mut decoded = ManifestListEntry::try_from_v2_with_types(entry, types, promote)?;
if !has_partition_types {
decoded.partitions = None;
}
Ok(decoded)
}
FormatVersion::V3 => {
let entry = apache_avro::from_value::<ManifestListEntryV3>(&entry)?;
let promote = self.schema.is_some();
let (types, has_partition_types) =
self.partition_types(entry.partition_spec_id, entry.added_snapshot_id)?;
let mut decoded = ManifestListEntry::try_from_v3_with_types(entry, types)?;
let mut decoded = ManifestListEntry::try_from_v3_with_types(entry, types, promote)?;
if !has_partition_types {
decoded.partitions = None;
}
Expand All @@ -634,12 +640,11 @@ impl<'a> ManifestListEntryDecoder<'a> {
}
}

/// Decodes an entry's partition field summaries, resolving types from the entry's own
/// `partition_spec_id` against the scan schema and decoding each bound with the spec's
/// width-inference rule ([`super::manifest::decode_bound`]).
/// Decodes partition summaries against the selected schema, promoting bounds only for scans.
fn decode_partition_summaries(
partitions: Option<Vec<FieldSummarySerde>>,
types: &[Option<Type>],
promote: bool,
) -> Result<Option<Vec<FieldSummary>>, Error> {
let Some(summaries) = partitions else {
return Ok(None);
Expand All @@ -650,18 +655,32 @@ fn decode_partition_summaries(
summaries
.into_iter()
.zip(types.iter())
.map(|(summary, data_type)| FieldSummary::from_serde(summary, data_type.as_ref()))
.map(|(summary, data_type)| {
FieldSummary::from_serde_with_mode(summary, data_type.as_ref(), promote)
})
.collect::<Result<Vec<_>, Error>>()
.map(Some)
}

impl FieldSummary {
#[cfg(test)]
fn from_serde(value: FieldSummarySerde, data_type: Option<&Type>) -> Result<Self, Error> {
Self::from_serde_with_mode(value, data_type, true)
}

fn from_serde_with_mode(
value: FieldSummarySerde,
data_type: Option<&Type>,
promote: bool,
) -> Result<Self, Error> {
let decode = |bytes: Option<ByteBuf>| -> Result<Option<Value>, Error> {
match (bytes, data_type) {
(Some(bytes), Some(data_type)) => {
(Some(bytes), Some(data_type)) if promote => {
super::manifest::decode_bound(&bytes, data_type).map(Some)
}
(Some(bytes), Some(data_type)) => {
super::manifest::decode_bound_for_copy(&bytes, data_type).map(Some)
}
_ => Ok(None),
}
};
Expand Down
31 changes: 31 additions & 0 deletions iceberg-rust/src/table/manifest.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -488,6 +488,11 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
schema: &'schema AvroSchema,
table_metadata: &'metadata TableMetadata,
) -> Result<(Self, FilteredManifestStats), Error> {
if manifest.partition_spec_id != table_metadata.default_spec_id {
return Err(Error::NotSupported(
"COW rewrite across partition-spec evolution".to_owned(),
));
}
let inherited_snapshot_id = manifest.added_snapshot_id;
let manifest_reader = ManifestReader::new(bytes)?;
let mut next_inherited_row_id = manifest.first_row_id;
Expand All @@ -496,6 +501,22 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
let mut filtered_stats = FilteredManifestStats::default();
let mut existing_files = 0;
let mut existing_rows = 0;
let current_spec = table_metadata.default_partition_spec()?;
let current_partition_types =
current_spec.data_types(table_metadata.current_schema()?.fields())?;
manifest.partitions = Some(
current_spec
.fields()
.iter()
.map(|_| FieldSummary {
contains_null: false,
contains_nan: None,
lower_bound: None,
upper_bound: None,
})
.collect(),
);
manifest.partition_spec_id = table_metadata.default_spec_id;

writer.add_user_metadata(
"format-version".to_string(),
Expand Down Expand Up @@ -578,6 +599,16 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
existing_files += 1;
existing_rows += entry.data_file().record_count();
*entry.status_mut() = Status::Existing;
entry
.data_file_mut()
.promote_partition_to_types(current_spec, &current_partition_types)?;
update_partitions(
manifest.partitions.as_mut().ok_or_else(|| {
Error::InvalidFormat("Missing partition summaries".to_owned())
})?,
entry.data_file().partition(),
current_spec.fields(),
)?;
writer.append_ser(entry)?;
}
}
Expand Down
Loading
Loading