datafusion_iceberg 0.1.2

TableProvider to Apache iceberg tables with Datafusion
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

use std::{any::Any, sync::Arc};

use datafusion::{
    catalog::schema::SchemaProvider,
    datasource::TableProvider,
    error::{DataFusionError, Result},
};
use iceberg_rust::catalog::{identifier::Identifier, namespace::Namespace};

use crate::catalog::mirror::Mirror;

pub struct IcebergSchema {
    schema: Namespace,
    catalog: Arc<Mirror>,
}

impl IcebergSchema {
    pub(crate) fn new(schema: Namespace, catalog: Arc<Mirror>) -> Self {
        IcebergSchema { schema, catalog }
    }
}

#[async_trait::async_trait]
impl SchemaProvider for IcebergSchema {
    fn as_any(&self) -> &dyn Any {
        self
    }
    fn table_names(&self) -> Vec<String> {
        let tables = self.catalog.table_names(&self.schema);
        match tables {
            Err(_) => vec![],
            Ok(schemas) => schemas.into_iter().map(|x| x.name().to_owned()).collect(),
        }
    }
    async fn table(&self, name: &str) -> Option<Arc<dyn TableProvider>> {
        self.catalog
            .table(
                Identifier::try_new(&[self.schema.levels(), &[name.to_string()]].concat()).unwrap(),
            )
            .await
    }
    fn table_exist(&self, name: &str) -> bool {
        self.catalog.table_exists(
            Identifier::try_new(&[self.schema.levels(), &[name.to_string()]].concat()).unwrap(),
        )
    }

    fn register_table(
        &self,
        name: String,
        table: Arc<dyn TableProvider>,
    ) -> Result<Option<Arc<dyn TableProvider>>> {
        let mut full_name = Vec::from(self.schema.levels());
        full_name.push(name.to_owned());
        let identifier = Identifier::try_new(&full_name)
            .map_err(|err| DataFusionError::Internal(err.to_string()))?;
        self.catalog.register_table(identifier, table)
    }
    fn deregister_table(&self, name: &str) -> Result<Option<Arc<dyn TableProvider>>> {
        let mut full_name = Vec::from(self.schema.levels());
        full_name.push(name.to_owned());
        let identifier = Identifier::try_new(&full_name)
            .map_err(|err| DataFusionError::Internal(err.to_string()))?;
        self.catalog.deregister_table(identifier)
    }
}