Python Connect

datum.connect runs a Datum Connect server and sends Python-built blueprints to it. The transport is Datum-native with protobuf frames, token authentication, demand-tracked Arrow result streams, and negotiated payload formats. It ships over plaintext TCP by default and over encrypted QUIC when you select it (see Transports).

Use it when the process that builds the blueprint should be separate from the process that executes the stream, or when remote Arrow UDF subprocess workers are useful.

Local Server

import datum

server = datum.connect.serve()
session = datum.connect.connect(server.addr, token=server.token)

try:
    graph = (
        datum.Source.range(0, 20)
        .map_multiply(3)
        .filter_greater_than(20)
        .take(5)
        .to_mat(datum.Sink.collect())
    )

    assert session.run(graph) == [21, 24, 27, 30, 33]
finally:
    session.close()
    server.close()

serve() binds to localhost by default and generates a token when you do not pass one. Pass the token to connect(). A bad token is rejected during the handshake, before execution.

Both objects are context-manager friendly:

with datum.connect.serve() as server:
    with datum.connect.connect(server.addr, token=server.token) as session:
        ...

Transports

Connect selects its byte carrier with the transport argument. The session protocol, handshake, wire-format negotiation, cancellation, and terminal semantics are identical across carriers — only the byte transport differs.

Transport	Encryption	Use
"tcp" (default)	none (plaintext)	localhost and trusted same-host links
"quic"	TLS 1.3 over QUIC	cluster-internal / cross-host links needing encryption

# Encrypted QUIC with the localhost dev defaults: the server generates a
# self-signed certificate and the client skips certificate verification.
with datum.connect.serve(transport="quic") as server:
    with datum.connect.connect(server.addr, token=server.token, transport="quic") as session:
        graph = datum.Source.range(0, 20).take(5).to_mat(datum.Sink.collect())
        assert session.run(graph) == [0, 1, 2, 3, 4]

The self-signed certificate and insecure_skip_verify default pair only makes sense for localhost development. For real deployments supply an explicit certificate on the server and verify it on the client:

server = datum.connect.serve(
    transport="quic",
    cert=server_cert_pem,       # PEM-encoded certificate chain
    key=server_key_pem,         # PEM-encoded PKCS#8 private key
    server_names=["datum.internal"],
)
session = datum.connect.connect(
    server.addr,
    token=server.token,
    transport="quic",
    ca_cert=trust_root_pem,     # PEM-encoded trust root; enables verification
    server_name="datum.internal",
)

transport="quic" requires a build of datum-stream compiled with QUIC support. The default published wheel is TCP-only — it does not bundle the QUIC stack (quinn/ring/rustls) so it stays small and cross-compiles cleanly. To use QUIC, build from source with the feature enabled:

maturin develop --features quic   # or: pip install --no-binary datum-stream ... with the feature

Check availability at runtime with datum._native._connect_quic_supported(); requesting transport="quic" on a TCP-only build raises a clear error. QUIC trades localhost throughput for encryption: on loopback it costs roughly 2.7–3.0x wall and 4.6–5.5x CPU versus plaintext TCP at essentially equal memory, so TCP stays the default. See the PY-4 QUIC-vs-TCP benchmark for the full numbers.

Trust Model

Authenticated clients can run arbitrary Python code on the server when a submitted plan contains Python UDFs. That includes explicit map_batches() UDFs and integer-stream map / filter / flat_map callables, which are desugared to Arrow UDF batches before plan submission. This is Spark-style remote execution, not a sandbox.

The token and localhost default are the first trust boundary. UDF subprocesses provide dependency isolation, crash containment, and CPU parallelism across workers; they do not make untrusted code safe. Do not expose a Connect server to clients you would not allow to execute code in that environment.

Remote Arrow UDFs

Connect serializes Arrow-batch Python callables with cloudpickle==3.1.2 and executes them in warm Python subprocess workers. The callable contract for map_batches() is the same as in-process: RecordBatch -> RecordBatch with a declared output schema. Integer-stream map / filter / flat_map callables are shipped as generated RecordBatch -> RecordBatch wrappers and use the same worker pool unchanged.

import datum
import pyarrow as pa
import pyarrow.compute as pc

input_batches = [
    pa.record_batch(
        [
            pa.array([1, 2, 3, 4], type=pa.int64()),
            pa.array([10, 20, 30, 40], type=pa.int64()),
        ],
        names=["left", "right"],
    )
]
schema = pa.schema([("total", pa.int64())])

def add_columns(batch):
    total = pc.add(batch.column(0), batch.column(1))
    return pa.record_batch([total], schema=schema)

graph = (
    datum.Source.from_arrow(input_batches)
    .map_batches(add_columns, output_schema=schema)
    .to_mat(datum.BatchSink.count_rows())
)

with datum.connect.serve(udf_workers=4) as server:
    with datum.connect.connect(server.addr, token=server.token) as session:
        rows = session.run(graph)

The PY-3b benchmark record measured real CPU parallelism for a CPU-bound pure-Python UDF over 1,048,576-byte batches:

Workers	Wall ms	Server+worker CPU ms	Rows/s	Speedup vs 1 worker
1	3,113.105	3,160.000	421,033.059	baseline
2	1,580.738	3,110.000	829,182.187	1.97x
4	832.890	3,200.000	1,573,700.482	3.74x

The same record shows the original remote UDF crossing cost is much higher than in-process: connect_udf_identity added 2,614.778 us/batch at 1,048,576 batch bytes and 49,257.559 us/batch at 16,777,216 batch bytes; connect_udf_compute added 2,454.034 us/batch and 62,586.366 us/batch for the same batch sizes. That cost is pipe IPC, Arrow IPC encode/decode, and Python worker object construction.

The published datum-stream wheel does not compile the shared-memory dependency or transport. To make the PY-5 path available, build from source with the default-off udf-shm Cargo feature:

cd crates/datum-py
maturin develop --release --locked --features udf-shm

Then enable shared memory at runtime before starting the server:

export DATUM_CONNECT_UDF_SHM=on

The Cargo feature is the build-time capability gate; DATUM_CONNECT_UDF_SHM remains the secondary runtime switch. Without udf-shm, setting the environment variable has no effect and workers stay on pipes. With both gates enabled, the parent writes Arrow IPC into a private mmap-backed /dev/shm region. The Python worker reads it with pyarrow.memory_map; large results use a second parent-owned region, while results below 256 KiB remain inline. Inputs below 1 MiB stay on pipes. The pool bounds outstanding regions to 512 MiB; worker writes are checked before every mapped-region write, so an expanding UDF retries inline without growing its pre-sized output file past the reservation. The pool removes request segments after success or worker death, removes its private directory at shutdown, and reaps stale directories left by crashed server processes. If /dev/shm is unavailable, mapping fails, or the budget is full, execution automatically falls back to the pipe path.

This path is build-time and runtime opt-in because the PY-5 A/B found a crossover between 512 KiB and 1 MiB and two negative required rows at eight workers. At 1 MiB it delivered 1.07–1.13x pipe wall throughput through four workers; at 16 MiB it delivered 1.15–1.16x through four workers and reduced server+worker CPU and RSS on every measured worker count. Pipes therefore remain the regression-free default.

Wire Formats

Connect negotiates one of:

Format	Use
arrow-ipc	Default for localhost and same-host sessions
arrow-ipc-lz4	Supported negotiated format, not a current default
arrow-ipc-zstd	Default for remote or constrained links
parquet	Explicit bulk/cold/file-interchange option

The defaults are measurement-based. In the PY-3c wire-format A/B, arrow-ipc had the best normalized localhost throughput: 15,330,445 rows/s at 1,048,576 batch bytes and 10,020,592 rows/s at 16,777,216 batch bytes. On the verified 100 Mbit/s profile, arrow-ipc-zstd won rows/s at both sizes: 1,624,756 rows/s at 1,048,576 batch bytes and 1,304,192 rows/s at 16,777,216 batch bytes.

Parquet remains explicit rather than default. In the 1,048,576-byte constrained row it was smaller on the wire at 4,154,055 bytes, but end-to-end throughput was 295,236 rows/s; arrow-ipc-zstd used 10,106,175 wire bytes and delivered 1,624,756 rows/s. At 16,777,216 bytes, arrow-ipc-zstd was both smaller and faster on the measured generated-int workload: 10,836,225 wire bytes and 1,304,192 rows/s versus Parquet's 13,452,420 wire bytes and 841,893 rows/s.

Override the negotiated default per run when needed:

session.run(graph, wire_format="parquet")

The override must be among the formats negotiated during the handshake.

Streaming Control

session.run(graph) submits, demands the whole result, and collects it. For manual streaming, use start():

execution = session.start(graph, initial_demand=1)
first = execution.next_batch()
execution.cancel()

Cancellation and disconnect cleanup are part of the current test surface.

Current Limits

Connect ships over plaintext TCP (default) or encrypted QUIC (transport="quic"); see Transports. Plaintext TCP is not encrypted — do not expose a TCP Connect server over an untrusted network. QUIC's localhost dev default (self-signed certificate + skipped verification) authenticates nothing; configure an explicit certificate and trust root for deployments that need real peer authentication.

Connect currently accepts Python RunnableGraph, GraphRunnableGraph, and BatchRunnableGraph plans. The GraphDSL connect subset covers integer FlowShape graphs over Broadcast, Balance, Merge, and typed PartitionStrategy.MODULO partitions. Both Sink.fold_sum() and Sink.fold_product() round-trip as typed sink plan values. Local in-process Python GraphDSL also exposes Zip, Concat, and Interleave.