iceberg/
table.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
18//! Table API for Apache Iceberg
19
20use std::sync::Arc;
21
22use crate::arrow::ArrowReaderBuilder;
23use crate::encryption::EncryptionManager;
24use crate::encryption::kms::KeyManagementClient;
25use crate::inspect::MetadataTable;
26use crate::io::FileIO;
27use crate::io::object_cache::ObjectCache;
28use crate::runtime::Runtime;
29use crate::scan::TableScanBuilder;
30use crate::spec::{ManifestListReader, SchemaRef, SnapshotRef, TableMetadata, TableMetadataRef};
31use crate::{Error, ErrorKind, Result, TableIdent};
32
33/// Builder to create table scan.
34pub struct TableBuilder {
35    file_io: Option<FileIO>,
36    metadata_location: Option<String>,
37    metadata: Option<TableMetadataRef>,
38    identifier: Option<TableIdent>,
39    kms_client: Option<Arc<dyn KeyManagementClient>>,
40    readonly: bool,
41    disable_cache: bool,
42    cache_size_bytes: Option<u64>,
43    runtime: Option<Runtime>,
44}
45
46impl TableBuilder {
47    pub(crate) fn new() -> Self {
48        Self {
49            file_io: None,
50            metadata_location: None,
51            metadata: None,
52            identifier: None,
53            kms_client: None,
54            readonly: false,
55            disable_cache: false,
56            cache_size_bytes: None,
57            runtime: None,
58        }
59    }
60
61    /// required - sets the necessary FileIO to use for the table
62    pub fn file_io(mut self, file_io: FileIO) -> Self {
63        self.file_io = Some(file_io);
64        self
65    }
66
67    /// optional - sets the tables metadata location
68    pub fn metadata_location<T: Into<String>>(mut self, metadata_location: T) -> Self {
69        self.metadata_location = Some(metadata_location.into());
70        self
71    }
72
73    /// required - passes in the TableMetadata to use for the Table
74    pub fn metadata<T: Into<TableMetadataRef>>(mut self, metadata: T) -> Self {
75        self.metadata = Some(metadata.into());
76        self
77    }
78
79    /// required - passes in the TableIdent to use for the Table
80    pub fn identifier(mut self, identifier: TableIdent) -> Self {
81        self.identifier = Some(identifier);
82        self
83    }
84
85    /// specifies if the Table is readonly or not (default not)
86    pub fn readonly(mut self, readonly: bool) -> Self {
87        self.readonly = readonly;
88        self
89    }
90
91    /// specifies if the Table's metadata cache will be disabled,
92    /// so that reads of Manifests and ManifestLists will never
93    /// get cached.
94    pub fn disable_cache(mut self) -> Self {
95        self.disable_cache = true;
96        self
97    }
98
99    /// optionally set a non-default metadata cache size
100    pub fn cache_size_bytes(mut self, cache_size_bytes: u64) -> Self {
101        self.cache_size_bytes = Some(cache_size_bytes);
102        self
103    }
104
105    /// Set the Runtime for this table to use when spawning tasks.
106    pub fn runtime(mut self, runtime: Runtime) -> Self {
107        self.runtime = Some(runtime);
108        self
109    }
110
111    /// optional - sets the KMS client used to unwrap keys for table encryption.
112    ///
113    /// If the table metadata has the `encryption.key-id` property set, a
114    /// [`KeyManagementClient`] must be provided here so the table can build
115    /// an [`EncryptionManager`]; otherwise [`Self::build`] will return an error.
116    pub fn kms_client(mut self, kms_client: Arc<dyn KeyManagementClient>) -> Self {
117        self.kms_client = Some(kms_client);
118        self
119    }
120
121    /// build the Table
122    pub fn build(self) -> Result<Table> {
123        let Self {
124            file_io,
125            metadata_location,
126            metadata,
127            identifier,
128            kms_client,
129            readonly,
130            disable_cache,
131            cache_size_bytes,
132            runtime,
133        } = self;
134
135        let Some(file_io) = file_io else {
136            return Err(Error::new(
137                ErrorKind::DataInvalid,
138                "FileIO must be provided with TableBuilder.file_io()",
139            ));
140        };
141
142        let Some(metadata) = metadata else {
143            return Err(Error::new(
144                ErrorKind::DataInvalid,
145                "TableMetadataRef must be provided with TableBuilder.metadata()",
146            ));
147        };
148
149        let Some(identifier) = identifier else {
150            return Err(Error::new(
151                ErrorKind::DataInvalid,
152                "TableIdent must be provided with TableBuilder.identifier()",
153            ));
154        };
155
156        let Some(runtime) = runtime else {
157            return Err(Error::new(
158                ErrorKind::DataInvalid,
159                "Runtime must be provided with TableBuilder.runtime()",
160            ));
161        };
162
163        let encryption_manager =
164            EncryptionManager::from_table_metadata(kms_client.as_ref(), &metadata)?;
165
166        let object_cache = if disable_cache {
167            Arc::new(ObjectCache::with_disabled_cache(
168                file_io.clone(),
169                encryption_manager.clone(),
170            ))
171        } else if let Some(cache_size_bytes) = cache_size_bytes {
172            Arc::new(ObjectCache::new_with_capacity(
173                file_io.clone(),
174                cache_size_bytes,
175                encryption_manager.clone(),
176            ))
177        } else {
178            Arc::new(ObjectCache::new(
179                file_io.clone(),
180                encryption_manager.clone(),
181            ))
182        };
183
184        Ok(Table {
185            file_io,
186            metadata_location,
187            metadata,
188            identifier,
189            readonly,
190            object_cache,
191            runtime,
192            encryption_manager,
193        })
194    }
195}
196
197/// Table represents a table in the catalog.
198#[derive(Debug, Clone)]
199pub struct Table {
200    file_io: FileIO,
201    metadata_location: Option<String>,
202    metadata: TableMetadataRef,
203    identifier: TableIdent,
204    readonly: bool,
205    object_cache: Arc<ObjectCache>,
206    runtime: Runtime,
207    encryption_manager: Option<Arc<EncryptionManager>>,
208}
209
210impl Table {
211    /// Sets the [`Table`] metadata and returns an updated instance with the new metadata applied.
212    pub(crate) fn with_metadata(mut self, metadata: TableMetadataRef) -> Self {
213        self.metadata = metadata;
214        self
215    }
216
217    /// Sets the [`Table`] metadata location and returns an updated instance.
218    pub(crate) fn with_metadata_location(mut self, metadata_location: String) -> Self {
219        self.metadata_location = Some(metadata_location);
220        self
221    }
222
223    /// Sets the [`Table`] `FileIO` and returns an updated instance.
224    pub(crate) fn with_file_io(mut self, file_io: FileIO) -> Self {
225        self.object_cache = Arc::new(
226            self.object_cache
227                .as_ref()
228                .clone()
229                .with_file_io(file_io.clone()),
230        );
231        self.file_io = file_io;
232        self
233    }
234
235    /// Returns a TableBuilder to build a table
236    pub fn builder() -> TableBuilder {
237        TableBuilder::new()
238    }
239
240    /// Returns table identifier.
241    pub fn identifier(&self) -> &TableIdent {
242        &self.identifier
243    }
244    /// Returns current metadata.
245    pub fn metadata(&self) -> &TableMetadata {
246        &self.metadata
247    }
248
249    /// Returns current metadata ref.
250    pub fn metadata_ref(&self) -> TableMetadataRef {
251        self.metadata.clone()
252    }
253
254    /// Returns current metadata location.
255    pub fn metadata_location(&self) -> Option<&str> {
256        self.metadata_location.as_deref()
257    }
258
259    /// Returns current metadata location in a result.
260    pub fn metadata_location_result(&self) -> Result<&str> {
261        self.metadata_location.as_deref().ok_or(Error::new(
262            ErrorKind::DataInvalid,
263            format!(
264                "Metadata location does not exist for table: {}",
265                self.identifier
266            ),
267        ))
268    }
269
270    /// Returns file io used in this table.
271    pub fn file_io(&self) -> &FileIO {
272        &self.file_io
273    }
274
275    /// Returns this table's object cache
276    pub(crate) fn object_cache(&self) -> Arc<ObjectCache> {
277        self.object_cache.clone()
278    }
279
280    /// Returns the [`EncryptionManager`] for this table, if encryption is
281    /// configured.
282    ///
283    /// A manager is present iff the table metadata has the
284    /// `encryption.key-id` property set and a [`KeyManagementClient`] was
285    /// supplied to the [`TableBuilder`].
286    pub fn encryption_manager(&self) -> Option<&EncryptionManager> {
287        self.encryption_manager.as_deref()
288    }
289
290    /// Creates a table scan.
291    pub fn scan(&self) -> TableScanBuilder<'_> {
292        TableScanBuilder::new(self)
293    }
294
295    /// Creates a metadata table which provides table-like APIs for inspecting metadata.
296    /// See [`MetadataTable`] for more details.
297    pub fn inspect(&self) -> MetadataTable<'_> {
298        MetadataTable::new(self)
299    }
300
301    /// Returns the [`Runtime`] for this table.
302    pub(crate) fn runtime(&self) -> &Runtime {
303        &self.runtime
304    }
305
306    /// Returns the flag indicating whether the `Table` is readonly or not
307    pub fn readonly(&self) -> bool {
308        self.readonly
309    }
310
311    /// Returns the current schema as a shared reference.
312    pub fn current_schema_ref(&self) -> SchemaRef {
313        self.metadata.current_schema().clone()
314    }
315
316    /// Creates a [`ManifestListReader`] for the given snapshot.
317    pub fn manifest_list_reader(&self, snapshot: &SnapshotRef) -> ManifestListReader {
318        ManifestListReader::new(
319            snapshot.clone(),
320            self.file_io.clone(),
321            self.metadata.clone(),
322            self.encryption_manager.clone(),
323        )
324    }
325
326    /// Create a reader for the table.
327    pub fn reader_builder(&self) -> ArrowReaderBuilder {
328        ArrowReaderBuilder::new(self.file_io.clone(), self.runtime().clone())
329    }
330}
331
332/// `StaticTable` is a read-only table struct that can be created from a metadata file or from `TableMetaData` without a catalog.
333/// It can only be used to read metadata and for table scan.
334/// # Examples
335///
336/// ```rust, no_run
337/// # use iceberg::io::FileIO;
338/// # use iceberg::table::StaticTable;
339/// # use iceberg::TableIdent;
340/// # async fn example() {
341/// let metadata_file_location = "s3://bucket_name/path/to/metadata.json";
342/// let file_io = FileIO::new_with_fs();
343/// let static_identifier = TableIdent::from_strs(["static_ns", "static_table"]).unwrap();
344/// let static_table =
345///     StaticTable::from_metadata_file(&metadata_file_location, static_identifier, file_io)
346///         .await
347///         .unwrap();
348/// let snapshot_id = static_table
349///     .metadata()
350///     .current_snapshot()
351///     .unwrap()
352///     .snapshot_id();
353/// # }
354/// ```
355#[derive(Debug, Clone)]
356pub struct StaticTable(Table);
357
358impl StaticTable {
359    /// Creates a static table from a given `TableMetadata` and `FileIO`
360    pub async fn from_metadata(
361        metadata: TableMetadata,
362        table_ident: TableIdent,
363        file_io: FileIO,
364    ) -> Result<Self> {
365        let table = Table::builder()
366            .metadata(metadata)
367            .identifier(table_ident)
368            .file_io(file_io.clone())
369            .runtime(Runtime::try_current()?)
370            .readonly(true)
371            .build();
372
373        Ok(Self(table?))
374    }
375    /// Creates a static table directly from metadata file and `FileIO`
376    pub async fn from_metadata_file(
377        metadata_location: &str,
378        table_ident: TableIdent,
379        file_io: FileIO,
380    ) -> Result<Self> {
381        let metadata = TableMetadata::read_from(&file_io, metadata_location).await?;
382
383        let table = Table::builder()
384            .metadata(metadata)
385            .metadata_location(metadata_location)
386            .identifier(table_ident)
387            .file_io(file_io.clone())
388            .runtime(Runtime::try_current()?)
389            .readonly(true)
390            .build();
391
392        Ok(Self(table?))
393    }
394
395    /// Create a TableScanBuilder for the static table.
396    pub fn scan(&self) -> TableScanBuilder<'_> {
397        self.0.scan()
398    }
399
400    /// Get TableMetadataRef for the static table
401    pub fn metadata(&self) -> TableMetadataRef {
402        self.0.metadata_ref()
403    }
404
405    /// Consumes the `StaticTable` and return it as a `Table`
406    /// Please use this method carefully as the Table it returns remains detached from a catalog
407    /// and can't be used to perform modifications on the table.
408    pub fn into_table(self) -> Table {
409        self.0
410    }
411
412    /// Create a reader for the table.
413    pub fn reader_builder(&self) -> ArrowReaderBuilder {
414        self.0.reader_builder()
415    }
416}
417
418#[cfg(test)]
419mod tests {
420    use std::fs;
421
422    use super::*;
423    use crate::encryption::SensitiveBytes;
424    use crate::encryption::kms::MemoryKeyManagementClient;
425    use crate::io::{FileIOBuilder, MemoryStorageFactory};
426    use crate::spec::TableProperties;
427    use crate::test_utils::test_runtime;
428
429    fn load_test_metadata(filename: &str) -> TableMetadata {
430        let path = format!(
431            "{}/testdata/table_metadata/{}",
432            env!("CARGO_MANIFEST_DIR"),
433            filename
434        );
435        let json = fs::read_to_string(path).unwrap();
436        serde_json::from_str(&json).unwrap()
437    }
438
439    #[tokio::test]
440    async fn test_static_table_from_file() {
441        let metadata_file_name = "TableMetadataV2Valid.json";
442        let metadata_file_path = format!(
443            "{}/testdata/table_metadata/{}",
444            env!("CARGO_MANIFEST_DIR"),
445            metadata_file_name
446        );
447        let file_io = FileIO::new_with_fs();
448        let static_identifier = TableIdent::from_strs(["static_ns", "static_table"]).unwrap();
449        let static_table =
450            StaticTable::from_metadata_file(&metadata_file_path, static_identifier, file_io)
451                .await
452                .unwrap();
453        let snapshot_id = static_table
454            .metadata()
455            .current_snapshot()
456            .unwrap()
457            .snapshot_id();
458        assert_eq!(
459            snapshot_id, 3055729675574597004,
460            "snapshot id from metadata don't match"
461        );
462    }
463
464    #[tokio::test]
465    async fn test_static_into_table() {
466        let metadata_file_name = "TableMetadataV2Valid.json";
467        let metadata_file_path = format!(
468            "{}/testdata/table_metadata/{}",
469            env!("CARGO_MANIFEST_DIR"),
470            metadata_file_name
471        );
472        let file_io = FileIO::new_with_fs();
473        let static_identifier = TableIdent::from_strs(["static_ns", "static_table"]).unwrap();
474        let static_table =
475            StaticTable::from_metadata_file(&metadata_file_path, static_identifier, file_io)
476                .await
477                .unwrap();
478        let table = static_table.into_table();
479        assert!(table.readonly());
480        assert_eq!(table.identifier.name(), "static_table");
481        assert_eq!(
482            table.metadata_location(),
483            Some(metadata_file_path).as_deref()
484        );
485    }
486
487    #[tokio::test]
488    async fn test_table_readonly_flag() {
489        let metadata_file_name = "TableMetadataV2Valid.json";
490        let metadata_file_path = format!(
491            "{}/testdata/table_metadata/{}",
492            env!("CARGO_MANIFEST_DIR"),
493            metadata_file_name
494        );
495        let file_io = FileIO::new_with_fs();
496        let metadata_file = file_io.new_input(metadata_file_path).unwrap();
497        let metadata_file_content = metadata_file.read().await.unwrap();
498        let table_metadata =
499            serde_json::from_slice::<TableMetadata>(&metadata_file_content).unwrap();
500        let static_identifier = TableIdent::from_strs(["ns", "table"]).unwrap();
501        let table = Table::builder()
502            .metadata(table_metadata)
503            .identifier(static_identifier)
504            .file_io(file_io.clone())
505            .runtime(Runtime::try_current().unwrap())
506            .build()
507            .unwrap();
508        assert!(!table.readonly());
509        assert_eq!(table.identifier.name(), "table");
510    }
511
512    #[test]
513    fn test_with_file_io_updates_object_cache_file_io() {
514        let metadata = load_test_metadata("TableMetadataV2ValidMinimal.json");
515        let original_file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory))
516            .with_prop("marker", "original")
517            .build();
518        let replacement_file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory))
519            .with_prop("marker", "replacement")
520            .build();
521
522        let table = Table::builder()
523            .metadata(metadata)
524            .identifier(TableIdent::from_strs(["ns", "table"]).unwrap())
525            .file_io(original_file_io)
526            .runtime(test_runtime())
527            .build()
528            .unwrap()
529            .with_file_io(replacement_file_io);
530
531        assert_eq!(
532            table.file_io().config().get("marker").map(String::as_str),
533            Some("replacement")
534        );
535        assert_eq!(
536            table
537                .object_cache()
538                .file_io()
539                .config()
540                .get("marker")
541                .map(String::as_str),
542            Some("replacement")
543        );
544    }
545
546    fn make_kms() -> Arc<dyn KeyManagementClient> {
547        let kms = MemoryKeyManagementClient::new();
548        kms.add_master_key("master-1").unwrap();
549        Arc::new(kms)
550    }
551
552    #[tokio::test]
553    async fn table_decrypts_manifest_list_via_object_cache() {
554        // The fixture contains a snapshot with key-id, encryption-keys (KEK + wrapped DEK),
555        // all generated with the master key bytes below.
556        let mut metadata: TableMetadata = load_test_metadata("TableMetadataV3ValidEncryption.json");
557
558        // Point the snapshot's manifest-list at the testdata file on disk.
559        let manifest_list_path = format!(
560            "{}/testdata/manifests_lists/manifest-list-v3-encrypted.avro",
561            env!("CARGO_MANIFEST_DIR"),
562        );
563        let snapshot = metadata.snapshots.get_mut(&1).unwrap();
564        let mut patched = snapshot.as_ref().clone();
565        patched.manifest_list = manifest_list_path;
566        *snapshot = Arc::new(patched);
567
568        // Seed the KMS with the same master key bytes used to generate the fixture.
569        let kms: Arc<dyn KeyManagementClient> = {
570            let k = MemoryKeyManagementClient::new();
571            k.add_master_key_bytes(
572                "master-1",
573                SensitiveBytes::new([
574                    0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c,
575                    0x0d, 0x0e, 0x0f,
576                ]),
577            )
578            .unwrap();
579            Arc::new(k)
580        };
581
582        let table = Table::builder()
583            .file_io(FileIO::new_with_fs())
584            .metadata(metadata)
585            .identifier(TableIdent::from_strs(["ns", "enc"]).unwrap())
586            .kms_client(kms)
587            .runtime(Runtime::try_current().unwrap())
588            .build()
589            .unwrap();
590
591        let snapshot_ref = table.metadata().current_snapshot().unwrap();
592        let manifest_list = table
593            .object_cache()
594            .get_manifest_list(snapshot_ref, &table.metadata_ref())
595            .await
596            .unwrap();
597        assert_eq!(manifest_list.entries().len(), 0);
598    }
599
600    #[tokio::test]
601    async fn table_builder_errors_when_encryption_key_id_set_but_no_kms() {
602        let metadata: TableMetadata = load_test_metadata("TableMetadataV3ValidEncryption.json");
603
604        let err = Table::builder()
605            .file_io(FileIO::new_with_memory())
606            .metadata(metadata)
607            .identifier(TableIdent::from_strs(["ns", "enc"]).unwrap())
608            .runtime(Runtime::try_current().unwrap())
609            .build()
610            .unwrap_err();
611        assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
612    }
613
614    #[tokio::test]
615    async fn table_builder_skips_encryption_on_pre_v3_table() {
616        // Encryption is a v3 spec feature; pre-v3 tables silently skip
617        // encryption even if encryption.key-id is set.
618        let mut metadata: TableMetadata = load_test_metadata("TableMetadataV2ValidMinimal.json");
619        metadata.properties.insert(
620            TableProperties::PROPERTY_ENCRYPTION_KEY_ID.to_string(),
621            "master-1".to_string(),
622        );
623
624        let table = Table::builder()
625            .file_io(FileIO::new_with_memory())
626            .metadata(metadata)
627            .identifier(TableIdent::from_strs(["ns", "enc"]).unwrap())
628            .kms_client(make_kms())
629            .runtime(Runtime::try_current().unwrap())
630            .build()
631            .unwrap();
632        assert!(table.encryption_manager().is_none());
633    }
634
635    #[tokio::test]
636    async fn table_builder_skips_encryption_when_property_absent() {
637        let metadata: TableMetadata = load_test_metadata("TableMetadataV3ValidMinimal.json");
638        let table = Table::builder()
639            .file_io(FileIO::new_with_memory())
640            .metadata(metadata)
641            .identifier(TableIdent::from_strs(["ns", "plain"]).unwrap())
642            .kms_client(make_kms())
643            .runtime(Runtime::try_current().unwrap())
644            .build()
645            .unwrap();
646        assert!(table.encryption_manager().is_none());
647    }
648}
iceberg/table.rs

iceberg/
table.rs
