iceberg/inspect/
manifests.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;
19use std::sync::Arc;
20
21use arrow_array::RecordBatch;
22use arrow_array::builder::{
23    BooleanBuilder, GenericListBuilder, ListBuilder, PrimitiveBuilder, StringBuilder, StructBuilder,
24};
25use arrow_array::types::{Int32Type, Int64Type};
26use arrow_schema::{DataType, Field, Fields};
27use futures::{StreamExt, stream};
28
29use crate::arrow::schema_to_arrow_schema;
30use crate::scan::ArrowRecordBatchStream;
31use crate::spec::{Datum, FieldSummary, ListType, NestedField, PrimitiveType, StructType, Type};
32use crate::table::Table;
33use crate::{Error, ErrorKind, Result};
34
35/// Manifests table.
36pub struct ManifestsTable<'a> {
37    table: &'a Table,
38}
39
40impl<'a> ManifestsTable<'a> {
41    /// Create a new Manifests table instance.
42    pub fn new(table: &'a Table) -> Self {
43        Self { table }
44    }
45
46    /// Returns the iceberg schema of the manifests table.
47    pub fn schema(&self) -> crate::spec::Schema {
48        let fields = vec![
49            NestedField::new(14, "content", Type::Primitive(PrimitiveType::Int), true),
50            NestedField::new(1, "path", Type::Primitive(PrimitiveType::String), true),
51            NestedField::new(2, "length", Type::Primitive(PrimitiveType::Long), true),
52            NestedField::new(
53                3,
54                "partition_spec_id",
55                Type::Primitive(PrimitiveType::Int),
56                true,
57            ),
58            NestedField::new(
59                4,
60                "added_snapshot_id",
61                Type::Primitive(PrimitiveType::Long),
62                true,
63            ),
64            NestedField::new(
65                5,
66                "added_data_files_count",
67                Type::Primitive(PrimitiveType::Int),
68                true,
69            ),
70            NestedField::new(
71                6,
72                "existing_data_files_count",
73                Type::Primitive(PrimitiveType::Int),
74                true,
75            ),
76            NestedField::new(
77                7,
78                "deleted_data_files_count",
79                Type::Primitive(PrimitiveType::Int),
80                true,
81            ),
82            NestedField::new(
83                15,
84                "added_delete_files_count",
85                Type::Primitive(PrimitiveType::Int),
86                true,
87            ),
88            NestedField::new(
89                16,
90                "existing_delete_files_count",
91                Type::Primitive(PrimitiveType::Int),
92                true,
93            ),
94            NestedField::new(
95                17,
96                "deleted_delete_files_count",
97                Type::Primitive(PrimitiveType::Int),
98                true,
99            ),
100            NestedField::new(
101                8,
102                "partition_summaries",
103                Type::List(ListType {
104                    element_field: Arc::new(NestedField::new(
105                        9,
106                        "item",
107                        Type::Struct(StructType::new(vec![
108                            Arc::new(NestedField::new(
109                                10,
110                                "contains_null",
111                                Type::Primitive(PrimitiveType::Boolean),
112                                true,
113                            )),
114                            Arc::new(NestedField::new(
115                                11,
116                                "contains_nan",
117                                Type::Primitive(PrimitiveType::Boolean),
118                                false,
119                            )),
120                            Arc::new(NestedField::new(
121                                12,
122                                "lower_bound",
123                                Type::Primitive(PrimitiveType::String),
124                                false,
125                            )),
126                            Arc::new(NestedField::new(
127                                13,
128                                "upper_bound",
129                                Type::Primitive(PrimitiveType::String),
130                                false,
131                            )),
132                        ])),
133                        true,
134                    )),
135                }),
136                true,
137            ),
138        ];
139
140        crate::spec::Schema::builder()
141            .with_fields(fields.into_iter().map(|f| f.into()))
142            .build()
143            .unwrap()
144    }
145
146    /// Scans the manifests table.
147    pub async fn scan(&self) -> Result<ArrowRecordBatchStream> {
148        let schema = schema_to_arrow_schema(&self.schema())?;
149
150        let mut content = PrimitiveBuilder::<Int32Type>::new();
151        let mut path = StringBuilder::new();
152        let mut length = PrimitiveBuilder::<Int64Type>::new();
153        let mut partition_spec_id = PrimitiveBuilder::<Int32Type>::new();
154        let mut added_snapshot_id = PrimitiveBuilder::<Int64Type>::new();
155        let mut added_data_files_count = PrimitiveBuilder::<Int32Type>::new();
156        let mut existing_data_files_count = PrimitiveBuilder::<Int32Type>::new();
157        let mut deleted_data_files_count = PrimitiveBuilder::<Int32Type>::new();
158        let mut added_delete_files_count = PrimitiveBuilder::<Int32Type>::new();
159        let mut existing_delete_files_count = PrimitiveBuilder::<Int32Type>::new();
160        let mut deleted_delete_files_count = PrimitiveBuilder::<Int32Type>::new();
161        let mut partition_summaries = self.partition_summary_builder()?;
162
163        if let Some(snapshot) = self.table.metadata().current_snapshot() {
164            let manifest_list = self.table.manifest_list_reader(snapshot).load().await?;
165            for manifest in manifest_list.entries() {
166                content.append_value(manifest.content as i32);
167                path.append_value(manifest.manifest_path.clone());
168                length.append_value(manifest.manifest_length);
169                partition_spec_id.append_value(manifest.partition_spec_id);
170                added_snapshot_id.append_value(manifest.added_snapshot_id);
171                added_data_files_count.append_value(manifest.added_files_count.unwrap_or(0) as i32);
172                existing_data_files_count
173                    .append_value(manifest.existing_files_count.unwrap_or(0) as i32);
174                deleted_data_files_count
175                    .append_value(manifest.deleted_files_count.unwrap_or(0) as i32);
176                added_delete_files_count
177                    .append_value(manifest.added_files_count.unwrap_or(0) as i32);
178                existing_delete_files_count
179                    .append_value(manifest.existing_files_count.unwrap_or(0) as i32);
180                deleted_delete_files_count
181                    .append_value(manifest.deleted_files_count.unwrap_or(0) as i32);
182
183                let spec = self
184                    .table
185                    .metadata()
186                    .partition_spec_by_id(manifest.partition_spec_id)
187                    .ok_or_else(|| {
188                        Error::new(
189                            ErrorKind::DataInvalid,
190                            format!(
191                                "Partition spec {} for manifest {} is not in table metadata",
192                                manifest.partition_spec_id, manifest.manifest_path
193                            ),
194                        )
195                    })?;
196                let spec_struct = spec.partition_type(self.table.metadata().current_schema())?;
197                self.append_partition_summaries(
198                    &mut partition_summaries,
199                    manifest.partitions.as_deref().unwrap_or(&[]),
200                    spec_struct,
201                );
202            }
203        }
204
205        let batch = RecordBatch::try_new(Arc::new(schema), vec![
206            Arc::new(content.finish()),
207            Arc::new(path.finish()),
208            Arc::new(length.finish()),
209            Arc::new(partition_spec_id.finish()),
210            Arc::new(added_snapshot_id.finish()),
211            Arc::new(added_data_files_count.finish()),
212            Arc::new(existing_data_files_count.finish()),
213            Arc::new(deleted_data_files_count.finish()),
214            Arc::new(added_delete_files_count.finish()),
215            Arc::new(existing_delete_files_count.finish()),
216            Arc::new(deleted_delete_files_count.finish()),
217            Arc::new(partition_summaries.finish()),
218        ])?;
219        Ok(stream::iter(vec![Ok(batch)]).boxed())
220    }
221
222    fn partition_summary_builder(&self) -> Result<GenericListBuilder<i32, StructBuilder>> {
223        let schema = schema_to_arrow_schema(&self.schema())?;
224        let partition_summary_fields =
225            match schema.field_with_name("partition_summaries")?.data_type() {
226                DataType::List(list_type) => match list_type.data_type() {
227                    DataType::Struct(fields) => fields.to_vec(),
228                    _ => unreachable!(),
229                },
230                _ => unreachable!(),
231            };
232
233        let partition_summaries = ListBuilder::new(StructBuilder::from_fields(
234            Fields::from(partition_summary_fields.clone()),
235            0,
236        ))
237        .with_field(Arc::new(
238            Field::new_struct("item", partition_summary_fields, false).with_metadata(
239                HashMap::from([("PARQUET:field_id".to_string(), "9".to_string())]),
240            ),
241        ));
242
243        Ok(partition_summaries)
244    }
245
246    fn append_partition_summaries(
247        &self,
248        builder: &mut GenericListBuilder<i32, StructBuilder>,
249        partitions: &[FieldSummary],
250        partition_struct: StructType,
251    ) {
252        let partition_summaries_builder = builder.values();
253        for (summary, field) in partitions.iter().zip(partition_struct.fields()) {
254            partition_summaries_builder
255                .field_builder::<BooleanBuilder>(0)
256                .unwrap()
257                .append_value(summary.contains_null);
258            partition_summaries_builder
259                .field_builder::<BooleanBuilder>(1)
260                .unwrap()
261                .append_option(summary.contains_nan);
262
263            partition_summaries_builder
264                .field_builder::<StringBuilder>(2)
265                .unwrap()
266                .append_option(summary.lower_bound.as_ref().map(|v| {
267                    Datum::try_from_bytes(v, field.field_type.as_primitive_type().unwrap().clone())
268                        .unwrap()
269                        .to_string()
270                }));
271            partition_summaries_builder
272                .field_builder::<StringBuilder>(3)
273                .unwrap()
274                .append_option(summary.upper_bound.as_ref().map(|v| {
275                    Datum::try_from_bytes(v, field.field_type.as_primitive_type().unwrap().clone())
276                        .unwrap()
277                        .to_string()
278                }));
279            partition_summaries_builder.append(true);
280        }
281        builder.append(true);
282    }
283}
284
285#[cfg(test)]
286mod tests {
287    use std::sync::Arc;
288
289    use expect_test::expect;
290    use futures::TryStreamExt;
291
292    use crate::scan::tests::TableTestFixture;
293    use crate::spec::TableMetadata;
294    use crate::test_utils::check_record_batches;
295
296    #[tokio::test]
297    async fn test_manifests_table() {
298        let mut fixture = TableTestFixture::new();
299        fixture.setup_manifest_files().await;
300
301        let record_batch = fixture.table.inspect().manifests().scan().await.unwrap();
302
303        check_record_batches(
304            record_batch.try_collect::<Vec<_>>().await.unwrap(),
305            expect![[r#"
306                Field { "content": Int32, metadata: {"PARQUET:field_id": "14"} },
307                Field { "path": Utf8, metadata: {"PARQUET:field_id": "1"} },
308                Field { "length": Int64, metadata: {"PARQUET:field_id": "2"} },
309                Field { "partition_spec_id": Int32, metadata: {"PARQUET:field_id": "3"} },
310                Field { "added_snapshot_id": Int64, metadata: {"PARQUET:field_id": "4"} },
311                Field { "added_data_files_count": Int32, metadata: {"PARQUET:field_id": "5"} },
312                Field { "existing_data_files_count": Int32, metadata: {"PARQUET:field_id": "6"} },
313                Field { "deleted_data_files_count": Int32, metadata: {"PARQUET:field_id": "7"} },
314                Field { "added_delete_files_count": Int32, metadata: {"PARQUET:field_id": "15"} },
315                Field { "existing_delete_files_count": Int32, metadata: {"PARQUET:field_id": "16"} },
316                Field { "deleted_delete_files_count": Int32, metadata: {"PARQUET:field_id": "17"} },
317                Field { "partition_summaries": List(non-null Struct("contains_null": non-null Boolean, metadata: {"PARQUET:field_id": "10"}, "contains_nan": Boolean, metadata: {"PARQUET:field_id": "11"}, "lower_bound": Utf8, metadata: {"PARQUET:field_id": "12"}, "upper_bound": Utf8, metadata: {"PARQUET:field_id": "13"}), metadata: {"PARQUET:field_id": "9"}), metadata: {"PARQUET:field_id": "8"} }"#]],
318            expect![[r#"
319                content: PrimitiveArray<Int32>
320                [
321                  0,
322                ],
323                path: (skipped),
324                length: (skipped),
325                partition_spec_id: PrimitiveArray<Int32>
326                [
327                  0,
328                ],
329                added_snapshot_id: PrimitiveArray<Int64>
330                [
331                  3055729675574597004,
332                ],
333                added_data_files_count: PrimitiveArray<Int32>
334                [
335                  1,
336                ],
337                existing_data_files_count: PrimitiveArray<Int32>
338                [
339                  1,
340                ],
341                deleted_data_files_count: PrimitiveArray<Int32>
342                [
343                  1,
344                ],
345                added_delete_files_count: PrimitiveArray<Int32>
346                [
347                  1,
348                ],
349                existing_delete_files_count: PrimitiveArray<Int32>
350                [
351                  1,
352                ],
353                deleted_delete_files_count: PrimitiveArray<Int32>
354                [
355                  1,
356                ],
357                partition_summaries: ListArray
358                [
359                  StructArray
360                -- validity:
361                [
362                  valid,
363                ]
364                [
365                -- child 0: "contains_null" (Boolean)
366                BooleanArray
367                [
368                  false,
369                ]
370                -- child 1: "contains_nan" (Boolean)
371                BooleanArray
372                [
373                  false,
374                ]
375                -- child 2: "lower_bound" (Utf8)
376                StringArray
377                [
378                  "100",
379                ]
380                -- child 3: "upper_bound" (Utf8)
381                StringArray
382                [
383                  "300",
384                ]
385                ],
386                ]"#]],
387            &["path", "length"],
388            Some("path"),
389        );
390    }
391
392    #[tokio::test]
393    async fn test_manifests_table_with_dropped_partition_source_column() {
394        let mut fixture = TableTestFixture::new();
395        fixture.setup_manifest_files().await;
396
397        // Evolve the table so that the manifests reference a historical spec whose source
398        // column is no longer in the current schema: add an unpartitioned default spec, then
399        // drop the source column of the original spec.
400        let mut metadata = serde_json::to_value(fixture.table.metadata()).unwrap();
401        let current_schema_id = metadata["current-schema-id"].clone();
402        let schemas = metadata["schemas"].as_array_mut().unwrap();
403        for schema in schemas {
404            if schema["schema-id"] == current_schema_id {
405                let fields = schema["fields"].as_array_mut().unwrap();
406                fields.retain(|field| field["id"] != 1);
407                let identifier_ids = schema["identifier-field-ids"].as_array_mut().unwrap();
408                identifier_ids.retain(|id| *id != 1);
409            }
410        }
411
412        metadata["partition-specs"]
413            .as_array_mut()
414            .unwrap()
415            .push(serde_json::json!({"spec-id": 1, "fields": []}));
416        metadata["default-spec-id"] = serde_json::json!(1);
417
418        let metadata: TableMetadata = serde_json::from_value(metadata).unwrap();
419        let table = fixture.table.clone().with_metadata(Arc::new(metadata));
420
421        match table.inspect().manifests().scan().await {
422            Err(err) => assert!(err.to_string().contains("No column with source column id")),
423            Ok(_) => panic!("expected scan to fail for a dropped partition source column"),
424        }
425    }
426}
iceberg/inspect/manifests.rs

iceberg/inspect/
manifests.rs
