iceberg/spec/
snapshot_summary.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19
20use itertools::Itertools;
21
22use super::{DataContentType, DataFile, PartitionSpecRef};
23use crate::error::invalid_data;
24use crate::spec::{ManifestContentType, ManifestFile, Operation, SchemaRef, Summary};
25use crate::{Error, ErrorKind, Result};
26
27const ADDED_DATA_FILES: &str = "added-data-files";
28const ADDED_DELETE_FILES: &str = "added-delete-files";
29const ADDED_EQUALITY_DELETES: &str = "added-equality-deletes";
30const ADDED_FILE_SIZE: &str = "added-files-size";
31const ADDED_POSITION_DELETES: &str = "added-position-deletes";
32const ADDED_POSITION_DELETE_FILES: &str = "added-position-delete-files";
33const ADDED_RECORDS: &str = "added-records";
34const DELETED_DATA_FILES: &str = "deleted-data-files";
35const DELETED_RECORDS: &str = "deleted-records";
36const ADDED_EQUALITY_DELETE_FILES: &str = "added-equality-delete-files";
37const REMOVED_DELETE_FILES: &str = "removed-delete-files";
38const REMOVED_EQUALITY_DELETES: &str = "removed-equality-deletes";
39const REMOVED_EQUALITY_DELETE_FILES: &str = "removed-equality-delete-files";
40const REMOVED_FILE_SIZE: &str = "removed-files-size";
41const REMOVED_POSITION_DELETES: &str = "removed-position-deletes";
42const REMOVED_POSITION_DELETE_FILES: &str = "removed-position-delete-files";
43const TOTAL_EQUALITY_DELETES: &str = "total-equality-deletes";
44const TOTAL_POSITION_DELETES: &str = "total-position-deletes";
45const TOTAL_DATA_FILES: &str = "total-data-files";
46const TOTAL_DELETE_FILES: &str = "total-delete-files";
47const TOTAL_RECORDS: &str = "total-records";
48const TOTAL_FILE_SIZE: &str = "total-files-size";
49const CHANGED_PARTITION_COUNT_PROP: &str = "changed-partition-count";
50const CHANGED_PARTITION_PREFIX: &str = "partitions.";
51const ICEBERG_VERSION_PROP: &str = "iceberg-version";
52
53/// `SnapshotSummaryCollector` collects and aggregates snapshot update metrics.
54/// It gathers metrics about added or removed data files and manifests, and tracks
55/// partition-specific updates.
56#[derive(Default)]
57pub struct SnapshotSummaryCollector {
58    metrics: UpdateMetrics,
59    partition_metrics: HashMap<String, UpdateMetrics>,
60    max_changed_partitions_for_summaries: u64,
61    properties: HashMap<String, String>,
62    trust_partition_metrics: bool,
63}
64
65impl SnapshotSummaryCollector {
66    /// Set properties for snapshot summary
67    pub fn set(&mut self, key: &str, value: &str) {
68        self.properties.insert(key.to_string(), value.to_string());
69    }
70
71    /// Sets the limit for including partition summaries. Summaries are not
72    /// included if the number of partitions is exceeded.
73    pub fn set_partition_summary_limit(&mut self, limit: u64) {
74        self.max_changed_partitions_for_summaries = limit;
75    }
76
77    /// Adds a data file to the summary collector
78    pub fn add_file(
79        &mut self,
80        data_file: &DataFile,
81        schema: SchemaRef,
82        partition_spec: PartitionSpecRef,
83    ) {
84        self.metrics.add_file(data_file);
85        if !data_file.partition.fields().is_empty() {
86            self.update_partition_metrics(schema, partition_spec, data_file, true);
87        }
88    }
89
90    /// Removes a data file from the summary collector
91    pub fn remove_file(
92        &mut self,
93        data_file: &DataFile,
94        schema: SchemaRef,
95        partition_spec: PartitionSpecRef,
96    ) {
97        self.metrics.remove_file(data_file);
98        if !data_file.partition.fields().is_empty() {
99            self.update_partition_metrics(schema, partition_spec, data_file, false);
100        }
101    }
102
103    /// Adds a manifest to the summary collector
104    pub fn add_manifest(&mut self, manifest: &ManifestFile) {
105        self.trust_partition_metrics = false;
106        self.partition_metrics.clear();
107        self.metrics.add_manifest(manifest);
108    }
109
110    /// Updates partition-specific metrics for a data file.
111    pub fn update_partition_metrics(
112        &mut self,
113        schema: SchemaRef,
114        partition_spec: PartitionSpecRef,
115        data_file: &DataFile,
116        is_add_file: bool,
117    ) {
118        let partition_path = partition_spec.partition_to_path(&data_file.partition, schema);
119        let metrics = self.partition_metrics.entry(partition_path).or_default();
120
121        if is_add_file {
122            metrics.add_file(data_file);
123        } else {
124            metrics.remove_file(data_file);
125        }
126    }
127
128    /// Merges another `SnapshotSummaryCollector` into the current one
129    pub fn merge(&mut self, summary: SnapshotSummaryCollector) {
130        self.metrics.merge(&summary.metrics);
131        self.properties.extend(summary.properties);
132
133        if self.trust_partition_metrics && summary.trust_partition_metrics {
134            for (partition, partition_metric) in summary.partition_metrics.iter() {
135                self.partition_metrics
136                    .entry(partition.to_string())
137                    .or_default()
138                    .merge(partition_metric);
139            }
140        } else {
141            self.partition_metrics.clear();
142            self.trust_partition_metrics = false;
143        }
144    }
145
146    /// Builds final map of summaries
147    pub fn build(&self) -> HashMap<String, String> {
148        let mut properties = self.metrics.to_map();
149        let changed_partitions_count = self.partition_metrics.len() as u64;
150        set_if_positive(
151            &mut properties,
152            changed_partitions_count,
153            CHANGED_PARTITION_COUNT_PROP,
154        );
155
156        if changed_partitions_count <= self.max_changed_partitions_for_summaries {
157            for (partition_path, update_metrics_partition) in &self.partition_metrics {
158                let property_key = format!("{CHANGED_PARTITION_PREFIX}{partition_path}");
159                let partition_summary = update_metrics_partition
160                    .to_map()
161                    .into_iter()
162                    .map(|(property, value)| format!("{property}={value}"))
163                    .join(",");
164
165                if !partition_summary.is_empty() {
166                    properties.insert(property_key, partition_summary);
167                }
168            }
169        }
170        properties
171    }
172}
173
174#[derive(Debug, Default)]
175struct UpdateMetrics {
176    added_file_size: u64,
177    removed_file_size: u64,
178    added_data_files: u32,
179    removed_data_files: u32,
180    added_eq_delete_files: u64,
181    removed_eq_delete_files: u64,
182    added_pos_delete_files: u64,
183    removed_pos_delete_files: u64,
184    added_delete_files: u32,
185    removed_delete_files: u32,
186    added_records: u64,
187    deleted_records: u64,
188    added_pos_deletes: u64,
189    removed_pos_deletes: u64,
190    added_eq_deletes: u64,
191    removed_eq_deletes: u64,
192}
193
194impl UpdateMetrics {
195    fn add_file(&mut self, data_file: &DataFile) {
196        self.added_file_size += data_file.file_size_in_bytes;
197        match data_file.content_type() {
198            DataContentType::Data => {
199                self.added_data_files += 1;
200                self.added_records += data_file.record_count;
201            }
202            DataContentType::PositionDeletes => {
203                self.added_delete_files += 1;
204                self.added_pos_delete_files += 1;
205                self.added_pos_deletes += data_file.record_count;
206            }
207            DataContentType::EqualityDeletes => {
208                self.added_delete_files += 1;
209                self.added_eq_delete_files += 1;
210                self.added_eq_deletes += data_file.record_count;
211            }
212        }
213    }
214
215    fn remove_file(&mut self, data_file: &DataFile) {
216        self.removed_file_size += data_file.file_size_in_bytes;
217        match data_file.content_type() {
218            DataContentType::Data => {
219                self.removed_data_files += 1;
220                self.deleted_records += data_file.record_count;
221            }
222            DataContentType::PositionDeletes => {
223                self.removed_delete_files += 1;
224                self.removed_pos_delete_files += 1;
225                self.removed_pos_deletes += data_file.record_count;
226            }
227            DataContentType::EqualityDeletes => {
228                self.removed_delete_files += 1;
229                self.removed_eq_delete_files += 1;
230                self.removed_eq_deletes += data_file.record_count;
231            }
232        }
233    }
234
235    fn add_manifest(&mut self, manifest: &ManifestFile) {
236        match manifest.content {
237            ManifestContentType::Data => {
238                self.added_data_files += manifest.added_files_count.unwrap_or(0);
239                self.added_records += manifest.added_rows_count.unwrap_or(0);
240                self.removed_data_files += manifest.deleted_files_count.unwrap_or(0);
241                self.deleted_records += manifest.deleted_rows_count.unwrap_or(0);
242            }
243            ManifestContentType::Deletes => {
244                self.added_delete_files += manifest.added_files_count.unwrap_or(0);
245                self.removed_delete_files += manifest.deleted_files_count.unwrap_or(0);
246            }
247        }
248    }
249
250    fn to_map(&self) -> HashMap<String, String> {
251        let mut properties = HashMap::new();
252        set_if_positive(&mut properties, self.added_file_size, ADDED_FILE_SIZE);
253        set_if_positive(&mut properties, self.removed_file_size, REMOVED_FILE_SIZE);
254        set_if_positive(&mut properties, self.added_data_files, ADDED_DATA_FILES);
255        set_if_positive(&mut properties, self.removed_data_files, DELETED_DATA_FILES);
256        set_if_positive(
257            &mut properties,
258            self.added_eq_delete_files,
259            ADDED_EQUALITY_DELETE_FILES,
260        );
261        set_if_positive(
262            &mut properties,
263            self.removed_eq_delete_files,
264            REMOVED_EQUALITY_DELETE_FILES,
265        );
266        set_if_positive(
267            &mut properties,
268            self.added_pos_delete_files,
269            ADDED_POSITION_DELETE_FILES,
270        );
271        set_if_positive(
272            &mut properties,
273            self.removed_pos_delete_files,
274            REMOVED_POSITION_DELETE_FILES,
275        );
276        set_if_positive(&mut properties, self.added_delete_files, ADDED_DELETE_FILES);
277        set_if_positive(
278            &mut properties,
279            self.removed_delete_files,
280            REMOVED_DELETE_FILES,
281        );
282        set_if_positive(&mut properties, self.added_records, ADDED_RECORDS);
283        set_if_positive(&mut properties, self.deleted_records, DELETED_RECORDS);
284        set_if_positive(
285            &mut properties,
286            self.added_pos_deletes,
287            ADDED_POSITION_DELETES,
288        );
289        set_if_positive(
290            &mut properties,
291            self.removed_pos_deletes,
292            REMOVED_POSITION_DELETES,
293        );
294        set_if_positive(
295            &mut properties,
296            self.added_eq_deletes,
297            ADDED_EQUALITY_DELETES,
298        );
299        set_if_positive(
300            &mut properties,
301            self.removed_eq_deletes,
302            REMOVED_EQUALITY_DELETES,
303        );
304        properties
305    }
306
307    fn merge(&mut self, other: &UpdateMetrics) {
308        self.added_file_size += other.added_file_size;
309        self.removed_file_size += other.removed_file_size;
310        self.added_data_files += other.added_data_files;
311        self.removed_data_files += other.removed_data_files;
312        self.added_eq_delete_files += other.added_eq_delete_files;
313        self.removed_eq_delete_files += other.removed_eq_delete_files;
314        self.added_pos_delete_files += other.added_pos_delete_files;
315        self.removed_pos_delete_files += other.removed_pos_delete_files;
316        self.added_delete_files += other.added_delete_files;
317        self.removed_delete_files += other.removed_delete_files;
318        self.added_records += other.added_records;
319        self.deleted_records += other.deleted_records;
320        self.added_pos_deletes += other.added_pos_deletes;
321        self.removed_pos_deletes += other.removed_pos_deletes;
322        self.added_eq_deletes += other.added_eq_deletes;
323        self.removed_eq_deletes += other.removed_eq_deletes;
324    }
325}
326
327fn set_if_positive<T>(properties: &mut HashMap<String, String>, value: T, property_name: &str)
328where T: PartialOrd + Default + ToString {
329    if value > T::default() {
330        properties.insert(property_name.to_string(), value.to_string());
331    }
332}
333
334pub(crate) fn set_iceberg_version(properties: &mut HashMap<String, String>) {
335    let version = env!("CARGO_PKG_VERSION");
336    properties.insert(
337        ICEBERG_VERSION_PROP.to_string(),
338        format!("Apache Iceberg Rust {version}"),
339    );
340}
341
342pub(crate) fn update_snapshot_summaries(
343    summary: Summary,
344    previous_summary: Option<&Summary>,
345    truncate_full_table: bool,
346) -> Result<Summary> {
347    // Validate that the operation is supported
348    if summary.operation != Operation::Append
349        && summary.operation != Operation::Overwrite
350        && summary.operation != Operation::Delete
351    {
352        return Err(invalid_data!("Operation is not supported."));
353    }
354
355    let mut summary = match previous_summary {
356        Some(prev_summary) if truncate_full_table && summary.operation == Operation::Overwrite => {
357            truncate_table_summary(summary, prev_summary).map_err(|err| {
358                Error::new(ErrorKind::Unexpected, "Failed to truncate table summary.")
359                    .with_source(err)
360            })?
361        }
362        _ => summary,
363    };
364
365    update_totals(
366        &mut summary,
367        previous_summary,
368        TOTAL_DATA_FILES,
369        ADDED_DATA_FILES,
370        DELETED_DATA_FILES,
371    );
372
373    update_totals(
374        &mut summary,
375        previous_summary,
376        TOTAL_DELETE_FILES,
377        ADDED_DELETE_FILES,
378        REMOVED_DELETE_FILES,
379    );
380
381    update_totals(
382        &mut summary,
383        previous_summary,
384        TOTAL_RECORDS,
385        ADDED_RECORDS,
386        DELETED_RECORDS,
387    );
388
389    update_totals(
390        &mut summary,
391        previous_summary,
392        TOTAL_FILE_SIZE,
393        ADDED_FILE_SIZE,
394        REMOVED_FILE_SIZE,
395    );
396
397    update_totals(
398        &mut summary,
399        previous_summary,
400        TOTAL_POSITION_DELETES,
401        ADDED_POSITION_DELETES,
402        REMOVED_POSITION_DELETES,
403    );
404
405    update_totals(
406        &mut summary,
407        previous_summary,
408        TOTAL_EQUALITY_DELETES,
409        ADDED_EQUALITY_DELETES,
410        REMOVED_EQUALITY_DELETES,
411    );
412    Ok(summary)
413}
414
415fn get_prop(previous_summary: &Summary, prop: &str) -> Result<u64> {
416    let value_str = previous_summary
417        .additional_properties
418        .get(prop)
419        .map(String::as_str)
420        .unwrap_or("0");
421    value_str.parse::<u64>().map_err(|err| {
422        Error::new(
423            ErrorKind::Unexpected,
424            format!("Failed to parse summary property '{prop}' value '{value_str}' as u64."),
425        )
426        .with_source(err)
427    })
428}
429
430fn truncate_table_summary(mut summary: Summary, previous_summary: &Summary) -> Result<Summary> {
431    for prop in [
432        TOTAL_DATA_FILES,
433        TOTAL_DELETE_FILES,
434        TOTAL_RECORDS,
435        TOTAL_FILE_SIZE,
436        TOTAL_POSITION_DELETES,
437        TOTAL_EQUALITY_DELETES,
438    ] {
439        summary
440            .additional_properties
441            .insert(prop.to_string(), "0".to_string());
442    }
443
444    let value = get_prop(previous_summary, TOTAL_DATA_FILES)?;
445    if value != 0 {
446        summary
447            .additional_properties
448            .insert(DELETED_DATA_FILES.to_string(), value.to_string());
449    }
450    let value = get_prop(previous_summary, TOTAL_DELETE_FILES)?;
451    if value != 0 {
452        summary
453            .additional_properties
454            .insert(REMOVED_DELETE_FILES.to_string(), value.to_string());
455    }
456    let value = get_prop(previous_summary, TOTAL_RECORDS)?;
457    if value != 0 {
458        summary
459            .additional_properties
460            .insert(DELETED_RECORDS.to_string(), value.to_string());
461    }
462    let value = get_prop(previous_summary, TOTAL_FILE_SIZE)?;
463    if value != 0 {
464        summary
465            .additional_properties
466            .insert(REMOVED_FILE_SIZE.to_string(), value.to_string());
467    }
468
469    let value = get_prop(previous_summary, TOTAL_POSITION_DELETES)?;
470    if value != 0 {
471        summary
472            .additional_properties
473            .insert(REMOVED_POSITION_DELETES.to_string(), value.to_string());
474    }
475
476    let value = get_prop(previous_summary, TOTAL_EQUALITY_DELETES)?;
477    if value != 0 {
478        summary
479            .additional_properties
480            .insert(REMOVED_EQUALITY_DELETES.to_string(), value.to_string());
481    }
482
483    Ok(summary)
484}
485
486fn update_totals(
487    summary: &mut Summary,
488    previous_summary: Option<&Summary>,
489    total_property: &str,
490    added_property: &str,
491    removed_property: &str,
492) {
493    let previous_total = match previous_summary {
494        None => 0,
495        Some(prev_summary) => match prev_summary.additional_properties.get(total_property) {
496            Some(value_str) => match value_str.parse::<u64>() {
497                Ok(v) => v,
498                Err(parse_err) => {
499                    tracing::warn!(
500                        "Property '{total_property}' could not be parsed in the previous snapshot summary: {parse_err}. \
501                         Skipping total computation.",
502                    );
503                    return;
504                }
505            },
506            None => {
507                tracing::debug!(
508                    "Property '{total_property}' was not set in the previous snapshot summary. \
509                     Skipping total computation."
510                );
511                return;
512            }
513        },
514    };
515
516    // Parse the added/removed deltas, tolerating an unparsable value by skipping
517    // the total entirely rather than panicking. Computed metrics always overwrite
518    // user-supplied summary properties (see `SnapshotProducer::summary`), so a bad
519    // value should only ever come from a previous snapshot's summary; matching
520    // iceberg-java's `updateTotal`, we ignore it instead of failing the commit.
521    let parse_delta = |property: &str| -> Option<u64> {
522        match summary.additional_properties.get(property) {
523            None => Some(0),
524            Some(value) => match value.parse::<u64>() {
525                Ok(v) => Some(v),
526                Err(parse_err) => {
527                    tracing::warn!(
528                        "Property '{property}' could not be parsed when computing '{total_property}': {parse_err}. \
529                         Skipping total computation.",
530                    );
531                    None
532                }
533            },
534        }
535    };
536
537    let (Some(added), Some(removed)) = (parse_delta(added_property), parse_delta(removed_property))
538    else {
539        return;
540    };
541
542    // Like Java, a total that would go negative is omitted rather than written as a false basis.
543    let Some(new_total) = previous_total.saturating_add(added).checked_sub(removed) else {
544        return;
545    };
546    summary
547        .additional_properties
548        .insert(total_property.to_string(), new_total.to_string());
549}
550
551#[cfg(test)]
552mod tests {
553    use std::collections::HashMap;
554    use std::sync::Arc;
555
556    use super::*;
557    use crate::spec::{
558        DataFileFormat, Datum, Literal, NestedField, PartitionSpec, PrimitiveType, Schema, Struct,
559        Transform, Type, UnboundPartitionField,
560    };
561
562    #[test]
563    fn test_update_snapshot_summaries_append() {
564        let prev_props: HashMap<String, String> = [
565            (TOTAL_DATA_FILES.to_string(), "10".to_string()),
566            (TOTAL_DELETE_FILES.to_string(), "5".to_string()),
567            (TOTAL_RECORDS.to_string(), "100".to_string()),
568            (TOTAL_FILE_SIZE.to_string(), "1000".to_string()),
569            (TOTAL_POSITION_DELETES.to_string(), "3".to_string()),
570            (TOTAL_EQUALITY_DELETES.to_string(), "2".to_string()),
571        ]
572        .into_iter()
573        .collect();
574
575        let previous_summary = Summary {
576            operation: Operation::Append,
577            additional_properties: prev_props,
578        };
579
580        let new_props: HashMap<String, String> = [
581            (ADDED_DATA_FILES.to_string(), "4".to_string()),
582            (DELETED_DATA_FILES.to_string(), "1".to_string()),
583            (ADDED_DELETE_FILES.to_string(), "2".to_string()),
584            (REMOVED_DELETE_FILES.to_string(), "1".to_string()),
585            (ADDED_RECORDS.to_string(), "40".to_string()),
586            (DELETED_RECORDS.to_string(), "10".to_string()),
587            (ADDED_FILE_SIZE.to_string(), "400".to_string()),
588            (REMOVED_FILE_SIZE.to_string(), "100".to_string()),
589            (ADDED_POSITION_DELETES.to_string(), "5".to_string()),
590            (REMOVED_POSITION_DELETES.to_string(), "2".to_string()),
591            (ADDED_EQUALITY_DELETES.to_string(), "3".to_string()),
592            (REMOVED_EQUALITY_DELETES.to_string(), "1".to_string()),
593        ]
594        .into_iter()
595        .collect();
596
597        let summary = Summary {
598            operation: Operation::Append,
599            additional_properties: new_props,
600        };
601
602        let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap();
603
604        assert_eq!(
605            updated.additional_properties.get(TOTAL_DATA_FILES).unwrap(),
606            "13"
607        );
608        assert_eq!(
609            updated
610                .additional_properties
611                .get(TOTAL_DELETE_FILES)
612                .unwrap(),
613            "6"
614        );
615        assert_eq!(
616            updated.additional_properties.get(TOTAL_RECORDS).unwrap(),
617            "130"
618        );
619        assert_eq!(
620            updated.additional_properties.get(TOTAL_FILE_SIZE).unwrap(),
621            "1300"
622        );
623        assert_eq!(
624            updated
625                .additional_properties
626                .get(TOTAL_POSITION_DELETES)
627                .unwrap(),
628            "6"
629        );
630        assert_eq!(
631            updated
632                .additional_properties
633                .get(TOTAL_EQUALITY_DELETES)
634                .unwrap(),
635            "4"
636        );
637    }
638
639    #[test]
640    fn test_truncate_table_summary() {
641        let prev_props: HashMap<String, String> = [
642            (TOTAL_DATA_FILES.to_string(), "10".to_string()),
643            (TOTAL_DELETE_FILES.to_string(), "5".to_string()),
644            (TOTAL_RECORDS.to_string(), "100".to_string()),
645            (TOTAL_FILE_SIZE.to_string(), "1000".to_string()),
646            (TOTAL_POSITION_DELETES.to_string(), "3".to_string()),
647            (TOTAL_EQUALITY_DELETES.to_string(), "2".to_string()),
648        ]
649        .into_iter()
650        .collect();
651
652        let previous_summary = Summary {
653            operation: Operation::Overwrite,
654            additional_properties: prev_props,
655        };
656
657        let mut new_props = HashMap::new();
658        new_props.insert("dummy".to_string(), "value".to_string());
659        let summary = Summary {
660            operation: Operation::Overwrite,
661            additional_properties: new_props,
662        };
663
664        let truncated = truncate_table_summary(summary, &previous_summary).unwrap();
665
666        assert_eq!(
667            truncated
668                .additional_properties
669                .get(TOTAL_DATA_FILES)
670                .unwrap(),
671            "0"
672        );
673        assert_eq!(
674            truncated
675                .additional_properties
676                .get(TOTAL_DELETE_FILES)
677                .unwrap(),
678            "0"
679        );
680        assert_eq!(
681            truncated.additional_properties.get(TOTAL_RECORDS).unwrap(),
682            "0"
683        );
684        assert_eq!(
685            truncated
686                .additional_properties
687                .get(TOTAL_FILE_SIZE)
688                .unwrap(),
689            "0"
690        );
691        assert_eq!(
692            truncated
693                .additional_properties
694                .get(TOTAL_POSITION_DELETES)
695                .unwrap(),
696            "0"
697        );
698        assert_eq!(
699            truncated
700                .additional_properties
701                .get(TOTAL_EQUALITY_DELETES)
702                .unwrap(),
703            "0"
704        );
705
706        assert_eq!(
707            truncated
708                .additional_properties
709                .get(DELETED_DATA_FILES)
710                .unwrap(),
711            "10"
712        );
713        assert_eq!(
714            truncated
715                .additional_properties
716                .get(REMOVED_DELETE_FILES)
717                .unwrap(),
718            "5"
719        );
720        assert_eq!(
721            truncated
722                .additional_properties
723                .get(DELETED_RECORDS)
724                .unwrap(),
725            "100"
726        );
727        assert_eq!(
728            truncated
729                .additional_properties
730                .get(REMOVED_FILE_SIZE)
731                .unwrap(),
732            "1000"
733        );
734        assert_eq!(
735            truncated
736                .additional_properties
737                .get(REMOVED_POSITION_DELETES)
738                .unwrap(),
739            "3"
740        );
741        assert_eq!(
742            truncated
743                .additional_properties
744                .get(REMOVED_EQUALITY_DELETES)
745                .unwrap(),
746            "2"
747        );
748    }
749
750    #[test]
751    fn test_update_snapshot_summaries_overwrite_truncate_handles_totals_above_i32_max() {
752        // A table can legitimately accumulate more than i32::MAX rows or files
753        // over its lifetime. Truncating such a table on overwrite must succeed
754        // and surface the previous totals into the deleted-* counters.
755        let big = (i32::MAX as u64 + 1).to_string(); // 2_147_483_648
756        let prev_props: HashMap<String, String> = [
757            (TOTAL_DATA_FILES.to_string(), big.clone()),
758            (TOTAL_DELETE_FILES.to_string(), "0".to_string()),
759            (TOTAL_RECORDS.to_string(), big.clone()),
760            (TOTAL_FILE_SIZE.to_string(), "0".to_string()),
761            (TOTAL_POSITION_DELETES.to_string(), "0".to_string()),
762            (TOTAL_EQUALITY_DELETES.to_string(), "0".to_string()),
763        ]
764        .into_iter()
765        .collect();
766
767        let previous_summary = Summary {
768            operation: Operation::Overwrite,
769            additional_properties: prev_props,
770        };
771
772        let summary = Summary {
773            operation: Operation::Overwrite,
774            additional_properties: HashMap::new(),
775        };
776
777        let updated = update_snapshot_summaries(summary, Some(&previous_summary), true)
778            .expect("overwrite truncation should accept totals above i32::MAX");
779        assert_eq!(
780            updated
781                .additional_properties
782                .get(DELETED_DATA_FILES)
783                .unwrap(),
784            &big
785        );
786        assert_eq!(
787            updated.additional_properties.get(DELETED_RECORDS).unwrap(),
788            &big
789        );
790    }
791
792    #[test]
793    fn test_update_snapshot_summaries_overwrite_truncate_returns_err_on_malformed_total() {
794        // Non-numeric values in the previous summary (corruption, manual edits,
795        // a foreign implementation) must surface as a recoverable Err - not
796        // crash the process.
797        let prev_props: HashMap<String, String> = [
798            (TOTAL_DATA_FILES.to_string(), "not_a_number".to_string()),
799            (TOTAL_DELETE_FILES.to_string(), "0".to_string()),
800            (TOTAL_RECORDS.to_string(), "0".to_string()),
801            (TOTAL_FILE_SIZE.to_string(), "0".to_string()),
802            (TOTAL_POSITION_DELETES.to_string(), "0".to_string()),
803            (TOTAL_EQUALITY_DELETES.to_string(), "0".to_string()),
804        ]
805        .into_iter()
806        .collect();
807
808        let previous_summary = Summary {
809            operation: Operation::Overwrite,
810            additional_properties: prev_props,
811        };
812
813        let summary = Summary {
814            operation: Operation::Overwrite,
815            additional_properties: HashMap::new(),
816        };
817
818        let err = update_snapshot_summaries(summary, Some(&previous_summary), true)
819            .expect_err("malformed previous summary must produce an Err, not a panic");
820        assert!(
821            err.message().contains("truncate table summary"),
822            "expected wrapped 'Failed to truncate table summary' context, got: {}",
823            err.message()
824        );
825    }
826
827    #[test]
828    fn test_snapshot_summary_collector_build() {
829        let schema = Arc::new(
830            Schema::builder()
831                .with_fields(vec![
832                    NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
833                    NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)).into(),
834                ])
835                .build()
836                .unwrap(),
837        );
838
839        let partition_spec = Arc::new(
840            PartitionSpec::builder(schema.clone())
841                .add_unbound_fields(vec![
842                    UnboundPartitionField::builder()
843                        .source_ids(vec![2])
844                        .name("year".to_string())
845                        .transform(Transform::Identity)
846                        .build()
847                        .unwrap(),
848                ])
849                .unwrap()
850                .with_spec_id(1)
851                .build()
852                .unwrap(),
853        );
854
855        let mut collector = SnapshotSummaryCollector::default();
856        collector.set_partition_summary_limit(10);
857
858        let file1 = DataFile {
859            content: DataContentType::Data,
860            file_path: "s3://testbucket/path/to/file1.parquet".to_string(),
861            file_format: DataFileFormat::Parquet,
862            partition: Struct::from_iter(vec![]),
863            record_count: 10,
864            file_size_in_bytes: 100,
865            column_sizes: HashMap::from([(1, 46), (2, 48), (3, 48)]),
866            value_counts: HashMap::from([(1, 10), (2, 10), (3, 10)]),
867            null_value_counts: HashMap::from([(1, 0), (2, 0), (3, 0)]),
868            nan_value_counts: HashMap::new(),
869            lower_bounds: HashMap::from([
870                (1, Datum::long(1)),
871                (2, Datum::string("a")),
872                (3, Datum::string("x")),
873            ]),
874            upper_bounds: HashMap::from([
875                (1, Datum::long(1)),
876                (2, Datum::string("a")),
877                (3, Datum::string("x")),
878            ]),
879            key_metadata: None,
880            split_offsets: Some(vec![4]),
881            equality_ids: None,
882            sort_order_id: Some(0),
883            partition_spec_id: 0,
884            first_row_id: None,
885            referenced_data_file: None,
886            content_offset: None,
887            content_size_in_bytes: None,
888        };
889
890        let file2 = DataFile {
891            content: DataContentType::Data,
892            file_path: "s3://testbucket/path/to/file2.parquet".to_string(),
893            file_format: DataFileFormat::Parquet,
894            partition: Struct::from_iter(vec![Some(Literal::string("2025"))]),
895            record_count: 20,
896            file_size_in_bytes: 200,
897            column_sizes: HashMap::from([(1, 46), (2, 48), (3, 48)]),
898            value_counts: HashMap::from([(1, 20), (2, 20), (3, 20)]),
899            null_value_counts: HashMap::from([(1, 0), (2, 0), (3, 0)]),
900            nan_value_counts: HashMap::new(),
901            lower_bounds: HashMap::from([
902                (1, Datum::long(1)),
903                (2, Datum::string("a")),
904                (3, Datum::string("x")),
905            ]),
906            upper_bounds: HashMap::from([
907                (1, Datum::long(1)),
908                (2, Datum::string("a")),
909                (3, Datum::string("x")),
910            ]),
911            key_metadata: None,
912            split_offsets: Some(vec![4]),
913            equality_ids: None,
914            sort_order_id: Some(0),
915            partition_spec_id: 0,
916            first_row_id: None,
917            referenced_data_file: None,
918            content_offset: None,
919            content_size_in_bytes: None,
920        };
921
922        collector.add_file(&file1, schema.clone(), partition_spec.clone());
923        collector.add_file(&file2, schema.clone(), partition_spec.clone());
924
925        collector.remove_file(&file1, schema.clone(), partition_spec.clone());
926
927        let props = collector.build();
928
929        assert_eq!(props.get(ADDED_FILE_SIZE).unwrap(), "300");
930        assert_eq!(props.get(REMOVED_FILE_SIZE).unwrap(), "100");
931
932        let partition_key = format!("{}{}", CHANGED_PARTITION_PREFIX, "year=2025");
933
934        assert!(props.contains_key(&partition_key));
935
936        let partition_summary = props.get(&partition_key).unwrap();
937        assert!(partition_summary.contains(&format!("{ADDED_FILE_SIZE}=200")));
938        assert!(partition_summary.contains(&format!("{ADDED_DATA_FILES}=1")));
939        assert!(partition_summary.contains(&format!("{ADDED_RECORDS}=20")));
940    }
941
942    #[test]
943    fn test_snapshot_summary_collector_add_manifest() {
944        let mut collector = SnapshotSummaryCollector::default();
945        collector.set_partition_summary_limit(10);
946
947        let manifest = ManifestFile {
948            manifest_path: "file://dummy.manifest".to_string(),
949            manifest_length: 0,
950            partition_spec_id: 0,
951            content: ManifestContentType::Data,
952            sequence_number: 0,
953            min_sequence_number: 0,
954            added_snapshot_id: 0,
955            added_files_count: Some(3),
956            existing_files_count: Some(0),
957            deleted_files_count: Some(1),
958            added_rows_count: Some(100),
959            existing_rows_count: Some(0),
960            deleted_rows_count: Some(50),
961            partitions: Some(Vec::new()),
962            key_metadata: None,
963            first_row_id: None,
964        };
965
966        collector
967            .partition_metrics
968            .insert("dummy".to_string(), UpdateMetrics::default());
969        collector.add_manifest(&manifest);
970
971        let props = collector.build();
972        assert_eq!(props.get(ADDED_DATA_FILES).unwrap(), "3");
973        assert_eq!(props.get(DELETED_DATA_FILES).unwrap(), "1");
974        assert_eq!(props.get(ADDED_RECORDS).unwrap(), "100");
975        assert_eq!(props.get(DELETED_RECORDS).unwrap(), "50");
976    }
977
978    #[test]
979    fn test_snapshot_summary_collector_merge() {
980        let schema = Arc::new(
981            Schema::builder()
982                .with_fields(vec![
983                    NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
984                    NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)).into(),
985                ])
986                .build()
987                .unwrap(),
988        );
989
990        let partition_spec = Arc::new(
991            PartitionSpec::builder(schema.clone())
992                .add_unbound_fields(vec![
993                    UnboundPartitionField::builder()
994                        .source_ids(vec![2])
995                        .name("year".to_string())
996                        .transform(Transform::Identity)
997                        .build()
998                        .unwrap(),
999                ])
1000                .unwrap()
1001                .with_spec_id(1)
1002                .build()
1003                .unwrap(),
1004        );
1005
1006        let mut summary_one = SnapshotSummaryCollector::default();
1007        let mut summary_two = SnapshotSummaryCollector::default();
1008
1009        summary_one.add_file(
1010            &DataFile {
1011                content: DataContentType::Data,
1012                file_path: "test.parquet".into(),
1013                file_format: DataFileFormat::Parquet,
1014                partition: Struct::from_iter(vec![]),
1015                record_count: 10,
1016                file_size_in_bytes: 100,
1017                column_sizes: HashMap::new(),
1018                value_counts: HashMap::new(),
1019                null_value_counts: HashMap::new(),
1020                nan_value_counts: HashMap::new(),
1021                lower_bounds: HashMap::new(),
1022                upper_bounds: HashMap::new(),
1023                key_metadata: None,
1024                split_offsets: None,
1025                equality_ids: None,
1026                sort_order_id: None,
1027                partition_spec_id: 0,
1028                first_row_id: None,
1029                referenced_data_file: None,
1030                content_offset: None,
1031                content_size_in_bytes: None,
1032            },
1033            schema.clone(),
1034            partition_spec.clone(),
1035        );
1036
1037        summary_two.add_file(
1038            &DataFile {
1039                content: DataContentType::Data,
1040                file_path: "test.parquet".into(),
1041                file_format: DataFileFormat::Parquet,
1042                partition: Struct::from_iter(vec![]),
1043                record_count: 20,
1044                file_size_in_bytes: 200,
1045                column_sizes: HashMap::new(),
1046                value_counts: HashMap::new(),
1047                null_value_counts: HashMap::new(),
1048                nan_value_counts: HashMap::new(),
1049                lower_bounds: HashMap::new(),
1050                upper_bounds: HashMap::new(),
1051                key_metadata: None,
1052                split_offsets: None,
1053                equality_ids: None,
1054                sort_order_id: None,
1055                partition_spec_id: 0,
1056                first_row_id: None,
1057                referenced_data_file: None,
1058                content_offset: None,
1059                content_size_in_bytes: None,
1060            },
1061            schema.clone(),
1062            partition_spec.clone(),
1063        );
1064
1065        summary_one.merge(summary_two);
1066        let props = summary_one.build();
1067        assert_eq!(props.get(ADDED_DATA_FILES).unwrap(), "2");
1068        assert_eq!(props.get(ADDED_RECORDS).unwrap(), "30");
1069
1070        let mut summary_three = SnapshotSummaryCollector::default();
1071        let mut summary_four = SnapshotSummaryCollector::default();
1072
1073        summary_three.add_manifest(&ManifestFile {
1074            manifest_path: "test.manifest".to_string(),
1075            manifest_length: 0,
1076            partition_spec_id: 0,
1077            content: ManifestContentType::Data,
1078            sequence_number: 0,
1079            min_sequence_number: 0,
1080            added_snapshot_id: 0,
1081            added_files_count: Some(1),
1082            existing_files_count: Some(0),
1083            deleted_files_count: Some(0),
1084            added_rows_count: Some(5),
1085            existing_rows_count: Some(0),
1086            deleted_rows_count: Some(0),
1087            partitions: Some(Vec::new()),
1088            key_metadata: None,
1089            first_row_id: None,
1090        });
1091
1092        summary_four.add_file(
1093            &DataFile {
1094                content: DataContentType::Data,
1095                file_path: "test.parquet".into(),
1096                file_format: DataFileFormat::Parquet,
1097                partition: Struct::from_iter(vec![]),
1098                record_count: 1,
1099                file_size_in_bytes: 10,
1100                column_sizes: HashMap::new(),
1101                value_counts: HashMap::new(),
1102                null_value_counts: HashMap::new(),
1103                nan_value_counts: HashMap::new(),
1104                lower_bounds: HashMap::new(),
1105                upper_bounds: HashMap::new(),
1106                key_metadata: None,
1107                split_offsets: None,
1108                equality_ids: None,
1109                sort_order_id: None,
1110                partition_spec_id: 0,
1111                first_row_id: None,
1112                referenced_data_file: None,
1113                content_offset: None,
1114                content_size_in_bytes: None,
1115            },
1116            schema.clone(),
1117            partition_spec.clone(),
1118        );
1119
1120        summary_three.merge(summary_four);
1121        let props = summary_three.build();
1122
1123        assert_eq!(props.get(ADDED_DATA_FILES).unwrap(), "2");
1124        assert_eq!(props.get(ADDED_RECORDS).unwrap(), "6");
1125        assert!(
1126            props
1127                .iter()
1128                .all(|(k, _)| !k.starts_with(CHANGED_PARTITION_PREFIX))
1129        );
1130    }
1131
1132    #[test]
1133    fn test_update_totals_skipped_when_previous_summary_missing_totals() {
1134        let prev_props: HashMap<String, String> = [(TOTAL_DATA_FILES, "8")]
1135            .into_iter()
1136            .map(|(k, v)| (k.to_string(), v.to_string()))
1137            .collect();
1138
1139        let previous_summary = Summary {
1140            operation: Operation::Overwrite,
1141            additional_properties: prev_props,
1142        };
1143
1144        let new_props: HashMap<String, String> = [
1145            (ADDED_DATA_FILES, "4"),
1146            (ADDED_DELETE_FILES, "2"),
1147            (ADDED_RECORDS, "40"),
1148            (ADDED_FILE_SIZE, "400"),
1149            (ADDED_POSITION_DELETES, "5"),
1150            (ADDED_EQUALITY_DELETES, "3"),
1151        ]
1152        .into_iter()
1153        .map(|(k, v)| (k.to_string(), v.to_string()))
1154        .collect();
1155
1156        let summary = Summary {
1157            operation: Operation::Append,
1158            additional_properties: new_props,
1159        };
1160
1161        let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap();
1162        let props = &updated.additional_properties;
1163
1164        assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "12");
1165
1166        for total_field in [
1167            TOTAL_DELETE_FILES,
1168            TOTAL_RECORDS,
1169            TOTAL_FILE_SIZE,
1170            TOTAL_POSITION_DELETES,
1171            TOTAL_EQUALITY_DELETES,
1172        ] {
1173            assert!(
1174                !props.contains_key(total_field),
1175                "{total_field} should not be set when previous summary lacks it",
1176            );
1177        }
1178    }
1179
1180    #[test]
1181    fn test_update_totals_tolerates_unparsable_added_value() {
1182        // A non-integer added value (which can survive in a previous snapshot's
1183        // summary) must not panic the commit. Matching iceberg-java's `updateTotal`
1184        // try/catch, the affected total is skipped while other totals still compute.
1185        let prev_props: HashMap<String, String> = [(TOTAL_DATA_FILES, "8"), (TOTAL_RECORDS, "80")]
1186            .into_iter()
1187            .map(|(k, v)| (k.to_string(), v.to_string()))
1188            .collect();
1189
1190        let previous_summary = Summary {
1191            operation: Operation::Append,
1192            additional_properties: prev_props,
1193        };
1194
1195        let new_props: HashMap<String, String> =
1196            [(ADDED_DATA_FILES, "not-a-number"), (ADDED_RECORDS, "40")]
1197                .into_iter()
1198                .map(|(k, v)| (k.to_string(), v.to_string()))
1199                .collect();
1200
1201        let summary = Summary {
1202            operation: Operation::Append,
1203            additional_properties: new_props,
1204        };
1205
1206        // Must not panic.
1207        let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap();
1208        let props = &updated.additional_properties;
1209
1210        // The total whose added delta was unparsable is skipped...
1211        assert!(
1212            !props.contains_key(TOTAL_DATA_FILES),
1213            "TOTAL_DATA_FILES should be skipped when its added value is unparsable",
1214        );
1215        // ...while a sibling total with valid deltas still computes.
1216        assert_eq!(props.get(TOTAL_RECORDS).unwrap(), "120");
1217    }
1218
1219    #[test]
1220    fn test_update_totals_computed_when_no_previous_summary() {
1221        let new_props: HashMap<String, String> = [
1222            (ADDED_DATA_FILES, "4"),
1223            (ADDED_RECORDS, "40"),
1224            (ADDED_FILE_SIZE, "400"),
1225        ]
1226        .into_iter()
1227        .map(|(k, v)| (k.to_string(), v.to_string()))
1228        .collect();
1229
1230        let summary = Summary {
1231            operation: Operation::Append,
1232            additional_properties: new_props,
1233        };
1234
1235        let updated = update_snapshot_summaries(summary, None, false).unwrap();
1236        let props = &updated.additional_properties;
1237
1238        assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "4");
1239        assert_eq!(props.get(TOTAL_RECORDS).unwrap(), "40");
1240        assert_eq!(props.get(TOTAL_FILE_SIZE).unwrap(), "400");
1241    }
1242
1243    #[test]
1244    fn test_update_totals_with_removes_only() {
1245        let prev_props: HashMap<String, String> = [
1246            (TOTAL_DATA_FILES, "10"),
1247            (TOTAL_DELETE_FILES, "5"),
1248            (TOTAL_RECORDS, "100"),
1249            (TOTAL_FILE_SIZE, "1000"),
1250            (TOTAL_POSITION_DELETES, "3"),
1251            (TOTAL_EQUALITY_DELETES, "2"),
1252        ]
1253        .into_iter()
1254        .map(|(k, v)| (k.to_string(), v.to_string()))
1255        .collect();
1256
1257        let previous_summary = Summary {
1258            operation: Operation::Overwrite,
1259            additional_properties: prev_props,
1260        };
1261
1262        let new_props: HashMap<String, String> = [
1263            (DELETED_DATA_FILES, "2"),
1264            (REMOVED_DELETE_FILES, "1"),
1265            (DELETED_RECORDS, "20"),
1266            (REMOVED_FILE_SIZE, "200"),
1267            (REMOVED_POSITION_DELETES, "1"),
1268            (REMOVED_EQUALITY_DELETES, "1"),
1269        ]
1270        .into_iter()
1271        .map(|(k, v)| (k.to_string(), v.to_string()))
1272        .collect();
1273
1274        let summary = Summary {
1275            operation: Operation::Delete,
1276            additional_properties: new_props,
1277        };
1278
1279        let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap();
1280        let props = &updated.additional_properties;
1281
1282        assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "8");
1283        assert_eq!(props.get(TOTAL_DELETE_FILES).unwrap(), "4");
1284        assert_eq!(props.get(TOTAL_RECORDS).unwrap(), "80");
1285        assert_eq!(props.get(TOTAL_FILE_SIZE).unwrap(), "800");
1286        assert_eq!(props.get(TOTAL_POSITION_DELETES).unwrap(), "2");
1287        assert_eq!(props.get(TOTAL_EQUALITY_DELETES).unwrap(), "1");
1288    }
1289
1290    #[test]
1291    fn test_update_totals_omits_total_when_removed_exceeds_previous_total() {
1292        let previous_summary = Summary {
1293            operation: Operation::Append,
1294            additional_properties: HashMap::from([(TOTAL_DATA_FILES.to_string(), "2".to_string())]),
1295        };
1296        let summary = Summary {
1297            operation: Operation::Delete,
1298            additional_properties: HashMap::from([(
1299                DELETED_DATA_FILES.to_string(),
1300                "3".to_string(),
1301            )]),
1302        };
1303
1304        let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap();
1305
1306        assert!(!updated.additional_properties.contains_key(TOTAL_DATA_FILES));
1307    }
1308}
iceberg/spec/snapshot_summary.rs

iceberg/spec/
snapshot_summary.rs
