Low level create table

rust/src/action.rs

@@ -362,6 +362,16 @@ pub struct Format {
     options: Option<HashMap<String, String>>,
 }
 
+impl Format {
+    /// Allows creation of a new action::Format
+    pub fn new(provider: String, options: Option<HashMap<String,String>>) -> Self {
+        Self {
+            provider,
+            options
+        }
+    }
+}
+
 /// Action that describes the metadata of the table.
 /// This is a top-level action in Delta log entries.
 #[derive(Serialize, Deserialize, Debug, Default, Clone)]

rust/src/delta.rs

@@ -207,6 +207,30 @@ pub struct DeltaTableMetaData {
     pub configuration: HashMap<String, String>,
 }
 
+impl DeltaTableMetaData {
+    ///
+    pub fn new(
+        name: Option<String>, 
+        description: Option<String>, 
+        schema: Schema, 
+        partition: Vec<String>, 
+        config: HashMap<String, String>
+    ) -> Self {
+        // https://github.com/delta-io/delta/blob/master/core/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala#L350
+
+        Self {
+            id: Uuid::new_v4().to_string(), // create a new GUID
+            name: name,
+            description: description,
+            format: action::Format::new("parquet".to_string(), None),
+            schema: schema,
+            partition_columns: partition,
+            created_time: Utc::now().timestamp_millis(),  // create a timestamp for current timestamp
+            configuration: config
+        }
+    }
+}
+
 impl fmt::Display for DeltaTableMetaData {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(
@@ -1034,6 +1058,34 @@ impl DeltaTable {
         })
     }
 
+    /// Create a DeltaTable with version 0 given the provided MetaData and Protocol
+    pub async fn create(
+        &mut self, 
+        metadata: DeltaTableMetaData, 
+        protocol: action::Protocol
+    ) -> Result<(), DeltaTableError> {
+
+        let meta = action::MetaData::try_from(metadata)?;
+
+        // TODO add commit info action
+        let actions = vec![
+            Action::protocol(protocol), 
+            Action::metaData(meta)
+        ];
+
+        let mut transaction = self.create_transaction(None);
+        transaction.add_actions(actions);
+
+        // Need better error handling here
+        let prepared_commit = transaction.prepare_commit(None).await.unwrap();
+        self.try_commit_transaction(&prepared_commit, 0).await.unwrap();
+
+        // Can we mutate the DeltaTable's state using process_action()
+        // in order to get most up-to-date state based on the commit above
+
+        Ok(())
+    }
+
     /// Time travel Delta table to latest version that's created at or before provided `datetime`
     /// argument.
     ///
@@ -1577,4 +1629,61 @@ mod tests {
             assert!(parquet_filename.contains("col1=a/col2=b/part-00000-"));
         }
     }
+
+    #[tokio::test]
+    async fn test_create_delta_table() {
+
+        let test_schema = Schema::new(
+            "test".to_string(),
+            vec![
+                SchemaField::new(
+                    "Id".to_string(), 
+                    SchemaDataType::primitive("integer".to_string()),
+                    true,
+                    HashMap::new()
+                ),
+                SchemaField::new(
+                    "Name".to_string(), 
+                    SchemaDataType::primitive("string".to_string()),
+                    true,
+                    HashMap::new()
+                )
+            ]
+        );
+
+        let delta_md = DeltaTableMetaData::new(
+            None, 
+            None, 
+            test_schema, 
+            vec![], 
+            HashMap::new()
+        );
+
+        //assert delta_md auto generated values [id, format, created_time] are of the correct type
+
+        let protocol = action::Protocol{
+            min_reader_version: 1,
+            min_writer_version: 2
+        };
+
+        let tmp_dir = tempdir::TempDir::new("create_table_test").unwrap(); 
+        let table_dir = tmp_dir.path().join("test_create");
+
+        let path = table_dir.to_str().unwrap();
+        let backend = Box::new(
+            storage::file::FileStorageBackend::new(tmp_dir.path().to_str().unwrap())
+        );
+
+        let mut dt = DeltaTable::new(
+            path, 
+            backend
+        ).unwrap();
+
+        dt.create(delta_md, protocol).await.unwrap();
+
+        // assert new log file created before deletion
+        // assert DeltaTable version is now 0
+        assert_eq!(dt.version, 0);
+
+    }
 }

rust/src/schema.rs

@@ -44,6 +44,21 @@ pub struct SchemaField {
 }
 
 impl SchemaField {
+
+    pub fn new(
+        name: String, 
+        r#type: SchemaDataType,
+        nullable: bool, 
+        metadata: HashMap<String, String>
+    ) -> Self {
+        Self {
+            name,
+            r#type,
+            nullable,
+            metadata
+        }
+    }
+
     /// The column name of the schema field.
     pub fn get_name(&self) -> &str {
         &self.name
@@ -158,4 +173,11 @@ impl Schema {
     pub fn get_fields(&self) -> &Vec<SchemaField> {
         &self.fields
     }
+
+    pub fn new(r#type: String, fields: Vec<SchemaField>) -> Self {
+        Self {
+            r#type,
+            fields
+        }
+    }
 }