cdc delete

Author: ion-elgreco

crates/core/src/operations/delete.rs

@@ -21,6 +21,7 @@ use core::panic;
 use std::sync::Arc;
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 
+use super::writer::DeltaWriter;
 use crate::logstore::LogStoreRef;
 use datafusion::execution::context::{SessionContext, SessionState};
 use datafusion::physical_plan::filter::FilterExec;
@@ -29,18 +30,21 @@ use datafusion::prelude::Expr;
 use datafusion_common::scalar::ScalarValue;
 use datafusion_common::DFSchema;
 use futures::future::BoxFuture;
+use object_store::prefix::PrefixStore;
 use parquet::file::properties::WriterProperties;
 use serde::Serialize;
 
+use super::cdc::should_write_cdc;
 use super::datafusion_utils::Expression;
 use super::transaction::{CommitBuilder, CommitProperties, PROTOCOL};
-use super::write::WriterStatsConfig;
+use super::write::{write_execution_plan_cdc, write_execution_plan_cdf, WriterStatsConfig};
+use super::writer::WriterConfig;
 use crate::delta_datafusion::expr::fmt_expr_to_sql;
 use crate::delta_datafusion::{
     find_files, register_store, DataFusionMixins, DeltaScanBuilder, DeltaSessionContext,
 };
 use crate::errors::DeltaResult;
-use crate::kernel::{Action, Add, Remove};
+use crate::kernel::{Action, Add, AddCDCFile, Remove};
 use crate::operations::write::write_execution_plan;
 use crate::protocol::DeltaOperation;
 use crate::table::state::DeltaTableState;
@@ -130,7 +134,7 @@ async fn excute_non_empty_expr(
     metrics: &mut DeleteMetrics,
     rewrite: &[Add],
     writer_properties: Option<WriterProperties>,
-) -> DeltaResult<Vec<Add>> {
+) -> DeltaResult<Vec<Action>> {
     // For each identified file perform a parquet scan + filter + limit (1) + count.
     // If returned count is not zero then append the file to be rewritten and removed from the log. Otherwise do nothing to the file.
 
@@ -160,27 +164,53 @@ async fn excute_non_empty_expr(
             .map(|v| v.iter().map(|v| v.to_string()).collect::<Vec<String>>()),
     );
 
-    let add_actions = write_execution_plan(
+    let mut actions: Vec<Action> = write_execution_plan(
         Some(snapshot),
         state.clone(),
         filter.clone(),
         table_partition_cols.clone(),
         log_store.object_store(),
         Some(snapshot.table_config().target_file_size() as usize),
         None,
-        writer_properties,
+        writer_properties.clone(),
         false,
         None,
-        writer_stats_config,
+        writer_stats_config.clone(),
         None,
     )
     .await?
     .into_iter()
     .map(|a| match a {
         Action::Add(a) => a,
         _ => panic!("Expected Add action"),
-    })
-    .collect::<Vec<Add>>();
+    }).into_iter().map(Action::Add).collect();
+
+    // CDC logic, simply filters data with predicate and adds the _change_type="delete" as literal column
+    match should_write_cdc(&snapshot) {
+        Ok(true) => {
+            let cdc_predicate_expr =
+                state.create_physical_expr(expression.clone(), &input_dfschema)?;
+            let cdc_scan: Arc<dyn ExecutionPlan> =
+                Arc::new(FilterExec::try_new(cdc_predicate_expr, scan.clone())?);
+            let cdc_actions = write_execution_plan_cdc(
+                Some(snapshot),
+                state.clone(),
+                cdc_scan.clone(),
+                table_partition_cols.clone(),
+                log_store.object_store(),
+                Some(snapshot.table_config().target_file_size() as usize),
+                None,
+                writer_properties,
+                false,
+                None,
+                writer_stats_config,
+                None,
+            )
+            .await?;
+            actions.extend(cdc_actions)
+        }
+        _ => (),
+    };
 
     let read_records = scan.parquet_scan.metrics().and_then(|m| m.output_rows());
     let filter_records = filter.metrics().and_then(|m| m.output_rows());
@@ -189,7 +219,7 @@ async fn excute_non_empty_expr(
         .zip(filter_records)
         .map(|(read, filter)| read - filter);
 
-    Ok(add_actions)
+    Ok(actions)
 }
 
 async fn execute(
@@ -209,7 +239,7 @@ async fn execute(
 
     let predicate = predicate.unwrap_or(Expr::Literal(ScalarValue::Boolean(Some(true))));
 
-    let add = if candidates.partition_scan {
+    let mut actions = if candidates.partition_scan {
         Vec::new()
     } else {
         let write_start = Instant::now();
@@ -233,7 +263,6 @@ async fn execute(
         .unwrap()
         .as_millis() as i64;
 
-    let mut actions: Vec<Action> = add.into_iter().map(Action::Add).collect();
     metrics.num_removed_files = remove.len();
     metrics.num_added_files = actions.len();
 

crates/core/src/operations/write.rs

@@ -40,6 +40,7 @@ use datafusion_common::DFSchema;
 use datafusion_expr::Expr;
 use futures::future::BoxFuture;
 use futures::StreamExt;
+use object_store::prefix::PrefixStore;
 use parquet::file::properties::WriterProperties;
 use tracing::log::*;
 
@@ -52,7 +53,7 @@ use crate::delta_datafusion::expr::parse_predicate_expression;
 use crate::delta_datafusion::{find_files, register_store, DeltaScanBuilder};
 use crate::delta_datafusion::{DataFusionMixins, DeltaDataChecker};
 use crate::errors::{DeltaResult, DeltaTableError};
-use crate::kernel::{Action, Add, Metadata, PartitionsExt, Remove, StructType};
+use crate::kernel::{Action, Add, AddCDCFile, Metadata, PartitionsExt, Remove, StructType};
 use crate::logstore::LogStoreRef;
 use crate::operations::cast::{cast_record_batch, merge_schema};
 use crate::protocol::{DeltaOperation, SaveMode};
@@ -461,6 +462,54 @@ async fn write_execution_plan_with_predicate(
     Ok(actions)
 }
 
+pub(crate) async fn write_execution_plan_cdc(
+    snapshot: Option<&DeltaTableState>,
+    state: SessionState,
+    plan: Arc<dyn ExecutionPlan>,
+    partition_columns: Vec<String>,
+    object_store: ObjectStoreRef,
+    target_file_size: Option<usize>,
+    write_batch_size: Option<usize>,
+    writer_properties: Option<WriterProperties>,
+    safe_cast: bool,
+    schema_mode: Option<SchemaMode>,
+    writer_stats_config: WriterStatsConfig,
+    sender: Option<Sender<RecordBatch>>,
+) -> DeltaResult<Vec<Action>> {
+    let cdc_store = Arc::new(PrefixStore::new(
+        object_store,
+        "_change_data",
+    ));
+    Ok(write_execution_plan(
+        snapshot, 
+        state, 
+        plan, 
+        partition_columns,
+        cdc_store,
+        target_file_size, 
+        write_batch_size, 
+        writer_properties, 
+        safe_cast, 
+        schema_mode, 
+        writer_stats_config, 
+        sender).await?.into_iter().map(|add| {
+            // Modify add actions into CDC actions
+            match add {
+                Action::Add(add) => { Action::Cdc(AddCDCFile {
+                                        // This is a gnarly hack, but the action needs the nested path, not the
+                                        // path isnide the prefixed store
+                                        path: format!("_change_data/{}", add.path),
+                                        size: add.size,
+                                        partition_values: add.partition_values,
+                                        data_change: false,
+                                        tags: add.tags,
+                                    })},
+                _ => panic!("Expected Add action"),
+            }
+        }).collect::<Vec<_>>())
+}
+
+
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn write_execution_plan(
     snapshot: Option<&DeltaTableState>,