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

iceberg/
table.rs
