chore: add deprecation notices for commit logic on DeltaTable

python/src/lib.rs

@@ -19,7 +19,9 @@ use deltalake::delta_datafusion::DeltaDataChecker;
 use deltalake::operations::transaction::commit;
 use deltalake::operations::vacuum::VacuumBuilder;
 use deltalake::partitions::PartitionFilter;
-use deltalake::{DeltaDataTypeLong, DeltaDataTypeTimestamp, DeltaTableMetaData, Invariant, Schema};
+use deltalake::{
+    DeltaConfigKey, DeltaDataTypeLong, DeltaDataTypeTimestamp, DeltaOps, Invariant, Schema,
+};
 use pyo3::create_exception;
 use pyo3::exceptions::PyException;
 use pyo3::exceptions::PyValueError;
@@ -29,6 +31,8 @@ use pyo3::types::PyType;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::convert::TryFrom;
+use std::future::IntoFuture;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
@@ -740,33 +744,45 @@ fn write_new_deltalake(
     configuration: Option<HashMap<String, Option<String>>>,
     storage_options: Option<HashMap<String, String>>,
 ) -> PyResult<()> {
-    let mut table = DeltaTableBuilder::from_uri(table_uri)
+    let table = DeltaTableBuilder::from_uri(table_uri)
         .with_storage_options(storage_options.unwrap_or_default())
         .build()
         .map_err(PyDeltaTableError::from_raw)?;
 
-    let metadata = DeltaTableMetaData::new(
-        name,
-        description,
-        None, // Format
-        (&schema.0)
-            .try_into()
-            .map_err(PyDeltaTableError::from_arrow)?,
-        partition_by,
-        configuration.unwrap_or_default(),
-    );
-
-    let fut = table.create(
-        metadata,
-        action::Protocol {
-            min_reader_version: 1,
-            min_writer_version: 1, // TODO: Make sure we comply with protocol
-        },
-        None, // TODO
-        Some(add_actions.iter().map(|add| add.into()).collect()),
-    );
-
-    rt()?.block_on(fut).map_err(PyDeltaTableError::from_raw)?;
+    let schema: Schema = (&schema.0)
+        .try_into()
+        .map_err(PyDeltaTableError::from_arrow)?;
+
+    let configuration = configuration
+        .unwrap_or_default()
+        .into_iter()
+        .filter_map(|(key, value)| {
+            if let Ok(key) = DeltaConfigKey::from_str(&key) {
+                Some((key, value))
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    let mut builder = DeltaOps(table)
+        .create()
+        .with_columns(schema.get_fields().clone())
+        .with_partition_columns(partition_by)
+        .with_configuration(configuration)
+        .with_actions(add_actions.iter().map(|add| Action::add(add.into())));
+
+    if let Some(name) = &name {
+        builder = builder.with_table_name(name);
+    };
+
+    if let Some(description) = &description {
+        builder = builder.with_comment(description);
+    };
+
+    rt()?
+        .block_on(builder.into_future())
+        .map_err(PyDeltaTableError::from_raw)?;
 
     Ok(())
 }

python/tests/test_writer.py

@@ -144,7 +144,7 @@ def test_roundtrip_metadata(tmp_path: pathlib.Path, sample_data: pa.Table):
         sample_data,
         name="test_name",
         description="test_desc",
-        configuration={"configTest": "foobar"},
+        configuration={"delta.appendOnly": "false"},
     )
 
     delta_table = DeltaTable(tmp_path)
@@ -153,7 +153,7 @@ def test_roundtrip_metadata(tmp_path: pathlib.Path, sample_data: pa.Table):
 
     assert metadata.name == "test_name"
     assert metadata.description == "test_desc"
-    assert metadata.configuration == {"configTest": "foobar"}
+    assert metadata.configuration == {"delta.appendOnly": "false"}
 
 
 @pytest.mark.parametrize(

rust/examples/recordbatch-writer.rs

@@ -8,7 +8,6 @@
  */
 
 use chrono::prelude::*;
-use deltalake::action::*;
 use deltalake::arrow::array::*;
 use deltalake::arrow::record_batch::RecordBatch;
 use deltalake::writer::{DeltaWriter, RecordBatchWriter};
@@ -75,8 +74,8 @@ struct WeatherRecord {
 }
 
 impl WeatherRecord {
-    fn schema() -> Schema {
-        Schema::new(vec![
+    fn columns() -> Vec<SchemaField> {
+        vec![
             SchemaField::new(
                 "timestamp".to_string(),
                 SchemaDataType::primitive("timestamp".to_string()),
@@ -101,7 +100,7 @@ impl WeatherRecord {
                 true,
                 HashMap::new(),
             ),
-        ])
+        ]
     }
 }
 
@@ -189,27 +188,11 @@ fn convert_to_batch(table: &DeltaTable, records: &Vec<WeatherRecord>) -> RecordB
  * Table in an existing directory that doesn't currently contain a Delta table
  */
 async fn create_initialized_table(table_path: &Path) -> DeltaTable {
-    let mut table = DeltaTableBuilder::from_uri(table_path).build().unwrap();
-    let table_schema = WeatherRecord::schema();
-    let mut commit_info = serde_json::Map::<String, serde_json::Value>::new();
-    commit_info.insert(
-        "operation".to_string(),
-        serde_json::Value::String("CREATE TABLE".to_string()),
-    );
-    commit_info.insert(
-        "userName".to_string(),
-        serde_json::Value::String("test user".to_string()),
-    );
-
-    let protocol = Protocol {
-        min_reader_version: 1,
-        min_writer_version: 1,
-    };
-
-    let metadata = DeltaTableMetaData::new(None, None, None, table_schema, vec![], HashMap::new());
-
-    table
-        .create(metadata, protocol, Some(commit_info), None)
+    let table = DeltaOps::try_from_uri(table_path)
+        .await
+        .unwrap()
+        .create()
+        .with_columns(WeatherRecord::columns())
         .await
         .unwrap();
 

rust/src/delta.rs

@@ -1112,6 +1112,10 @@ impl DeltaTable {
     }
 
     /// Vacuum the delta table. See [`VacuumBuilder`] for more information.
+    #[deprecated(
+        since = "0.10.0",
+        note = "use DelaOps from operations module to create a Vacuum operation."
+    )]
     pub async fn vacuum(
         &mut self,
         retention_hours: Option<u64>,
@@ -1133,6 +1137,11 @@ impl DeltaTable {
     /// Creates a new DeltaTransaction for the DeltaTable.
     /// The transaction holds a mutable reference to the DeltaTable, preventing other references
     /// until the transaction is dropped.
+    #[deprecated(
+        since = "0.10.0",
+        note = "use 'commit' function from operations module to commit to Delta table."
+    )]
+    #[allow(deprecated)]
     pub fn create_transaction(
         &mut self,
         options: Option<DeltaTransactionOptions>,
@@ -1145,6 +1154,11 @@ impl DeltaTable {
     /// This is low-level transaction API. If user does not want to maintain the commit loop then
     /// the `DeltaTransaction.commit` is desired to be used as it handles `try_commit_transaction`
     /// with retry logic.
+    #[deprecated(
+        since = "0.10.0",
+        note = "use 'commit' function from operations module to commite to Delta table."
+    )]
+    #[allow(deprecated)]
     pub async fn try_commit_transaction(
         &mut self,
         commit: &PreparedCommit,
@@ -1168,6 +1182,11 @@ impl DeltaTable {
     }
 
     /// Create a DeltaTable with version 0 given the provided MetaData, Protocol, and CommitInfo
+    #[deprecated(
+        since = "0.10.0",
+        note = "use DelaOps from operations module to create a Create operation."
+    )]
+    #[allow(deprecated)]
     pub async fn create(
         &mut self,
         metadata: DeltaTableMetaData,
@@ -1322,12 +1341,17 @@ impl Default for DeltaTransactionOptions {
 /// Please not that in case of non-retryable error the temporary commit file such as
 /// `_delta_log/_commit_<uuid>.json` will orphaned in storage.
 #[derive(Debug)]
+#[deprecated(
+    since = "0.10.0",
+    note = "use 'commit' function from operations module to commit to Delta table."
+)]
 pub struct DeltaTransaction<'a> {
     delta_table: &'a mut DeltaTable,
     actions: Vec<Action>,
     options: DeltaTransactionOptions,
 }
 
+#[allow(deprecated)]
 impl<'a> DeltaTransaction<'a> {
     /// Creates a new delta transaction.
     /// Holds a mutable reference to the delta table to prevent outside mutation while a transaction commit is in progress.
@@ -1377,7 +1401,6 @@ impl<'a> DeltaTransaction<'a> {
         // } else {
         //     IsolationLevel::Serializable
         // };
-
         let prepared_commit = self.prepare_commit(operation, app_metadata).await?;
 
         // try to commit in a loop in case other writers write the next version first
@@ -1430,6 +1453,7 @@ impl<'a> DeltaTransaction<'a> {
         Ok(PreparedCommit { uri: path })
     }
 
+    #[allow(deprecated)]
     async fn try_commit_loop(
         &mut self,
         commit: &PreparedCommit,
@@ -1473,6 +1497,10 @@ impl<'a> DeltaTransaction<'a> {
 /// Holds the uri to prepared commit temporary file created with `DeltaTransaction.prepare_commit`.
 /// Once created, the actual commit could be executed with `DeltaTransaction.try_commit`.
 #[derive(Debug)]
+#[deprecated(
+    since = "0.10.0",
+    note = "use 'commit' function from operations module to commit to Delta table."
+)]
 pub struct PreparedCommit {
     uri: Path,
 }
@@ -1624,6 +1652,7 @@ mod tests {
             serde_json::Value::String("test user".to_string()),
         );
         // Action
+        #[allow(deprecated)]
         dt.create(delta_md.clone(), protocol.clone(), Some(commit_info), None)
             .await
             .unwrap();

rust/src/delta_config.rs

@@ -10,6 +10,7 @@ use crate::DeltaTableError;
 /// Typed property keys that can be defined on a delta table
 /// <https://docs.delta.io/latest/table-properties.html#delta-table-properties-reference>
 /// <https://learn.microsoft.com/en-us/azure/databricks/delta/table-properties>
+#[derive(PartialEq, Eq, Hash)]
 pub enum DeltaConfigKey {
     /// true for this Delta table to be append-only. If append-only,
     /// existing records cannot be deleted, and existing values cannot be updated.

rust/src/lib.rs

@@ -109,6 +109,7 @@ pub mod writer;
 pub use self::builder::*;
 pub use self::data_catalog::{get_data_catalog, DataCatalog, DataCatalogError};
 pub use self::delta::*;
+pub use self::delta_config::*;
 pub use self::partitions::*;
 pub use self::schema::*;
 pub use object_store::{path::Path, Error as ObjectStoreError, ObjectMeta, ObjectStore};

rust/src/operations/create.rs

@@ -125,8 +125,11 @@ impl CreateBuilder {
     }
 
     /// Specify columns to append to schema
-    pub fn with_columns(mut self, columns: impl IntoIterator<Item = SchemaField>) -> Self {
-        self.columns.extend(columns);
+    pub fn with_columns(
+        mut self,
+        columns: impl IntoIterator<Item = impl Into<SchemaField>>,
+    ) -> Self {
+        self.columns.extend(columns.into_iter().map(|c| c.into()));
         self
     }
 
@@ -269,7 +272,7 @@ impl CreateBuilder {
         actions.extend(
             self.actions
                 .into_iter()
-                .filter(|a| matches!(a, Action::protocol(_))),
+                .filter(|a| !matches!(a, Action::protocol(_))),
         );
 
         Ok((table, actions, operation))

rust/src/writer/mod.rs

@@ -128,6 +128,7 @@ pub trait DeltaWriter<T> {
         table: &mut DeltaTable,
     ) -> Result<DeltaDataTypeVersion, DeltaTableError> {
         let mut adds = self.flush().await?;
+        #[allow(deprecated)]
         let mut tx = table.create_transaction(None);
         tx.add_actions(adds.drain(..).map(Action::add).collect());
         let version = tx.commit(None, None).await?;

rust/src/writer/test_utils.rs

@@ -1,3 +1,4 @@
+#![allow(deprecated)]
 //! Utilities for writing unit tests
 use super::*;
 use crate::{