feat(datafusion): Add DDL support with PRIMARY KEY constraint syntax (#237)

- Add PaimonDdlHandler for CREATE TABLE, ALTER TABLE (ADD/DROP/RENAME COLUMN, RENAME TABLE)
- Add PaimonTableFactory for CREATE EXTERNAL TABLE via DataFusion TableProviderFactory
- Extend PaimonCatalogProvider/SchemaProvider with CREATE/DROP SCHEMA and DROP TABLE
- Add arrow_to_paimon_type and arrow_fields_to_paimon to paimon::arrow for Arrow-to-Paimon type conversion
- Support PRIMARY KEY (col, ...) constraint syntax in CREATE TABLE DDL

Author: JingsongLi

crates/integrations/datafusion/src/catalog.rs

@@ -18,6 +18,7 @@
 //! Paimon catalog integration for DataFusion.
 
 use std::any::Any;
+use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::Arc;
 
@@ -86,6 +87,50 @@ impl CatalogProvider for PaimonCatalogProvider {
             "paimon catalog access thread panicked",
         )
     }
+
+    fn register_schema(
+        &self,
+        name: &str,
+        _schema: Arc<dyn SchemaProvider>,
+    ) -> DFResult<Option<Arc<dyn SchemaProvider>>> {
+        let catalog = Arc::clone(&self.catalog);
+        let name = name.to_string();
+        block_on_with_runtime(
+            async move {
+                catalog
+                    .create_database(&name, false, HashMap::new())
+                    .await
+                    .map_err(to_datafusion_error)?;
+                Ok(Some(
+                    Arc::new(PaimonSchemaProvider::new(Arc::clone(&catalog), name))
+                        as Arc<dyn SchemaProvider>,
+                ))
+            },
+            "paimon catalog access thread panicked",
+        )
+    }
+
+    fn deregister_schema(
+        &self,
+        name: &str,
+        cascade: bool,
+    ) -> DFResult<Option<Arc<dyn SchemaProvider>>> {
+        let catalog = Arc::clone(&self.catalog);
+        let name = name.to_string();
+        block_on_with_runtime(
+            async move {
+                catalog
+                    .drop_database(&name, false, cascade)
+                    .await
+                    .map_err(to_datafusion_error)?;
+                Ok(Some(
+                    Arc::new(PaimonSchemaProvider::new(Arc::clone(&catalog), name))
+                        as Arc<dyn SchemaProvider>,
+                ))
+            },
+            "paimon catalog access thread panicked",
+        )
+    }
 }
 
 /// Represents a [`SchemaProvider`] for the Paimon [`Catalog`], managing
@@ -159,4 +204,36 @@ impl SchemaProvider for PaimonSchemaProvider {
             "paimon catalog access thread panicked",
         )
     }
+
+    fn register_table(
+        &self,
+        _name: String,
+        table: Arc<dyn TableProvider>,
+    ) -> DFResult<Option<Arc<dyn TableProvider>>> {
+        // DataFusion calls register_table after table creation, so we just
+        // acknowledge it here.
+        Ok(Some(table))
+    }
+
+    fn deregister_table(&self, name: &str) -> DFResult<Option<Arc<dyn TableProvider>>> {
+        let catalog = Arc::clone(&self.catalog);
+        let identifier = Identifier::new(self.database.clone(), name);
+        block_on_with_runtime(
+            async move {
+                // Try to get the table first so we can return it.
+                let table = match catalog.get_table(&identifier).await {
+                    Ok(t) => t,
+                    Err(paimon::Error::TableNotExist { .. }) => return Ok(None),
+                    Err(e) => return Err(to_datafusion_error(e)),
+                };
+                let provider = PaimonTableProvider::try_new(table)?;
+                catalog
+                    .drop_table(&identifier, false)
+                    .await
+                    .map_err(to_datafusion_error)?;
+                Ok(Some(Arc::new(provider) as Arc<dyn TableProvider>))
+            },
+            "paimon catalog access thread panicked",
+        )
+    }
 }