feat(python): expose delete operation

Naively expose the delete operation, with the option to provide a
predicate.

python/deltalake/table.py

@@ -199,9 +199,9 @@ def _filters_to_expression(filters: FilterType) -> Expression:
 Predicates are expressed in disjunctive normal form (DNF), like [("x", "=", "a"), ...].
 DNF allows arbitrary boolean logical combinations of single partition predicates.
 The innermost tuples each describe a single partition predicate. The list of inner
-predicates is interpreted as a conjunction (AND), forming a more selective and 
-multiple partition predicates. Each tuple has format: (key, op, value) and compares 
-the key with the value. The supported op are: `=`, `!=`, `in`, and `not in`. If 
+predicates is interpreted as a conjunction (AND), forming a more selective and
+multiple partition predicates. Each tuple has format: (key, op, value) and compares
+the key with the value. The supported op are: `=`, `!=`, `in`, and `not in`. If
 the op is in or not in, the value must be a collection such as a list, a set or a tuple.
 The supported type for value is str. Use empty string `''` for Null partition value.
 
@@ -302,13 +302,13 @@ def files(
 
     files.__doc__ = f"""
 Get the .parquet files of the DeltaTable.
-    
+
 The paths are as they are saved in the delta log, which may either be
 relative to the table root or absolute URIs.
 
-:param partition_filters: the partition filters that will be used for 
+:param partition_filters: the partition filters that will be used for
     getting the matched files
-:return: list of the .parquet files referenced for the current version 
+:return: list of the .parquet files referenced for the current version
     of the DeltaTable
 {_DNF_filter_doc}
     """
@@ -666,6 +666,20 @@ def get_add_actions(self, flatten: bool = False) -> pyarrow.RecordBatch:
         """
         return self._table.get_add_actions(flatten)
 
+    def delete(self, predicate: Optional[str] = None) -> Dict[str, Any]:
+        """Delete records from a Delta Table that statisfy a predicate.
+
+        When a predicate is not provided then all records are deleted from the Delta
+        Table. Otherwise a scan of the Delta table is performed to mark any files
+        that contain records that satisfy the predicate. Once files are determined
+        they are rewritten without the records.
+
+        :param predicate: a logical expression, defaults to None
+        :return: the metrics from delete
+        """
+        metrics = self._table.delete(predicate)
+        return json.loads(metrics)
+
 
 class TableOptimizer:
     """API for various table optimization commands."""

python/src/lib.rs

@@ -21,6 +21,7 @@ use deltalake::checkpoints::create_checkpoint;
 use deltalake::datafusion::prelude::SessionContext;
 use deltalake::delta_datafusion::DeltaDataChecker;
 use deltalake::errors::DeltaTableError;
+use deltalake::operations::delete::DeleteBuilder;
 use deltalake::operations::optimize::{OptimizeBuilder, OptimizeType};
 use deltalake::operations::restore::RestoreBuilder;
 use deltalake::operations::transaction::commit;
@@ -594,6 +595,20 @@ impl RawDeltaTable {
                 .map_err(PythonError::from)?,
         ))
     }
+
+    /// Run the delete command on the delta table: delete records following a predicate and return the delete metrics.
+    #[pyo3(signature = (predicate = None))]
+    pub fn delete(&mut self, predicate: Option<String>) -> PyResult<String> {
+        let mut cmd = DeleteBuilder::new(self._table.object_store(), self._table.state.clone());
+        if let Some(predicate) = predicate {
+            cmd = cmd.with_predicate(predicate);
+        }
+        let (table, metrics) = rt()?
+            .block_on(cmd.into_future())
+            .map_err(PythonError::from)?;
+        self._table.state = table.state;
+        Ok(serde_json::to_string(&metrics).unwrap())
+    }
 }
 
 fn convert_partition_filters<'a>(

rust/src/operations/delete.rs

@@ -30,6 +30,7 @@ use datafusion_common::scalar::ScalarValue;
 use datafusion_common::DFSchema;
 use futures::future::BoxFuture;
 use parquet::file::properties::WriterProperties;
+use serde::Serialize;
 use serde_json::Map;
 use serde_json::Value;
 
@@ -62,7 +63,7 @@ pub struct DeleteBuilder {
     app_metadata: Option<Map<String, serde_json::Value>>,
 }
 
-#[derive(Default, Debug)]
+#[derive(Default, Debug, Serialize)]
 /// Metrics for the Delete Operation
 pub struct DeleteMetrics {
     /// Number of files added
@@ -115,7 +116,7 @@ impl DeleteBuilder {
         self
     }
 
-    /// Writer properties passed to parquet writer for when fiiles are rewritten
+    /// Writer properties passed to parquet writer for when files are rewritten
     pub fn with_writer_properties(mut self, writer_properties: WriterProperties) -> Self {
         self.writer_properties = Some(writer_properties);
         self