Author a batch transform

A batch transform is a trusted native compute unit in an engine pipeline. It accepts one Arrow IPC stream, emits zero or more Arrow IPC streams, and loads in the dataflow executor as a content-addressed Rust cdylib through the NATIVE_DYLIB tier. Use this path for reviewed platform code. Tenant-supplied WASM uses the separate capability-limited Wasmtime UDF lane, while untrusted native or Python code uses the Firecracker MICROVM lane on opted-in KVM nodes.

The conversation extraction transform is the maintained native example:

• input schema: redgold.ai.conversation_message_gateway.ConversationMessageGateway
• transform: crates/exec/data-transform-batch-conversation-message-extract
• output schema: redgold.ai.conversation_message_extracted.ConversationMessageExtracted
• runnable pipeline: pipelines/jobs/memory-r1-messages.toml
• inline ABI test: crates/exec/data-transform-batch-conversation-message-extract/src/lib.rs

Create one

rac transform new example-rows

This writes crates/exec/data-transform-batch-example-rows/{Cargo.toml,src/lib.rs,pipeline.toml,job.toml}. The generated transform is an identity stage with an inline ABI test. The scaffold test uses placeholder bytes, so replace it with a real Arrow IPC fixture as part of implementing the row conversion. Replace run_transform while retaining the exported transform_batch_abi_version static and transform_batch function. pipeline.toml is the route-less PipelineManifest registered for inspection; job.toml is its directly executable standalone form.

Fill in [package.metadata.batch-transform] with the input and output dataset names, stable Rust schema type identifiers, the checked-in example job, and the pipeline manifest. Move or adapt job.toml under pipelines/jobs/. Its shape is:

[dataflow.source]
kind = "jsonl"
hos_prefix = "raw/example/"

[[dataflow.stages]]
kind = "function"
qualified_name = "data-transform-batch-example-rows"
pin_artifact_hash = "PUBLISH_RESULT_GOES_HERE"

[dataflow.sink]
kind = "hos"
hos_path = "derived/example/{run_id}/example_rows"

Stored and wire schemas use the component's checked-in Rust serde types and CBOR. Reuse the focused schema micro-crate in transform code, and treat Arrow, Lance, SQL, and index layouts as projections of that source type.

The qualified_name identifies the function. The pin_artifact_hash selects the exact published bytes. Keep both in the committed job so an executor cannot silently run a newer build under the same name.

Check and publish

rac transform check crates/exec/data-transform-batch-example-rows
rac run \
  --with-secret DTS_DATABASE_URL \
  --with-secret HETZNER_S3_URL \
  --with-secret HETZNER_S3_ACCESS \
  --with-secret HETZNER_S3_SECRET \
  -- \
  rac transform publish crates/exec/data-transform-batch-example-rows

check validates the package prefix, cdylib target, ABI dependency and version, metadata identity, exported ABI symbols, pipeline reference, and local job reference. It is a structural check; run the crate's inline test to verify the actual Arrow conversion before publication.

publish builds the release library, computes its BLAKE3 digest, uploads it to transforms/batch/<digest>.so in HOS, registers the code metadata, injects the pin into the route-less pipeline text in memory, and publishes the resulting signed PipelineSpec to DES. The command prints one registered transform=... artifact_blake3=... line after all registration steps succeed. Publication requires the four canonical vault items shown above, a configured DATA_ENGINE_URL, and a registered local signing identity created through rac auth keygen. A failed build, upload, or registration returns an error without a usable new pin.

Copy the printed artifact_blake3 value into the committed job's pin_artifact_hash. Executors with HOS artifact fetching enabled retrieve that exact object and verify its BLAKE3 digest before loading it. Re-publishing unchanged bytes returns the same content address.

Run the canonical example

The bounded conversation job carries a checked-in artifact pin:

dataflow-runner --user-id system-backfill submit \
  --manifest pipelines/jobs/memory-r1-messages-validate.toml \
  --run-id conversation-extract-check

The bounded job reads 50 gateway objects, invokes the Rust transform, deduplicates on its data_hash output column, and writes Lance output to HOS. The full memory-r1-messages.toml job uses the same transform and pin; its companion memory-r1-messages-compact.toml performs cross-partition deduplication.

The route-less manifest in the transform crate and the two runnable memory jobs currently pin different published revisions. Treat each checked-in hash as an independent deployment choice; publication does not rewrite any of those files.

The source crate is the implementation reference because its inline test constructs real Arrow IPC input, calls the exported ABI, decodes every emitted batch, and verifies data_hash = blake3(primary_key) for each result.

Before committing a new pin, run the transform crate's focused inline test and validate the two manifest forms:

timeout 300 rac cargo test \
  -p redgold-data-transform-batch-example-rows
rac transform check crates/exec/data-transform-batch-example-rows