Function execute_cow_delete_inner

crates/integrations/datafusion/src/delete.rs:125–159

(
    ctx: &SessionContext,
    cow_table_name: &str,
    delete: &Delete,
    writer: &mut CopyOnWriteMergeWriter,
)

}

async fn execute_cow_delete_inner(
    ctx: &SessionContext,
    cow_table_name: &str,
    delete: &Delete,
    writer: &mut CopyOnWriteMergeWriter,
) -> DFResult<u64> {
    let where_clause = match &delete.selection {
        Some(expr) => format!(" WHERE {expr}"),
        None => String::new(),
    };

    // Safety: where_clause comes from sqlparser AST to_string(), not raw user input.
    let query_sql = format!(
        "SELECT \"__paimon_file_idx\", \"__paimon_row_offset\" FROM {cow_table_name}{where_clause}"
    );
    let batches = ctx.sql(&query_sql).await?.collect().await?;

    let mut total_count: u64 = 0;
    for batch in &batches {
        if batch.num_rows() == 0 {
            continue;
        }

        let (file_idx_col, row_offset_col) = extract_tracking_columns(batch)?;

        for row in 0..batch.num_rows() {
            let file_idx = file_idx_col.value(row) as usize;
            let row_offset = row_offset_col.value(row) as usize;
            writer.add_matched_delete(file_idx, row_offset);
            total_count += 1;
        }
    }

    Ok(total_count)
}