Merge branch 'main' into feature/implement-project-node-for-insert-into-datafusion

Author: fvaleye

.github/workflows/ci_typos.yml

@@ -42,4 +42,4 @@ jobs:
     steps:
       - uses: actions/checkout@v5
       - name: Check typos
-        uses: crate-ci/typos@v1.36.3
+        uses: crate-ci/typos@v1.37.2

Cargo.lock

@@ -49,7 +49,7 @@ include = [
 ignore = ["F403", "F405"]
 
 [tool.hatch.envs.dev]
-dependencies = ["maturin>=1.0,<2.0", "pytest>=8.3.2", "datafusion==45.*", "pyiceberg[sql-sqlite,pyarrow]>=0.10.0", "fastavro>=1.11.1"]
+dependencies = ["maturin>=1.0,<2.0", "pytest>=8.3.2", "datafusion==45.*", "pyiceberg[sql-sqlite,pyarrow]>=0.10.0", "fastavro>=1.11.1", "pydantic<2.12.0"]
 
 [tool.hatch.envs.dev.scripts]
 build = "maturin build --out dist --sdist"

bindings/python/pyproject.toml

@@ -37,7 +37,7 @@ const OWNER: &str = "owner";
 const COMMENT: &str = "comment";
 /// hive metatore `location` property
 const LOCATION: &str = "location";
-/// hive metatore `metadat_location` property
+/// hive metatore `metadata_location` property
 const METADATA_LOCATION: &str = "metadata_location";
 /// hive metatore `external` property
 const EXTERNAL: &str = "EXTERNAL";

crates/catalog/hms/src/utils.rs

@@ -765,13 +765,30 @@ impl Catalog for SqlCatalog {
 
     async fn register_table(
         &self,
-        _table_ident: &TableIdent,
-        _metadata_location: String,
+        table_ident: &TableIdent,
+        metadata_location: String,
     ) -> Result<Table> {
-        Err(Error::new(
-            ErrorKind::FeatureUnsupported,
-            "Registering a table is not supported yet",
-        ))
+        if self.table_exists(table_ident).await? {
+            return table_already_exists_err(table_ident);
+        }
+
+        let metadata = TableMetadata::read_from(&self.fileio, &metadata_location).await?;
+
+        let namespace = table_ident.namespace();
+        let tbl_name = table_ident.name().to_string();
+
+        self.execute(&format!(
+            "INSERT INTO {CATALOG_TABLE_NAME}
+             ({CATALOG_FIELD_CATALOG_NAME}, {CATALOG_FIELD_TABLE_NAMESPACE}, {CATALOG_FIELD_TABLE_NAME}, {CATALOG_FIELD_METADATA_LOCATION_PROP}, {CATALOG_FIELD_RECORD_TYPE})
+             VALUES (?, ?, ?, ?, ?)
+            "), vec![Some(&self.name), Some(&namespace.join(".")), Some(&tbl_name), Some(&metadata_location), Some(CATALOG_FIELD_TABLE_RECORD_TYPE)], None).await?;
+
+        Ok(Table::builder()
+            .identifier(table_ident.clone())
+            .metadata_location(metadata_location)
+            .metadata(metadata)
+            .file_io(self.fileio.clone())
+            .build()?)
     }
 
     async fn update_table(&self, _commit: TableCommit) -> Result<Table> {
@@ -1908,4 +1925,63 @@ mod tests {
             "Unexpected => No such table: TableIdent { namespace: NamespaceIdent([\"a\"]), name: \"tbl1\" }"
         );
     }
+
+    #[tokio::test]
+    async fn test_register_table_throws_error_if_table_with_same_name_already_exists() {
+        let warehouse_loc = temp_path();
+        let catalog = new_sql_catalog(warehouse_loc.clone()).await;
+        let namespace_ident = NamespaceIdent::new("a".into());
+        create_namespace(&catalog, &namespace_ident).await;
+        let table_name = "tbl1";
+        let table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
+        create_table(&catalog, &table_ident).await;
+
+        assert_eq!(
+            catalog
+                .register_table(&table_ident, warehouse_loc)
+                .await
+                .unwrap_err()
+                .to_string(),
+            format!("Unexpected => Table {:?} already exists.", &table_ident)
+        );
+    }
+
+    #[tokio::test]
+    async fn test_register_table() {
+        let warehouse_loc = temp_path();
+        let catalog = new_sql_catalog(warehouse_loc.clone()).await;
+        let namespace_ident = NamespaceIdent::new("a".into());
+        create_namespace(&catalog, &namespace_ident).await;
+
+        let table_name = "abc";
+        let location = warehouse_loc.clone();
+        let table_creation = TableCreation::builder()
+            .name(table_name.into())
+            .location(location.clone())
+            .schema(simple_table_schema())
+            .build();
+
+        let table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
+        let expected_table = catalog
+            .create_table(&namespace_ident, table_creation)
+            .await
+            .unwrap();
+
+        let metadata_location = expected_table
+            .metadata_location()
+            .expect("Expected metadata location to be set")
+            .to_string();
+
+        assert_table_eq(&expected_table, &table_ident, &simple_table_schema());
+
+        let _ = catalog.drop_table(&table_ident).await;
+
+        let table = catalog
+            .register_table(&table_ident, metadata_location.clone())
+            .await
+            .unwrap();
+
+        assert_eq!(table.identifier(), expected_table.identifier());
+        assert_eq!(table.metadata_location(), Some(metadata_location.as_str()));
+    }
 }

crates/catalog/sql/src/catalog.rs

@@ -26,7 +26,7 @@ use serde_with::{DeserializeFromStr, SerializeDisplay};
 use super::_serde::DataFileSerde;
 use super::{Datum, FormatVersion, Schema, data_file_schema_v1, data_file_schema_v2};
 use crate::error::Result;
-use crate::spec::{Struct, StructType};
+use crate::spec::{DEFAULT_PARTITION_SPEC_ID, Struct, StructType};
 use crate::{Error, ErrorKind};
 
 /// Data file carries data file path, partition tuple, metrics, …
@@ -49,6 +49,7 @@ pub struct DataFile {
     ///
     /// Partition data tuple, schema based on the partition spec output using
     /// partition field ids for the struct field ids
+    #[builder(default = "Struct::empty()")]
     pub(crate) partition: Struct,
     /// field id: 103
     ///
@@ -156,6 +157,7 @@ pub struct DataFile {
     pub(crate) first_row_id: Option<i64>,
     /// This field is not included in spec. It is just store in memory representation used
     /// in process.
+    #[builder(default = "DEFAULT_PARTITION_SPEC_ID")]
     pub(crate) partition_spec_id: i32,
     /// field id: 143
     ///

crates/iceberg/src/spec/manifest/data_file.rs

@@ -155,7 +155,9 @@ impl PartitionSpec {
         true
     }
 
-    pub(crate) fn partition_to_path(&self, data: &Struct, schema: SchemaRef) -> String {
+    /// Returns partition path string containing partition type and partition
+    /// value as key-value pairs.
+    pub fn partition_to_path(&self, data: &Struct, schema: SchemaRef) -> String {
         let partition_type = self.partition_type(&schema).unwrap();
         let field_types = partition_type.fields();
 
@@ -194,6 +196,15 @@ impl PartitionKey {
         Self { spec, schema, data }
     }
 
+    /// Creates a new partition key from another partition key, with a new data field.
+    pub fn copy_with_data(&self, data: Struct) -> Self {
+        Self {
+            spec: self.spec.clone(),
+            schema: self.schema.clone(),
+            data,
+        }
+    }
+
     /// Generates a partition path based on the partition values.
     pub fn to_path(&self) -> String {
         self.spec.partition_to_path(&self.data, self.schema.clone())
@@ -207,6 +218,21 @@ impl PartitionKey {
             Some(pk) => pk.spec.is_unpartitioned(),
         }
     }
+
+    /// Returns the associated [`PartitionSpec`].
+    pub fn spec(&self) -> &PartitionSpec {
+        &self.spec
+    }
+
+    /// Returns the associated [`SchemaRef`].
+    pub fn schema(&self) -> &SchemaRef {
+        &self.schema
+    }
+
+    /// Returns the associated [`Struct`].
+    pub fn data(&self) -> &Struct {
+        &self.data
+    }
 }
 
 /// Reference to [`UnboundPartitionSpec`].
@@ -1789,6 +1815,9 @@ mod tests {
             .with_fields(vec![
                 NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
                 NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)).into(),
+                NestedField::required(3, "timestamp", Type::Primitive(PrimitiveType::Timestamp))
+                    .into(),
+                NestedField::required(4, "empty", Type::Primitive(PrimitiveType::String)).into(),
             ])
             .build()
             .unwrap();
@@ -1798,14 +1827,23 @@ mod tests {
             .unwrap()
             .add_partition_field("name", "name", Transform::Identity)
             .unwrap()
+            .add_partition_field("timestamp", "ts_hour", Transform::Hour)
+            .unwrap()
+            .add_partition_field("empty", "empty_void", Transform::Void)
+            .unwrap()
             .build()
             .unwrap();
 
-        let data = Struct::from_iter([Some(Literal::int(42)), Some(Literal::string("alice"))]);
+        let data = Struct::from_iter([
+            Some(Literal::int(42)),
+            Some(Literal::string("alice")),
+            Some(Literal::int(1000)),
+            Some(Literal::string("empty")),
+        ]);
 
         assert_eq!(
             spec.partition_to_path(&data, schema.into()),
-            "id=42/name=alice"
+            "id=42/name=alice/ts_hour=1000/empty_void=null"
         );
     }
 }

crates/iceberg/src/spec/partition.rs

@@ -137,19 +137,17 @@ pub enum Transform {
 impl Transform {
     /// Returns a human-readable String representation of a transformed value.
     pub fn to_human_string(&self, field_type: &Type, value: Option<&Literal>) -> String {
-        if let Some(value) = value {
-            if let Some(value) = value.as_primitive_literal() {
-                let field_type = field_type.as_primitive_type().unwrap();
-                let datum = Datum::new(field_type.clone(), value);
-                match self {
-                    Self::Identity => datum.to_human_string(),
-                    Self::Void => "null".to_string(),
-                    _ => {
-                        todo!()
-                    }
-                }
-            } else {
-                "null".to_string()
+        let Some(value) = value else {
+            return "null".to_string();
+        };
+
+        if let Some(value) = value.as_primitive_literal() {
+            let field_type = field_type.as_primitive_type().unwrap();
+            let datum = Datum::new(field_type.clone(), value);
+
+            match self {
+                Self::Void => "null".to_string(),
+                _ => datum.to_human_string(),
             }
         } else {
             "null".to_string()

crates/iceberg/src/spec/transform.rs

@@ -1271,7 +1271,7 @@ impl Datum {
     ///
     /// For string literals, this returns the raw string value without quotes.
     /// For all other literals, it falls back to [`to_string()`].
-    pub(crate) fn to_human_string(&self) -> String {
+    pub fn to_human_string(&self) -> String {
         match self.literal() {
             PrimitiveLiteral::String(s) => s.to_string(),
             _ => self.to_string(),