Function transfer

nodedb/src/control/server/pgwire/ddl/transfer.rs:28–85

Handle `SELECT TRANSFER(collection, source_key, dest_key, field, amount)`

(
    state: &SharedState,
    identity: &AuthenticatedIdentity,
    sql: &str,
)

/// Handle `SELECT TRANSFER(collection, source_key, dest_key, field, amount)`
pub async fn transfer(
    state: &SharedState,
    identity: &AuthenticatedIdentity,
    sql: &str,
) -> PgWireResult<Vec<Response>> {
    let args = super::kv_atomic::parse_function_args(sql, "TRANSFER")?;
    if args.len() < 5 {
        return Err(sqlstate_error(
            "42601",
            "TRANSFER requires 5 arguments: (collection, source_key, dest_key, field, amount)",
        ));
    }

    let collection = unquote(&args[0]).to_lowercase();
    let source_key = unquote(&args[1]);
    let dest_key = unquote(&args[2]);
    let field = unquote(&args[3]);
    let amount_str = args[4].trim().to_string();
    let amount: f64 = amount_str.parse().map_err(|_| {
        sqlstate_error(
            "42601",
            &format!("TRANSFER: amount must be a number, got '{amount_str}'"),
        )
    })?;

    if amount <= 0.0 {
        return Err(sqlstate_error("42601", "TRANSFER: amount must be positive"));
    }

    let tenant_id = identity.tenant_id;
    let vshard = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &collection);

    // Dispatch to Data Plane — entire read+validate+write is atomic (single TPC core).
    let plan = PhysicalPlan::Kv(KvOp::Transfer {
        collection,
        source_key: source_key.into_bytes(),
        dest_key: dest_key.into_bytes(),
        field,
        amount,
    });

    match crate::control::server::dispatch_utils::dispatch_to_data_plane(
        state,
        tenant_id,
        vshard,
        plan,
        TraceId::ZERO,
    )
    .await
    {
        Ok(resp) => {
            let payload_text =
                crate::data::executor::response_codec::decode_payload_to_json(&resp.payload);
            respond_json("transfer", &payload_text)
        }
        Err(e) => Err(sqlstate_error("XX000", &e.to_string())),
    }
}