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

iceberg/
table.rs
