Function execute_cow_delete_once

crates/integrations/datafusion/src/delete.rs:90–123

Single attempt of CoW DELETE execution.

(
    ctx: &SQLContext,
    delete: &Delete,
    table: &Table,
    table_ref: &str,
)

/// Single attempt of CoW DELETE execution.
async fn execute_cow_delete_once(
    ctx: &SQLContext,
    delete: &Delete,
    table: &Table,
    table_ref: &str,
) -> DFResult<DataFrame> {
    let where_str = delete.selection.as_ref().map(|e| e.to_string());
    let partition_set =
        build_partition_set_from_where(ctx, table, table_ref, where_str.as_deref()).await?;

    let mut writer = CopyOnWriteMergeWriter::new(table, vec![], partition_set)
        .await
        .map_err(to_datafusion_error)?;

    let mut temp_tracker = TempTableTracker::new(ctx);
    let (has_data, cow_table_name) =
        register_cow_target_table(ctx, table, &writer, &mut temp_tracker).await?;
    if !has_data {
        return ok_result(ctx.ctx(), 0);
    }

    let cow_target_qualified = cow_table_name;
    let result =
        execute_cow_delete_inner(ctx.ctx(), &cow_target_qualified, delete, &mut writer).await;
    let total_count = result?;

    let messages = writer.prepare_commit().await.map_err(to_datafusion_error)?;
    if !messages.is_empty() {
        let commit = table.new_write_builder().new_commit();
        commit.commit(messages).await.map_err(to_datafusion_error)?;
    }

    ok_result(ctx.ctx(), total_count)
}

async fn execute_cow_delete_inner(
    ctx: &SessionContext,