Register tables from config (#22)

* Remove iceberg call from create table

* Remove invalidate_cache

* Init

* Read tables from config

* Read tables from config

* Fix avro

Author: osipovartem

Cargo.lock

@@ -44,15 +44,16 @@ datafusion-expr = { version = "50.0.0" }
 datafusion-functions-json = { git = "https://github.com/Embucket/datafusion-functions-json.git", rev = "439cbd2282504c3ffaf262f1ffdb530a0fb1a151" }
 datafusion-macros = { version = "50.0.0" }
 datafusion-physical-plan = { version = "50.0.0" }
-datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "1eeb4515446119dd3b4dbb7ebd2f70ae5b4f827d" }
+datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b5089dc285667858469a6c2abb297af3d81a03f5" }
 futures = { version = "0.3" }
 http = "1.2"
 http-body-util = "0.1.0"
 iceberg = { git = "https://github.com/apache/iceberg-rust.git", rev="7a5ad1fcaf00d4638857812bab788105f6c60573"}
-iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "1eeb4515446119dd3b4dbb7ebd2f70ae5b4f827d" }
-iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "1eeb4515446119dd3b4dbb7ebd2f70ae5b4f827d" }
-iceberg-rust-spec = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "1eeb4515446119dd3b4dbb7ebd2f70ae5b4f827d" }
-iceberg-s3tables-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "1eeb4515446119dd3b4dbb7ebd2f70ae5b4f827d" }
+iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b5089dc285667858469a6c2abb297af3d81a03f5" }
+iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b5089dc285667858469a6c2abb297af3d81a03f5" }
+iceberg-rust-spec = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b5089dc285667858469a6c2abb297af3d81a03f5" }
+iceberg-s3tables-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b5089dc285667858469a6c2abb297af3d81a03f5" }
+
 indexmap = "2.7.1"
 jsonwebtoken = "9.3.1"
 lazy_static = { version = "1.5" }

Cargo.toml

@@ -1,5 +1,3 @@
-use std::{collections::HashMap, sync::Arc};
-
 use crate::error::{self as metastore_error, Result};
 use crate::models::{
     RwObject,
@@ -20,6 +18,7 @@ use iceberg_rust_spec::{
 };
 use object_store::{ObjectStore, PutPayload, path::Path};
 use snafu::ResultExt;
+use std::{collections::HashMap, sync::Arc};
 use tokio::sync::RwLock;
 use tracing::instrument;
 use uuid::Uuid;
@@ -60,6 +59,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
         ident: &TableIdent,
         table: TableCreateRequest,
     ) -> Result<RwObject<Table>>;
+    async fn register_table(&self, ident: &TableIdent, table: Table) -> Result<RwObject<Table>>;
     async fn get_table(&self, ident: &TableIdent) -> Result<Option<RwObject<Table>>>;
     async fn update_table(
         &self,
@@ -110,7 +110,8 @@ impl InMemoryMetastore {
         )
     }
 
-    fn table_key(ident: &TableIdent) -> (DatabaseIdent, String, String) {
+    #[must_use]
+    pub fn table_key(ident: &TableIdent) -> (DatabaseIdent, String, String) {
         (
             ident.database.to_ascii_lowercase(),
             ident.schema.to_ascii_lowercase(),
@@ -558,6 +559,13 @@ impl Metastore for InMemoryMetastore {
         Ok(row)
     }
 
+    async fn register_table(&self, ident: &TableIdent, table: Table) -> Result<RwObject<Table>> {
+        let mut state = self.state.write().await;
+        let row = RwObject::new(table);
+        state.tables.insert(Self::table_key(ident), row.clone());
+        Ok(row)
+    }
+
     async fn get_table(&self, ident: &TableIdent) -> Result<Option<RwObject<Table>>> {
         let state = self.state.read().await;
         Ok(state.tables.get(&Self::table_key(ident)).cloned())

crates/catalog-metastore/src/metastore.rs

@@ -19,6 +19,8 @@ tracing-subscriber = { version = "0.3.20", features = ["env-filter", "registry",
 tracing-opentelemetry = { version = "0.31.0" }
 tracing-allocations = { version = "0.1.0", optional = true }
 tracing-core = { version = "0.1.34"}
+object_store = { workspace = true }
+iceberg-rust = { workspace = true }
 opentelemetry-otlp = { version = "0.30.0", features = ["grpc-tonic"] }
 opentelemetry_sdk = { version = "0.30.0", features = [
     "experimental_trace_batch_span_processor_with_async_runtime",
@@ -36,6 +38,7 @@ utoipa-swagger-ui = { workspace = true }
 strum.workspace = true
 serde_yaml = { workspace = true }
 serde = { workspace = true }
+serde_json = { workspace = true }
 snafu = { workspace = true }
 
 [lints]

crates/embucketd/Cargo.toml

@@ -1,11 +1,16 @@
+use catalog_metastore::{
+    Database, Metastore, Schema, SchemaIdent, TableFormat, TableIdent, Volume, VolumeIdent,
+};
+use iceberg_rust::spec::table_metadata::TableMetadata;
+use iceberg_rust::spec::util::strip_prefix;
+use serde::Deserialize;
+use serde_json::Value;
+use snafu::prelude::*;
+use std::collections::HashMap;
 use std::{
     path::{Path, PathBuf},
     sync::Arc,
 };
-
-use catalog_metastore::{Database, Metastore, Schema, SchemaIdent, Volume, VolumeIdent};
-use serde::Deserialize;
-use snafu::prelude::*;
 use tokio::fs;
 
 #[derive(Debug, Deserialize, Default)]
@@ -16,6 +21,8 @@ pub struct MetastoreBootstrapConfig {
     databases: Vec<DatabaseEntry>,
     #[serde(default)]
     schemas: Vec<SchemaEntry>,
+    #[serde(default)]
+    tables: Vec<TableEntry>,
 }
 
 #[derive(Debug, Deserialize, Clone)]
@@ -38,6 +45,20 @@ struct SchemaEntry {
     schema: String,
 }
 
+#[derive(Debug, Deserialize, Clone)]
+struct TableEntry {
+    database: String,
+    schema: String,
+    table: String,
+    metadata_location: String,
+}
+
+impl TableEntry {
+    fn table_ident(&self) -> TableIdent {
+        TableIdent::new(&self.database, &self.schema, &self.table)
+    }
+}
+
 #[derive(Debug, Snafu)]
 pub enum ConfigError {
     #[snafu(display("Failed to read metastore config {path:?}: {source}"))]
@@ -54,6 +75,26 @@ pub enum ConfigError {
     Metastore {
         source: catalog_metastore::error::Error,
     },
+    #[snafu(display("Database {database} not found for table {table}"))]
+    TableDatabaseMissing { table: String, database: String },
+    #[snafu(display("Volume {volume} not found for table {table}"))]
+    TableVolumeMissing { table: String, volume: VolumeIdent },
+    #[snafu(display("Invalid metadata location for table {table}: {reason}"))]
+    InvalidMetadataLocation { table: String, reason: String },
+    #[snafu(display("Invalid metadata"))]
+    InvalidMetadata,
+    #[snafu(display("Failed to fetch metadata for table {table}: {source}"))]
+    MetadataFetch {
+        table: String,
+        #[snafu(source)]
+        source: object_store::Error,
+    },
+    #[snafu(display("Failed to parse metadata for table {table}: {source}"))]
+    MetadataParse {
+        table: String,
+        #[snafu(source)]
+        source: serde_json::Error,
+    },
 }
 
 const DEFAULT_SCHEMA_NAME: &str = "public";
@@ -84,6 +125,10 @@ impl MetastoreBootstrapConfig {
                 .await?;
         }
 
+        for table in &self.tables {
+            self.apply_table(table, metastore.clone()).await?;
+        }
+
         Ok(())
     }
 
@@ -182,4 +227,110 @@ impl MetastoreBootstrapConfig {
         }
         Ok(())
     }
+
+    async fn apply_table(
+        &self,
+        entry: &TableEntry,
+        metastore: Arc<dyn Metastore>,
+    ) -> Result<(), ConfigError> {
+        let table_ident = entry.table_ident();
+        let table_name = entry.table.clone();
+        if metastore
+            .table_exists(&table_ident)
+            .await
+            .context(MetastoreSnafu)?
+        {
+            tracing::debug!(table = %table_name, "Table already exists, skipping config create");
+            return Ok(());
+        }
+
+        let database = metastore
+            .get_database(&entry.database)
+            .await
+            .context(MetastoreSnafu)?
+            .ok_or_else(|| ConfigError::TableDatabaseMissing {
+                table: table_name.clone(),
+                database: entry.database.clone(),
+            })?;
+
+        self.ensure_schema(metastore.clone(), &entry.database, &entry.schema)
+            .await?;
+
+        let volume_ident = database.volume.clone();
+        let volume = metastore
+            .get_volume(&volume_ident)
+            .await
+            .context(MetastoreSnafu)?
+            .ok_or_else(|| ConfigError::TableVolumeMissing {
+                table: table_name.clone(),
+                volume: volume_ident.clone(),
+            })?;
+        let table_object_store = volume.get_object_store().context(MetastoreSnafu)?;
+
+        let bytes = table_object_store
+            .get(
+                &strip_prefix(&entry.metadata_location.clone())
+                    .as_str()
+                    .into(),
+            )
+            .await
+            .map_err(|e| ConfigError::InvalidMetadataLocation {
+                table: table_name.clone(),
+                reason: e.to_string(),
+            })?
+            .bytes()
+            .await
+            .context(MetadataFetchSnafu {
+                table: table_name.clone(),
+            })?;
+
+        let json_val: Value = serde_json::from_slice(&bytes).context(MetadataParseSnafu {
+            table: table_name.clone(),
+        })?;
+
+        // Patch missing iceberg spec fields
+        let json_val = patch_missing_operation(json_val)?;
+
+        // Convert back to bytes
+        let patched_bytes = serde_json::to_vec(&json_val).context(MetadataParseSnafu {
+            table: table_name.clone(),
+        })?;
+        // Deserialize normally
+        let metadata: TableMetadata =
+            serde_json::from_slice(&patched_bytes).context(MetadataParseSnafu {
+                table: table_name.clone(),
+            })?;
+
+        let stored_table = catalog_metastore::Table {
+            ident: table_ident.clone(),
+            metadata,
+            metadata_location: entry.metadata_location.clone(),
+            properties: HashMap::default(),
+            volume_ident: Some(volume.ident.clone()),
+            volume_location: None,
+            is_temporary: false,
+            format: TableFormat::Iceberg,
+        };
+        metastore
+            .register_table(&table_ident, stored_table)
+            .await
+            .context(MetastoreSnafu)?;
+        Ok(())
+    }
+}
+
+fn patch_missing_operation(mut value: Value) -> Result<Value, ConfigError> {
+    if let Some(snapshots) = value.get_mut("snapshots").and_then(|v| v.as_array_mut()) {
+        for snapshot in snapshots {
+            if let Some(summary) = snapshot.get_mut("summary")
+                && summary.get("operation").is_none()
+            {
+                summary
+                    .as_object_mut()
+                    .context(InvalidMetadataSnafu)?
+                    .insert("operation".to_string(), Value::String("append".into()));
+            }
+        }
+    }
+    Ok(value)
 }