Streaming bulk inserts

About

To insert many rows into a SQL database in one connection, the SQL connectors offer executemany — you pass the full set of rows as a rows list and the connector runs the statement once per row. This is convenient, but the whole dataset lives in memory at once: the rows list is fully materialized, and with drivers that pre-bind parameters (for example MSSQL fast_executemany) the whole parameter array is bound up front as well. For a very large transfer — hundreds of thousands or millions of rows — that peak can exhaust the workspace's memory and the execution is killed (out of memory).

The executemany_from_file mode removes that ceiling. Instead of holding the whole dataset in memory, it streams rows from a staged Cloudomation file and inserts them in bounded batches, so peak memory is set by the batch size, not by the total row count. It is available on all four SQL connectors — PostgreSQL, MySQL, MSSQL and Oracle — and returns None, just like executemany.

tip

Reach for executemany_from_file whenever the dataset is large or its size is unbounded/unknown (a nightly export, a full-table sync). For a small, known set of rows, plain executemany is simpler and fine.

executemany vs executemany_from_file

	executemany	executemany_from_file
Where the rows come from	a rows list passed inline	a staged Cloudomation file (NDJSON)
Peak memory	grows with the total number of rows	bounded by batch_size rows
Insert shape	one statement per row, all in one go	one statement per row, flushed in fixed-size batches
Best for	small/known row sets	large or unbounded transfers

The pattern

Using executemany_from_file is two steps:

1. Stage the rows into a Cloudomation file as newline-delimited JSON (NDJSON), writing it in chunks so staging itself never holds the whole dataset in memory.
2. Stream-insert from that file with executemany_from_file, choosing a batch_size that bounds peak memory.

import json

def handler(system, this, inputs):
    file_name = 'people-to-load.ndjson'
    ndjson = system.file(file_name)

    # 1) Stage rows as NDJSON with bounded memory: append in chunks, never
    #    building the whole file in memory. Each line is ONE row — the same
    #    shape as a single `rows` entry of `executemany`.
    buffer = []
    for person in read_source_rows():          # your streaming source
        buffer.append(json.dumps([person['name'], person['age']]))
        if len(buffer) >= 5000:                # flush every 5000 lines
            ndjson.append_bytes_chunk(bytes_=('\n'.join(buffer) + '\n').encode())
            buffer = []
    if buffer:
        ndjson.append_bytes_chunk(bytes_=('\n'.join(buffer) + '\n').encode())

    # 2) Stream-insert in bounded batches.
    this.connect(
        'my-postgres',
        name='bulk insert people',
        mode={
            'mode_name': 'executemany_from_file',
            'query': 'insert into people (name, age) values ($1, $2)',
            'file': file_name,
            'batch_size': 10000,
        },
    )

    # 3) Clean up the staged file.
    ndjson.delete(permanently=True)

Row format (NDJSON)

The file must be newline-delimited JSON: one JSON value per line, each line being exactly one row — the same shape as a single entry of an executemany rows list. Blank lines are ignored.

The bind-parameter shape follows the connector, matching the placeholders you use in the query:

Connector	Query placeholders	One line is
PostgreSQL	$1, $2, …	a JSON array of positional parameters
MySQL	%s	a JSON array of positional parameters
MSSQL	?	a JSON array of positional parameters
Oracle	:name	a JSON object of named parameters

For example, an insert of (name, age) rows looks like this for the positional connectors:

["Ada", 36]
["Alan", 41]
["Grace", 45]

and like this for Oracle (named parameters, keyed to the :name placeholders in the query):

{"name": "Ada", "age": 36}
{"name": "Alan", "age": 41}
{"name": "Grace", "age": 45}

Staging the file with bounded memory

The key to a memory-safe transfer is that you never build the whole file (or the whole rows list) in memory. system.file(...).append_bytes_chunk(bytes_=...) appends to a file without loading its existing content, and creates the file on the first call. Serialize a manageable number of rows to NDJSON, append that chunk, and repeat — so staging holds at most one chunk at a time.

note

The rows do not have to come from memory at all. Any streaming source works — a fetch-style read from another database, a downloaded file read in chunks, an API paged over — as long as you append each batch of lines and drop it before reading the next.

Inserting in bounded batches

executemany_from_file takes these inputs:

Input	Meaning
mode_name	executemany_from_file
query	the SQL statement to run once per row, with the connector's placeholders
file	the name of the Cloudomation file holding the NDJSON rows
batch_size	rows to insert per batch (default 10000). Must be a positive integer; it bounds peak memory.

The connector reads the file as a stream, decodes it one line at a time, and flushes a full INSERT batch every batch_size rows — so the driver's bound parameter buffers (including MSSQL fast_executemany) only ever hold one batch. The whole transfer runs inside a single connection and, like executemany, is committed as the connector's normal unit of work.

Choosing a batch size

batch_size is the memory-vs-overhead dial:

• Larger batches mean fewer round trips and higher throughput, but more memory per batch and — for a very large batch — a risk of the statement timing out.
• Smaller batches keep peak memory tiny and each round trip fast, at the cost of more round trips overall.

The default of 10000 is a good starting point. If rows are wide (many or large columns) lower it; if rows are narrow and the link is fast you can raise it. batch_size must be greater than zero — "insert everything at once" would defeat the point of streaming and is rejected.

Monitoring progress

The connection execution logs one executemany_from_file progress entry per flushed batch, with batch_index, batch_rows and a running rows_done. On a long transfer this gives you a live cadence of how far the insert has got — open the connection execution to watch it. Only metadata is logged (batch counters and a truncated statement), never the row data.

Cleaning up

The staged NDJSON file is a normal Cloudomation file. Delete it once the insert succeeds — system.file(name).delete(permanently=True) — so staged datasets do not accumulate. Putting the delete in a finally block (or using a short-lived, uniquely-named file per run) keeps things tidy even if the insert fails.