Function build_insert_batches

crates/integrations/datafusion/src/merge_into.rs:829–856

Build insert batches from not-matched rows, applying INSERT clause projections and predicates.

(
    ctx: &SQLContext,
    not_matched_batches: &[RecordBatch],
    inserts: &[MergeInsertClause],
    s_alias: &str,
    injected_columns: &[String],
    table_fields: &[String],
    temp_tracker: &

/// Build insert batches from not-matched rows, applying INSERT clause projections and predicates.
async fn build_insert_batches(
    ctx: &SQLContext,
    not_matched_batches: &[RecordBatch],
    inserts: &[MergeInsertClause],
    s_alias: &str,
    injected_columns: &[String],
    table_fields: &[String],
    temp_tracker: &mut TempTableTracker<'_>,
) -> DFResult<Vec<RecordBatch>> {
    if not_matched_batches.is_empty() || not_matched_batches.iter().all(|b| b.num_rows() == 0) {
        return Ok(Vec::new());
    }

    // Strip injected columns (_ROW_ID, __upd_*) — keep only source columns
    let source_batches = strip_non_source_columns(not_matched_batches, injected_columns)?;

    // Register as temp table for SQL-based projection/filtering
    let first_schema = source_batches[0].schema();
    let tmp_name = next_cow_table_name("__merge_not_matched");

    let mem_table = datafusion::datasource::MemTable::try_new(first_schema, vec![source_batches])?;
    ctx.register_temp_table(&tmp_name, Arc::new(mem_table))?;
    temp_tracker.register(&tmp_name);

    let result = build_insert_batches_inner(ctx, inserts, s_alias, &tmp_name, table_fields).await;

    result
}

/// Execute INSERT clause queries against the registered temp table.
async fn build_insert_batches_inner(