Staging (H2) Specification

Status: v1.18 (frozen contract — additive-only changes after this point) Owner: datapipelines.co core Depends on: Type System spec, Pipeline Contract spec, Configuration spec Last updated: 2026-09-16

1. Purpose

Staging is the in-memory H2 database that holds intermediate result sets during a pipeline execution. Each pipeline execution gets its own isolated H2 instance, created when the execution starts and destroyed when it ends. No state survives between executions.

This spec defines:

• The H2 instance lifecycle (create, populate, query, destroy).
• Table naming, identifier safety, and creation rules.
• Type mapping from canonical types to H2 column types.
• Streaming-in / streaming-out behavior.
• Memory management and limits.
• Cleanup guarantees.
• The Staging abstraction (interface) so future staging engines (DuckDB) can be swapped in.

In v1, H2 is the only supported staging engine. The abstraction is designed to allow DuckDB as an alternative in future versions.

2. Design Principles

1. Per-execution isolation. Every execution gets its own H2 instance. No cross-execution data leakage. No cleanup race conditions. No need for namespace prefixes.
2. In-memory only. H2 runs in MEMORY mode — no disk I/O, no persistence. The cost is RAM; the benefit is speed. Executions that exceed memory limits fail explicitly (rather than silently swapping to disk).
3. Created-on-demand, destroyed-on-completion. The instance is created when the executor starts and destroyed deterministically when the executor finishes, regardless of success/failure. Destruction is an explicit table drop (enumerate + DROP TABLE, §3.4) plus a connection close in a finally block — never a reliance on garbage collection.
4. Tables named per the Pipeline Contract. Tables use the exact output.table names declared in the pipeline (stg_orders, int_revenue). No prefixes, no UUIDs. Downstream template SQL references these names directly.
5. A bounded pool of connections, one owner per connection. Each instance holds a small pool of JDBC connections to its database — at most datapipelines.staging.h2.max-connections (default 4) — and every operation leases exactly one for the span of its own statements and cursor (§9). A JDBC Connection does not safely serialize concurrent callers on its own, and the executor runs nodes concurrently: the lease is the mechanism, not an implementation detail. Independent nodes overlap inside H2 across connections; nothing ever touches a leased connection but its lease. (v1.0–v1.13: one connection guarded by a Mutex; replaced by #118.)
6. Streaming, not buffering. Source ResultSets stream into H2 via batched inserts — constant memory regardless of result size.
7. Every generated identifier is validated and quoted. Table names are validated at pipeline save time; column names arrive from user-authored SQL at runtime and are validated and double-quoted before they reach any generated DDL or DML (§4.5).

3. Lifecycle

3.1 Creation

enum class StagingEngine { H2 }   // v1: H2 only; see enums.md

interface StagingFactory {
    fun create(executionId: UUID, engine: StagingEngine = StagingEngine.H2): Staging
}

class H2StagingFactory(private val config: H2StagingProperties) : StagingFactory {
    override fun create(executionId: UUID, engine: StagingEngine): Staging {
        require(engine == StagingEngine.H2) { "v1 supports only StagingEngine.H2" }
        val jdbcUrl = "jdbc:h2:mem:exec_${executionId};MODE=PostgreSQL;DATABASE_TO_LOWER=TRUE"
        // Two-phase, non-admin operational connection (§9.5). Bootstrap sa creates the
        // in-memory DB + a restricted user; the FIRST operational connection is opened as
        // that user BEFORE the bootstrap closes (a bootstrap closing first would take the DB
        // with it — §3.1 last-connection semantics), then the bootstrap is closed. The pool
        // opens further restricted connections on demand, up to max-connections (§9).
        val first = openRestrictedConnection(jdbcUrl)          // authenticates as STAGING_EXEC
        val pool = H2ConnectionPool(executionId, first, opener = { openRestrictedConnection(jdbcUrl) }, config.maxConnections)
        return H2Staging(executionId, pool, config)
    }
}

The signature is shared with DAG Executor §9 — that doc's call site is canonical; this spec conforms to it. An unsupported engine (e.g. DUCKDB requested but not on the classpath) fails with pipeline.staging.engine_unavailable; any other creation failure fails with pipeline.staging.creation_failed (§7.2).

Key points:

• JDBC URL: jdbc:h2:mem:exec_{execution_id} — per-execution, isolated, in-memory.
• No DB_CLOSE_DELAY. Default H2 semantics apply: the in-memory database exists only while at least one connection to it is open, and is discarded when the last connection closes. This is exactly the lifetime we want (§3.4). DB_CLOSE_DELAY=-1 would keep the database alive until JVM exit, which in a long-lived server is an unbounded leak — one abandoned staging DB per execution, forever.
• MODE=PostgreSQL: H2's PostgreSQL compatibility mode. Makes H2's SQL syntax closer to PG (which most users know), enables some PG-specific functions. This is a SQL-syntax choice, not a type-system choice — H2 still uses its own type system internally; we map canonical → H2 explicitly per Type System §6.
• DATABASE_TO_LOWER=TRUE is load-bearing, and hardcoded (not a config key). MODE=PostgreSQL alone does not make H2 lower-fold unquoted identifiers the way PG does — H2 keeps its native upper-folding, so the author style the specs themselves use (SELECT n FROM stg_orders, DAG Executor §6.5, Pipeline Contract §10.2, Templates §11) resolved as STG_ORDERS and failed with SQLState 42S03 against the staged quoted-lowercase tables (§4.5): the canonical multi-node pipeline was broken end to end. This parameter is what makes §11.3's "unquoted references to staged tables work" true (verified against the pinned driver, 2.3.232). A deployment that removed it would break every multi-node pipeline, which is why it is a correctness invariant of the identifier scheme rather than an operator choice. Consequence: H2 also lower-cases its own catalog names (information_schema, pg_catalog, user names, schema names), so any staging-internal or test query filtering on catalog values compares them case-insensitively (UPPER(...)) rather than against bare upper-case literals.
• Every operational connection is a non-admin user, not sa (§9.5). A transient sa bootstrap creates the database and the restricted user, then closes; author SQL runs de-privileged so it cannot reach the host. Connections the pool opens later use the same URL, mode, folding and restricted credential (retained privately by the pool's opener), and are opened only while another operational connection already holds the database open — so they always land in the execution's database and never create one. (The bootstrap sa itself keeps its empty password for the DB's lifetime — acceptable because author SQL can never open a new connection to reclaim it: the functions that would let it, LINK_SCHEMA/CREATE ALIAS, are exactly what the restricted user is refused. The containment target is author SQL, not arbitrary in-JVM code, which is already game-over independent of staging.)
• Memory limit resolution. The effective per-execution limit is the pipeline's settings.tempdb.config.max_memory_mb (Pipeline Contract §5.1) when present, otherwise the global datapipelines.staging.h2.max-memory-mb (Configuration §3.3). The factory resolves this once at creation and stores it on the instance; nothing re-reads global config mid-execution.

3.2 Population

For each node whose output.target is tempdb, the executor stages the source ResultSet:

// Since 108 §B a connection is taken per batch inside the drain, not around the whole method —
// see §4.3 and §9.2. The mapping below holds no lease at all: it reads the SOURCE cursor's
// metadata and touches no staging connection.
suspend fun stage(resultSet: ResultSet, tableName: String, sourceDialect: Dialect): StageResult {
    val metadata = resultSet.metaData
    val indices = 1..metadata.columnCount

    // Column names come from user SQL — validate before they touch generated DDL (§4.5).
    val columnNames = validateColumnNames(indices.map { metadata.getColumnLabel(it) })

    // CRITICAL: the SOURCE dialect's mapper, not H2's. A Postgres/Oracle/MySQL source's
    // JDBC type codes and type names mean different things than H2's — Oracle DATE (91) is a
    // TIMESTAMP, MySQL bit(n>1) is BINARY, etc. Mapping source metadata through H2IngressMapper
    // silently picks the wrong H2 storage type and loses data (e.g. Oracle DATE's time
    // component) BEFORE egress re-derivation can see it. The executor knows the dialect from
    // node.source; for a tempdb→tempdb node it passes Dialect.H2.
    val dialectMapper = TypeMappers.forDialect(sourceDialect)
    // mapColumn (not map) so an unknown source type's §8.2 warning names the column.
    val mapped = columnNames.mapIndexed { j, name ->
        dialectMapper.mapColumn(name, metadata.getColumnType(j + 1), metadata.getPrecision(j + 1),
            metadata.getScale(j + 1), metadata.getColumnTypeName(j + 1))    // → MappedColumn(column, warnings)
    }
    val warnings = mapped.flatMap { it.warnings }   // surfaced on StageResult; dag rolls them into the execution result
    val columns: List<ColumnSchema> = columnNames.zip(mappings) { name, m -> m.toColumnSchema(name) }

    val h2ColumnDecls = columns.map { c -> "\"${c.name}\" ${H2EgressMapper.toH2Type(c)}" }

    reserve(tableName)                                            // the in-process duplicate guard — §4.5
    pool.lease(INTERNAL) { createTable(it, tableName, h2ColumnDecls) }  // the database's own guard — §4.5
    // The drain leases a connection per BATCH, not for its whole length — §4.3, §9.2 (108 §B).
    val rowsStaged = drainInto(tableName, columns, mappings, resultSet) // rolls the partial table back on failure
    recordStaged(rowsStaged)                                      // the §8.2 counter, under the metadata lock

    return StageResult(tableName, rowsStaged, columns)
}

StageResult.columns is List<ColumnSchema> — the canonical, wire-facing descriptor defined in Type System §7.1. LogicalTypeMapping is the internal ingress artifact (source JDBC type + precision/scale + resolved canonical type); it stays inside the staging layer and is never returned across the interface.

3.3 Querying

For nodes with source: "tempdb", the executor runs the rendered SQL through withQuery, which leases one connection for the entire consumption of the cursor — creation, execution, and the caller's row-by-row drain — so the cursor's connection is never touched by anyone else while the cursor is open:

suspend fun <T> withQuery(sql: String, block: suspend (ResultSet) -> T): T = pool.lease(AUTHOR) { connection ->
    connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY).use { stmt ->
        stmt.setQueryTimeout(config.queryTimeoutSeconds)
        stmt.fetchSize = config.resultBatchSize
        block(stmt.executeQuery(sql))
    }
}

The block is where the executor does the downstream work:

• A downstream node's stage operation (streaming into a new H2 table).
• The caller node's result capture (Pipeline Contract §9) — materialized to the result store per REST API §7. That materialization is suspending Redis I/O, and it runs inside the lease: correctness requires no other statement execute against the cursor's connection until the cursor is fully drained.

This closes by construction the interleaving §9.2 warns about — earlier drafts returned a live ResultSet and relied on the caller's discipline to consume it before the next staging op, a guarantee the type system could not enforce (v1.5). The cost is that a long caller-node drain occupies one of the execution's connections for its duration; since #118 the other connections stay available to independent work, and at max-connections = 1 the drain serializes staging exactly as it did before.

3.4 Destruction

class H2Staging(
    override val executionId: UUID,
    private val pool: H2ConnectionPool,
    private val config: H2StagingProperties,
) : Staging {
    override fun close() {
        // Non-throwing, non-waiting (§6 of #118): the pool refuses new leases, runs the sweep on
        // one owned connection when no lease is outstanding, closes every idle connection, and
        // quarantines any lease still inside a driver call — that lease's own return finishes
        // the sweep and closes its connection. pipeline.staging.cleanup_failed is logged, never
        // rethrown.
        val outcome = pool.close { connection -> dropStagedTables(connection) }
        if (outcome.leasesOutstanding > 0) log.warn("tempdb closed with {} lease(s) in flight …", …)
    }
}

Two independent mechanisms, in order:

1. Enumerate-and-drop (INFORMATION_SCHEMA.TABLES → DROP TABLE per table) — releases every staged table's memory immediately and deterministically, before the last connection close. Belt. (DROP ALL OBJECTS would be simpler but is admin-gated in H2 2.3.232, and the staging user is non-admin by §9.5; the enumerate-and-drop uses only non-admin operations.) The sweep runs only when it cannot race an active borrower: on a connection the pool owns, after leases are refused, when none is outstanding — otherwise it is deferred to the last late return.
2. Closing every owned connection — with default close semantics (§3.1), closing the last connection destroys the in-memory database itself. Braces. This is the primary guarantee; step 1 only accelerates memory release within a long-lived JVM.

Neither step depends on garbage collection. close() does not throw for a SQL or runtime cleanup failure: it is invoked from the executor's finally block, where an exception would mask the execution's real failure. A failed sweep is logged and surfaces as pipeline.staging.cleanup_failed (§7.2) in the execution's error detail when the execution is otherwise successful — and it never skips the rest of the cleanup: the sweep and the physical closes are finalized in nested finally blocks, so one fault cannot leave an idle session open or the pool stuck in a closing state; a second close is a no-op. A physical close the driver refuses is logged and counted (refusedCloses) rather than reported as closed — the slot is released, but the raw session may still be live, and that residual is stated, not hidden. A JVM Error raised during cleanup propagates, but only after every owned connection was closed and the pool is terminal (146c).

close() never waits for a lease still inside the driver. The executor's node deadline can abandon a body that is blocked in a JDBC call the driver refuses to interrupt (DAG Executor §5.3); waiting for it in close() would turn one unresponsive statement into an unbounded shutdown wait, and closing or resetting the connection under it would corrupt a call in progress. Such a lease is quarantined: it keeps its connection — and with it the database — until the call returns, at which point the return closes the connection; the same applies to a checkout whose open was in flight and to a guardian replacing a failed session (§9.2). Whichever of them is the LAST session out runs the deferred sweep once first, when it is usable. The residual is stated honestly: a driver call that never returns keeps one connection and one in-memory database alive; the executor logs the abandonment (node.statement_abandoned) and the pool logs the outstanding lease count at close.

3.5 Lifecycle guarantee

The H2 instance cannot outlive the execution:

• Opened at execution start. The executor calls StagingFactory.create(executionId), and the instance's pool keeps at least one operational connection open for the whole execution — including the periods when no node is touching staging. This is what keeps the database alive: default H2 semantics discard an in-memory DB the moment its last connection closes, so an "open a connection per operation" model would destroy the staged tables between nodes. Idle pooled connections are never closed while the instance is open; a broken connection is replaced before it is closed (§9.2).
• Held in a local inside PipelineExecutor.execute(...); no reference escapes to a long-lived object, cache, or registry.
• Closed in the finally of that same function (DAG Executor §5, §9) — on success, on node failure, on execution timeout, and on cancellation (client disconnect beyond grace, DELETE /api/v1/executions/{id}, or shutdown). The pool, not a single connection, is the lifetime owner: close() closes every connection it holds (§3.4).
• No GC dependency anywhere. The previous version of this spec claimed the -1 close-delay flag tied the database's lifetime to open connections. That was factually wrong (§3.1), and the flag is gone.

A JVM crash mid-execution abandons the in-memory DB, which is fine — it is process memory, reclaimed by the OS.

4. Table Naming, Identifier Safety, and Creation

4.1 Table names

Table names come directly from the pipeline's output.table declarations:

• stg_orders, stg_customers — staging from source
• int_revenue, int_customer_summary — intermediate transformations

Rules are defined and validated at pipeline save time — see Pipeline Contract §10. Summary (that doc is authoritative):

• [a-z0-9_]+, length 1–63.
• Unique among all tempdb targets in the pipeline (the tempdb namespace shares one staging database).
• Not tempdb, and not any name starting and ending with __ (reserved namespace).

Because no pipeline can be saved with an invalid or colliding tempdb table name (Pipeline Contract §2, universal save-time validation), the staging layer treats a violation at runtime as a defect and fails loudly rather than sanitizing (§4.5).

4.2 CREATE TABLE generation

For a ResultSet with columns [id INTEGER, name VARCHAR(100), total_amount NUMERIC(18,2)]:

CREATE TABLE "stg_orders" (
  "id" INTEGER,
  "name" VARCHAR,
  "total_amount" DECIMAL(18, 2)
)

Notes:

• Every identifier is double-quoted (§4.5). Table names are already lowercase-only by contract, so quoting does not change how templates reference them; column names may be mixed-case, and quoting makes them exact.
• H2 VARCHAR without length spec = unbounded (practical limit 1GB). We don't propagate source length to H2 — source-DB length limits are not our concern (the source data is what it is).
• H2 DECIMAL(p, s) preserves exact precision. An exact-unsized numeric (canonical BIGDECIMAL with precision and scale omitted, type-system §4) is DDL'd as DECFLOAT(100000) — H2's exact, arbitrary-scale decimal — never DECIMAL(100000, 0): the driver reported scale 0 for unknown, and a declared 0 truncated every fraction on insert (defect 100).
• H2 INTEGER is 32-bit, BIGINT is 64-bit.

4.3 Batch inserts

The source cursor is drained holding no lease (108 §B, #118). The loop below is the shape: read up to insert-batch-size rows from the source cursor holding nothing, lease a connection for the INSERT, close the statement, return the lease, read the next batch. The materialised batch is the cost, and it is bounded by insert-batch-size; what it buys is that none of the execution's few connections is ever held across a network wait on the source database. A PreparedStatement never crosses a returned lease — the next batch may land on a different physical connection — so the insert is re-prepared per batch (H2's per-session query cache makes that a lookup after the first). stageRows pulls its child-row sequence in the same batches, holding no lease between them.

The drain reports its boundaries (149). stage and stageRows take a StageObserver (default NONE): connectionRequested/connectionAcquired around the CREATE TABLE lease and around each batch's insert lease, fetchStarted/fetchFinished(rows) around each batch read, batchWritten(rows, rowsSoFar) after the lease has RETURNED — the 108 §D rows-so-far figure rides that last call — and, only on failure, partialTableDropped() once the partial table's drop succeeded on its fresh lease (the confirmed undo; not reported when the drop failed or timed out, so the write's fate stays unknown to the observer). Every call is made outside the pool's metadata lock and, except connectionAcquired (the boundary that says "holding"), outside the lease; the observer must be non-suspending and prompt, since it runs on the drain's own path. The executor bridges it onto the node's operation tracker (DAG Executor §10); at capacity one, a node whose sibling holds the pool's only connection is seen WAITING at the CREATE TABLE lease, before its first batch — an unobserved wait there read as a slow source query.

private suspend fun drainBatches(...): Long {
    var rowCount = 0L
    while (true) {
        ensureActive()                                                   // a cancelled node stops at the batch boundary
        val batch = readBatch(rs, mappings, config.insertBatchSize)      // NO lease held here
        if (batch.isNotEmpty() || rowCount == 0L) {
            pool.lease(INTERNAL) { connection ->                         // one connection, this batch only
                insertBatch(connection, tableName, columns, sqlTypes, batch)   // prepare, bind, executeBatch, close
            }
            checkBudgetIfDue(...)                                        // the JVM's reading — no connection needed
        }
        rowCount += batch.size
        if (batch.size < config.insertBatchSize) break
    }
    checkMemoryBudget()
    return rowCount
}

The source statement has to stream, or none of this means anything. A driver that materialises the whole result set before handing back a cursor makes insert-batch-size a batching of rows that are already in the heap, and the memory budget a statement about a copy that already exists. The executor therefore sets fetchSize on every DQL source statement (datapipelines.executor.source-fetch-size, default 1000) and takes a Postgres source connection out of autocommit for the read — pgjdbc uses a server-side cursor only with both. MySQL is the driver-forced exception: Connector/J streams only at Integer.MIN_VALUE. Before 108 the executor set neither, on any dialect.

The pre-108 shape, for reference:

private fun batchInsert(
    tableName: String,
    columns: List<ColumnSchema>,
    mappings: List<LogicalTypeMapping>,
    rs: ResultSet,
): Long {
    val columnList = columns.joinToString(",") { "\"${it.name}\"" }
    val placeholders = columns.joinToString(",") { "?" }
    val sql = "INSERT INTO \"$tableName\" ($columnList) VALUES ($placeholders)"

    return connection.prepareStatement(sql).use { stmt ->
        var rowCount = 0L
        val batchSize = config.insertBatchSize

        while (rs.next()) {
            mappings.forEachIndexed { i, m ->
                val value = readValue(rs, i + 1, m)
                stmt.setObject(i + 1, value, H2EgressMapper.h2SqlType(columns[i]))
            }
            stmt.addBatch()

            if (++rowCount % batchSize == 0L) {
                stmt.executeBatch()
            }
        }
        if (rowCount == 0L || rowCount % batchSize != 0L) {
            stmt.executeBatch()
        }
        rowCount
    }
}

The column list is written explicitly (not positional INSERT INTO t VALUES (...)) so the statement is independent of H2's column ordering. insertBatchSize comes from configuration (§7.1).

Streaming: the loop reads one row at a time from the source ResultSet. Memory footprint of the transfer is bounded by the batch size (batch rows × row size); the staged table itself is accounted against the memory budget (§8). A 10M-row source ResultSet stages with constant transfer memory.

4.4 Value reading

Per canonical type, the value is read from the source ResultSet and converted to the appropriate Java type for the H2 insert (readValue, signature in §5.3). Two normative read rules (2026-08-08):

• STRING-canonical columns are read with getString(index), never getObject. Several §5.x mappings assign binary-coded or driver-object JDBC columns to canonical STRING (MySQL geometry as WKT, PG arrays, Oracle XMLType, CLOBs); getObject(...).toString() on those yields Java identity text ([B@6d06d69c) shipped as a plausible-looking value with no warning. getString makes the driver do the conversion.
• Temporal columns are read with JDBC 4.2 getObject(index, OffsetDateTime/LocalDate/LocalTime::class.java), never getTimestamp/getDate/getTime — the java.sql temporal types convert through the JVM default zone, and the typesystem's JsonEncoder/UtcNormalization reject them by design (§8.4 machine-independence).

4.5 Identifier safety (normative)

Two classes of identifier reach generated SQL, and they have different threat models.

Table names — trusted by construction. They are pipeline-declared and fully validated at save time by Pipeline Contract §10. No pipeline with an invalid tempdb table name can exist in the database (D2 universal save-time validation). The staging layer re-quotes them but does not re-derive the rule.

Column names — attacker-adjacent. They come from the result set metadata of user-authored SQL (SELECT x AS "whatever the author typed"), which is rendered from a template with pipeline parameters. They are never trusted. Before a column name is interpolated into any generated DDL or DML, the staging layer MUST:

1. Validate the shape. Each column label must match [A-Za-z_][A-Za-z0-9_]{0,62} (leading letter or underscore; letters, digits, underscores thereafter; total length 1–63, H2's identifier limit). An empty, null, over-long, or otherwise non-matching label fails the node.
2. Reject duplicates. Column labels must be unique within one staged result set. Comparison is case-insensitive, matching H2's unquoted-identifier folding — total and TOTAL collide. (SQL happily produces duplicate labels — SELECT a.id, b.id FROM ... — so this is a routine authoring mistake, not just an attack.)
3. Double-quote unconditionally. Every identifier — table and column — is emitted as "name" in generated CREATE TABLE, INSERT, and DROP statements. Validation is the security boundary; quoting is the second layer, and it also makes mixed-case labels exact rather than folded.

Failure of (1) or (2) → the node fails with pipeline.staging.invalid_column_name, with the offending label and its ordinal position in the error details. Sanitizing is explicitly forbidden: renaming a bad column to col_3 would silently change the schema the caller receives and the names downstream source: tempdb templates must use. The author must fix the alias in their SQL.

private val COLUMN_NAME = Regex("[A-Za-z_][A-Za-z0-9_]{0,62}")

/** Validates and returns the labels in order; throws StagingInvalidColumnNameException otherwise. */
private fun validateColumnNames(labels: List<String?>): List<String> {
    val seen = mutableSetOf<String>()
    return labels.mapIndexed { i, raw ->
        val label = raw ?: throw StagingInvalidColumnNameException(ordinal = i + 1, label = null)
        if (!COLUMN_NAME.matches(label)) throw StagingInvalidColumnNameException(i + 1, label)
        if (!seen.add(label.uppercase())) throw StagingInvalidColumnNameException(i + 1, label)
        label
    }
}

Duplicate staged table (defensive). createTable issues a bare CREATE TABLE — never CREATE TABLE IF NOT EXISTS, never an implicit DROP. If the table already exists in this execution's staging database, the node fails with pipeline.staging.table_already_exists. Save-time uniqueness validation (§4.1) is the primary guard and should make this unreachable; reaching it means either a validation gap or a node executing twice, and both are bugs worth surfacing loudly rather than papering over by overwriting a table another node is about to read.

5. Type Mapping to H2

See Type System §6 for the full canonical → H2 mapping table. Summary:

5.1 Why TIMESTAMP WITH TIME ZONE in H2

Canonical TIMESTAMP is always UTC (per Type System §8.4). H2's TIMESTAMP WITH TIME ZONE preserves the timezone (UTC). When we read back for egress, we get a UTC value directly, no normalization needed.

Plain H2 TIMESTAMP (without TZ) is a candidate, but TIMESTAMP WITH TIME ZONE makes the UTC assumption explicit and prevents accidental TZ-conversion bugs in H2's own functions.

5.2 Precision overflow handling

H2 2.x supports DECIMAL precision up to 100000 (Type System §6). Source precisions from any supported dialect fit. If a source declares precision beyond that limit, staging fails with pipeline.staging.precision_overflow.

5.3 Mappers and helper signatures

The H2 type translation is two directions, two objects — they are not inverses of one another in practice (egress must pick a DDL type string and a java.sql.Types code; ingress must recover a canonical descriptor from H2 metadata), and conflating them in one object hid that asymmetry.

/** Canonical → H2. Used when generating DDL and binding insert parameters (§4.2, §4.3). */
object H2EgressMapper {
    /** H2 column type as written in CREATE TABLE, e.g. "DECIMAL(18, 2)", "VARCHAR". */
    fun toH2Type(column: ColumnSchema): String

    /** java.sql.Types constant for PreparedStatement.setObject(index, value, targetSqlType). */
    fun h2SqlType(column: ColumnSchema): Int
}

/** H2 → canonical. Used when reading staged data back out (§6). */
object H2IngressMapper {
    /**
     * Builds the canonical descriptor for one column of an H2 ResultSet.
     * jdbcType/precision/scale come from ResultSetMetaData; label is already validated (§4.5).
     */
    fun fromH2(label: String, jdbcType: Int, precision: Int, scale: Int): ColumnSchema
}

/**
 * Reads one value from the SOURCE ResultSet per the canonical mapping table (§4.4),
 * applying wasNull checks and UTC normalization for TIMESTAMP. Returns null for SQL NULL.
 */
private fun readValue(rs: ResultSet, index: Int, mapping: LogicalTypeMapping): Any?

Both mapper names are the ones used in Module Structure §5.1 — H2IngressMapper and H2EgressMapper. There is no H2TypeMapper.

6. Streaming-Out

For the caller node's ResultSet (Pipeline Contract §9), the executor streams rows from the H2 ResultSet into the wire format (JSON / Arrow / CSV):

fun streamResult(rs: ResultSet, format: WireFormat): StreamedResult {
    val metadata = rs.metaData
    val columns: List<ColumnSchema> = (1..metadata.columnCount).map { i ->
        H2IngressMapper.fromH2(
            metadata.getColumnLabel(i),
            metadata.getColumnType(i),
            metadata.getPrecision(i),
            metadata.getScale(i),
        )
    }

    when (format) {
        WireFormat.JSON -> {
            val rowStream = sequence {
                while (rs.next()) {
                    yield((1..metadata.columnCount).map { i ->
                        encodeValue(rs, i, columns[i - 1])
                    })
                }
            }
            return StreamedResult.Json(columns, rowStream)
        }
        // Arrow, CSV similar
    }
}

The H2 → canonical mapping is the inverse of the source → H2 mapping (covered by the H2 row in Type System §5.5).

6.1 Memory-bounded result handling

Result delivery has a single path: every caller result is materialized to the result store (Redis) and read back through the cursor — see REST API §7. There is no inline-vs-claim-check branch and no size threshold that switches modes.

For the staging layer that means:

1. The H2 ResultSet is consumed row-by-row at datapipelines.staging.h2.result-batch-size rows per fetch (Configuration §3.3).
2. Rows are written straight through to the result store as they are read, inside the executor's connection.use block (DAG Executor §6.4) — never buffered whole in JVM memory.
3. The data_ready event's inline first page is read back from the stored result, not held aside during streaming.
4. If the accumulated result exceeds datapipelines.result.max-size-bytes (Configuration §3.5), the execution fails with result.too_large — a result-delivery error, not a staging error.

A 100M-row caller ResultSet therefore does not OOM the JVM; it either streams into the result store at steady memory cost or trips the size cap.

7. Configuration and Error Codes

7.1 Configuration keys

Staging reads its settings from datapipelines.staging.h2.*, defined — names, defaults, and descriptions — in Configuration §3.3. That document is the single authority; no defaults are restated here.

Keys consumed by this spec:

Per-pipeline override. settings.tempdb.engine selects the engine and settings.tempdb.config.max_memory_mb overrides max-memory-mb for that pipeline (Pipeline Contract §5.1, precedence per Configuration §4). No other staging key is per-pipeline overridable in v1.

7.2 Error codes

Staging error codes are cataloged centrally in Pipeline Contract §13.5. The codes this layer raises:

There is no pipeline.staging.h2_creation_failed — the engine-neutral creation_failed is the canonical code.

8. Memory Management

8.1 Per-execution memory limit

Each staging instance has a memory budget, resolved once at creation (§3.1): the pipeline's settings.tempdb.config.max_memory_mb when present, otherwise the global datapipelines.staging.h2.max-memory-mb. H2 does not enforce a hard cap on an in-memory database, so the staging layer measures and aborts.

8.2 Memory accounting — measured, not estimated

The staging layer does not estimate footprint from row counts and average row widths — that arithmetic is unreliable for VARCHAR/VARBINARY-heavy tables and was wrong in both directions.

Accounting is a direct measurement of JVM heap, read in-process — not via H2's MEMORY_USED():

fun usedHeapKb(): Long {
    System.gc()   // match MEMORY_USED()'s post-GC semantics; coarse guard, not per-poll hot path
    val rt = Runtime.getRuntime()
    return (rt.totalMemory() - rt.freeMemory()) / 1024
}

• Why not SELECT MEMORY_USED(): in H2 2.3.232, MEMORY_USED() requires admin rights (SQLState 90040) — and the staging connection is deliberately a non-admin user so author SQL cannot reach the host (§9.5). Empirically, MEMORY_USED() and (totalMemory − freeMemory) return the same number (both ~14271 KB after a 50k-row fill in the same instant): H2's MEMORY_USED() is itself "run a GC, then return used heap", not a measure of the database's own allocation. So the in-process reading is the identical quantity with no admin dependency.
• Polled on the first batch of a drain, then at most once per second, and unconditionally when the drain completes (108 §B), plus after each execute(sql) that writes to staging. Mid-drain polling is new: with the drain reading outside the lock, a per-table-only check meant a runaway node could put its whole result in the heap before anything looked. Mid-drain checks are two-tier: a cheap reading WITHOUT a collection (garbage included, so only ever an over-estimate — a false alarm, never a miss) runs on the cadence, and the accurate GC-forcing reading above decides only when the cheap one exceeds the budget — at the default batch size a 2M-row stage is 2 000 batches, and several drains run concurrently, so paying a System.gc() per batch would cost more than the inserts it guards. The first batch is always checked, so a budget already blown when the stage began is refused immediately.
• Reading is in kilobytes; budget in megabytes. Compare as usedHeapKb > maxMemoryMb * 1024.
• Exceeding the budget fails the current staging operation with pipeline.staging.memory_limit_exceeded, carrying the measured value and the budget in the error details.
• The same reading backs StagingStats.memoryUsedBytes (§10).

Rows staged per table and the table count are tracked as plain counters for observability; they are reported, not used to decide the limit.

Known limit — the reading is JVM-heap-wide, not per-execution (v1). Because it measures used heap for the whole JVM, it includes heap held by in-flight ResultSets, other executions' staging DBs, wire buffers, and the result-store writer — everything, not just this execution's tables. With concurrent executions every staging instance reads the same global number, so a single execution's max_memory_mb is in practice a shared JVM-heap ceiling, not an isolated per-execution budget: one heavy execution can trip a lighter one's check. This is a deliberate v1 simplification (it is the cheapest guard that reliably stops a genuine runaway before OOM), true regardless of which H2 user runs it; real per-execution memory isolation is a v1.1+ item (§13.2 lists the accounting model as not frozen). §8.4 is the hard JVM backstop underneath it.

8.3 Failure handling

On memory-limit failure:

1. The current staging operation throws StagingMemoryLimitException.
2. The executor catches it, wraps as NodeExecutionException, fails the node (DAG Executor §8.2).
3. The execution fails fast; pipeline_failed SSE event sent.
4. Cleanup runs in finally (§3.4); the staging DB is dropped and closed; memory freed.

8.4 JVM-level safety net

The JVM-level safety net is configured via:

• Container memory limit (Docker / k8s).
• JVM heap size (-Xmx) — sized as staging max-memory × max-concurrent-executions + baseline (see the resource-sizing guidance in Deployment).
• Off-heap buffer pool limits (for Arrow / large BLOB handling).

If the JVM OOMs mid-execution, the in-memory staging database dies with the process. No persistent state corruption (staging is in-memory only).

9. Connection Model

9.1 The choice

Each staging instance is backed by a bounded pool of operational JDBC connections to its database — at most datapipelines.staging.h2.max-connections (default 4, 1 allowed), the bootstrap-handoff connection included (Configuration §3.3; #118). This means:

• Every node that touches staging — staging in, querying out, DML against tempdb — leases one connection for the span of its own statements and cursor, and returns it.
• The pool grows on demand from the one bootstrap-handoff connection; four is a ceiling, not an eager allocation. At least one connection stays open for the whole execution, which is what keeps the in-memory database alive (§3.5).
• The cap is capacity, not parallelism: it bounds how many of one execution's tempdb operations can be inside H2 at the same instant. It creates no additional eligible DAG nodes (executor.max-parallel-nodes does) and promises no speedup — H2 keeps its own transaction, row and catalog locks, so two nodes touching one table still serialize inside the engine. A cap below max-parallel-nodes queues nodes safely; a cap above it is never fully opened.

Every operational connection authenticates as the same non-admin H2 user (§9.5) — the transient admin connection that creates the database also creates the restricted user, and is closed before the module does any work. Connections opened later reuse that credential, which the pool's opener retains privately for the execution's lifetime.

v1.0–v1.13 held one connection guarded by a kotlinx.coroutines.sync.Mutex; the pool replaced it because the executor runs up to max-parallel-nodes nodes concurrently and their tempdb work — CTAS chains, DML, staging inserts — could not overlap at all.

9.2 One owner per connection, and session state is per lease

The executor runs nodes concurrently (up to datapipelines.executor.max-parallel-nodes), so concurrent access to staging genuinely happens; two nodes can complete their source fetches at the same time and both try to stage. A JDBC Connection is not required by the JDBC spec to serialize concurrent callers safely, and H2's connection is not a safe multiplexing point: interleaved statement execution on one connection can corrupt statement state, scramble results, or throw obscure driver errors.

Therefore:

• A lease exclusively owns one physical connection through its statements and result sets, from checkout to the end of its finally, on every exit — value, exception, cancellation, a failed checkout. Nothing outside the lease touches that connection: not the reset, not close(). Admission past the cap is a cancellable coroutine Semaphore — a waiter suspends on the executor's bounded dispatcher rather than blocking a thread the lease holders need, and a waiter cancelled by its node or execution deadline never runs SQL and never strands a permit.
• Independent work overlaps across connections. Two nodes inside H2 at the same instant run on two different connections; H2StagingConcurrencyTest proves it with a barrier inside a driver call on two leases — a global lock around SQL, batches or cursor consumption reintroduced anywhere makes that guard fail. The per-connection half is the old invariant unchanged: no physical connection ever has two callers inside it.
• The invariant is about the CONNECTION, not about the network (108 §B). stage() reads a batch from the source cursor holding no lease, leases a connection for the INSERT and closes its statement before returning it, then reads the next batch (§4.3). A network wait on someone else's database never holds one of this execution's connections. stageRows pulls its child-row sequence the same way.
• Cursor consumption is enforced by construction, not convention: withQuery(sql) { rs -> … } (§3.3, §10) retains its lease for the whole lifetime of the cursor, including the caller node's suspending drain to the result store (§6.1). There is no API that returns a live ResultSet to be read after the lease is returned. Other connections remain available to independent work during a long drain.
• Direct SQL access goes through withConnection(block) (§10), which leases one connection for the duration of the block: the connection is never handed out unguarded, and the pool itself is not reachable — or even observable — from outside the implementation. Two rules bind the block: it must not re-enter any staging operation (stage/withQuery/execute/stats/withConnection from inside the block deadlocks at max-connections = 1 and, above it, holds one connection idle while waiting for another), and nothing derived from the connection (statements, cursors) may outlive the block.
• What is still coordinated, and how narrowly. Two things are shared across leases and guarded by one short metadata lock that never covers JDBC, a source read, a result-store write, a callback, a suspension or a wait for a lease: the set of table-name reservations (the deterministic table_already_exists guard, §4.5) and the successful staged-row total stats() reports (§8.2). A name is reserved before CREATE TABLE, so the loser of a duplicate race is refused at the reservation and never reaches the database — it cannot drop the winner's table. A failure after CREATE TABLE rolls the partial table back on a fresh, bounded lease (NonCancellable, because the node's own deadline is the usual cause) and frees the reservation only when the drop succeeded: a name whose partial table could not be removed stays owned, so a retry meets table_already_exists rather than silently reusing dirty data. stats() reads each figure once, on its own; it does not claim a transactionally frozen snapshot of a database other nodes are writing to.
• Ordinary staged tables are execution-wide database objects. Every statement runs in autocommit, so a producer node's CREATE TABLE and every flushed batch are committed before the node completes, and the executor starts a dependent node only after its dependencies completed — a dependent therefore sees the whole table whichever physical connection it draws. depends_on expresses real data dependencies; there is no implicit global SQL queue and no promised ordering between independent nodes that touch the same table.
• Session state is per lease, never a cross-node channel. Each callback or node SQL operation gets one session for its own statement/cursor lifetime. When a lease returns, the connection is sanitized against the pinned driver before anyone else can draw it: an open transaction is rolled back, never committed (setAutoCommit(true) mid-transaction would commit, so the rollback runs first); the isolation level, current schema and schema search path go back to their captured defaults; session variables (SET @var), local temporary tables and a session time zone are enumerated from INFORMATION_SCHEMA.SESSION_STATE's STATE_KEY — H2's own discriminator for exactly these kinds of state, so this is a catalog enumeration, not a parser over SQL — and undone one by one; QUERY_TIMEOUT and LOCK_TIMEOUT go back to the driver's defaults. Statements within one callback keep their session, as before. Any other SET an author can issue (VARIABLE_BINARY, NON_KEYWORDS, THROTTLE, …) is not restored and is unsupported across nodes: whichever idle connection a later node draws, it must not depend on what an earlier node left there — cross-node communication is ordinary tables plus depends_on, whatever the cap. max-connections = 1 selects capacity; it is not a promise to preserve leaked session state.
• Every session has exactly one owner, always. A physical session is opening (a checkout's slot, reserved under the metadata lock before the driver is called), idle, leased, a guardian (below) or closing — never two of these, never none. Every open, whether a checkout's or a replacement, is reconciled with the pool's state under the lock after the driver returns and before the session is published or any callback runs: a session that connected after close() began is closed by the thread that opened it and its lease is refused, exactly as an early lease would be. The invariant holds across exception boundaries too (146c): a fault thrown by session sanitisation — a driver SQLException, a runtime fault, even a JVM Error — still unlinks the session from its lease and retires it (closed, or guarding until its replacement) before the fault is rethrown; the operation's own failure is what the node reports, with the cleanup fault attached as a suppressed exception, and with no operation failure the cleanup fault is the result. A refused physical close is counted, never reported as closed (§3.4).
• Terminal admission is refused before waiting. A fresh withConnection/stage/withQuery/execute/stats call against a closing, closed or lost instance is refused at once — before it waits for a permit — so it can never park behind a permit held by a lease the executor has abandoned; the check is repeated after admission and after an open for the races that flip the state meanwhile. A caller that was already waiting for a permit when close() ran is not woken by it (a coroutine Semaphore has no broadcast): it is refused when a permit reaches it, or cancelled by its own node/execution deadline, which is the executor's existing bound on every wait.
• A failed reset is a discarded connection, and retirement counts only real holders. A connection whose sanitation throws, or that is found closed on return, is never offered to another node. It is closed at once only if another idle, leased or guardian session exists — never on the strength of an open still in flight, which may not have connected yet. Otherwise it becomes the guardian: it stays open, opens its replacement outside the lock, and closes only after the replacement is adopted — closing the last connection would destroy the database with every staged table on it. Two resets failing at the same instant therefore agree (the first to decide sees the other still leased and closes; the second sees nobody and guards), and the cap on usable connections holds while the process briefly carries one extra dead session.
• Continuity failure is explicit, never silent. If the guardian's replacement cannot be opened while nothing else holds the database, the pool becomes lost: every later lease is refused with an error naming the loss (it surfaces through the executor's phase mapping as the node's failure), the opener is not tried again, and the guardian is closed. A restricted opener could not recreate the database anyway; a privileged one must never be allowed to connect quietly to a fresh, empty database under the execution's name. Session sanitation and the ownership invariant are tested in H2ConnectionPoolTest, H2PoolOwnershipTest and the review's H2PoolReviewRaceTest (a dirtied session is returned at capacity one, forcibly reused and shown clean; a replacement finishing after close is closed; two simultaneous failed resets keep the staged table).

This corrects the earlier claim (and DAG Executor §12.1) that there is "no concurrent tempdb access." There is; it is bounded by the pool and safe because each connection has one owner at a time.

9.3 What the pool does and does not buy

• Overlap, not a multiplier. Independent CTAS/DML/staging inserts of one execution now run inside H2 at the same time, bounded by the cap. Whether that is faster depends on the pipeline: H2's DDL takes a database-wide meta lock, inserts into one table contend on that table, and the JVM's CPUs are shared. Measured figures live with the change's evidence, not in this spec; nothing here promises a speedup.
• Aggregate cost. Every connection is one H2 session with its own query working memory; max-concurrent-executions-per-instance executions can each hold up to max-connections of them. The staged tables themselves are shared by an execution's connections and are not multiplied, and the §8.2 budget check keeps its cadence (per drain, not per connection) and its process-wide reading.
• Simpler where it counts. No validation queries, no maximum lifetime, no eviction, no threads of its own: the database the pool fronts lives exactly as long as the pool does, so the general-purpose machinery has nothing to do here. H2's bundled JdbcConnectionPool was read against the pinned driver and not used: its exhausted checkout busy-polls on the caller's thread, its return reset is partial (rollback + autocommit) and swallows failures, and it has no discard or ownership model — the adapter needed around it would have been larger than the pool that replaced it.

9.4 When to revisit

If profiling shows the cap itself is the bottleneck (many independent tempdb-heavy nodes, max-parallel-nodes raised above four), raise max-connections for that deployment — it is a runtime setting. A per-pipeline override, connection validation or an acquisition-timeout knob are deliberately absent: the node and execution deadlines already bound every wait, and evidence, not anticipation, adds knobs.

9.5 Privilege containment — author SQL runs de-privileged (normative)

The rendered SQL that withQuery/execute run is author-authored (a pipeline author's template body, §4.4 of Templates). H2's admin-only surface reaches the host: FILE_READ/FILE_WRITE/CSVWRITE/CSVREAD read and write server files, CREATE ALIAS/CREATE TRIGGER … AS load JVM classes, RUNSCRIPT/LINK_SCHEMA fetch and execute. An sa (admin) staging session therefore turns "author may write tempdb SQL" into "author may read /proc/self/environ" — where DATAPIPELINES_DB_ENCRYPTION_KEY and DATAPIPELINES_JWT_SECRET live (Configuration §2). That is privilege escalation from author to all-datasource-credentials and session forgery, and it is not an accepted trade (contrast §4.4's SQL-injection note, which concerns the author's own authorized datasources, not the server's secrets).

Therefore:

• The database is created by a transient bootstrap admin (sa) connection, which immediately creates a restricted user (CREATE USER STAGING_EXEC PASSWORD '<256-bit random hex>' + GRANT ALTER ANY SCHEMA — no admin right) and is then closed. Every operational connection the module holds (§9.1) authenticates as that restricted user; the pool's connections are what keep the in-memory database alive thereafter, and connections opened after the bootstrap closed are still that user — H2StagingPoolLifecycleTest runs the refusals on three concurrent physical connections. (ALTER ANY SCHEMA is the least grant that lets the user do PUBLIC DDL — GRANT ALL ON SCHEMA PUBLIC alone leaves CREATE TABLE refused; a user-owned schema forces SET SCHEMA and rewrites. It also permits DDL inside INFORMATION_SCHEMA, which is accepted: the database is a throwaway per-execution in-memory instance with no host reach, and author SQL can already CREATE TABLE in it — a junk table in a schema about to be dropped is not an escalation.)
• Under that user, every host-reaching function is refused with SQLState 90040 ("Admin rights are required") — empirically verified against H2 2.3.232: FILE_READ, FILE_WRITE, CSVREAD, CSVWRITE, CREATE ALIAS, RUNSCRIPT, LINK_SCHEMA, CREATE TRIGGER … AS, plus the self-escalation routes ALTER USER … ADMIN TRUE / CREATE USER / SET. So author SQL cannot reach the host filesystem, load a class, or grant itself admin.
• The staging layer avoids the two admin-gated operations it would otherwise use. MEMORY_USED() and DROP ALL OBJECTS are also admin-gated (90040) in this H2 version — so accounting uses an in-process heap reading (§8.2) and cleanup enumerates INFORMATION_SCHEMA.TABLES and drops each table (§3.4), both non-admin. CREATE/INSERT/SELECT/DROP TABLE and reading INFORMATION_SCHEMA — everything the staging layer needs — are available to the restricted user.
• The admin-gating is guarded by a test (h2 in libs.versions.toml): it runs FILE_READ, CSVWRITE, and CREATE ALIAS as the restricted user and asserts each is refused. A driver upgrade re-runs it; if a future H2 un-gates one for non-admin users, the test fails and the containment is revisited before shipping.
• Optional deployment-level belt: launching the JVM with -Dh2.allowedClasses= (empty) denies CREATE ALIAS class loading independent of user rights. It must be a launch arg — SysProperties.ALLOWED_CLASSES is captured at H2 class-init, so a runtime System.setProperty is a no-op — and it is redundant once the operational user is non-admin (which already refuses CREATE ALIAS), so it is not required by this spec, only noted for deployments that want it.

10. The Staging Interface

The interface is engine-agnostic, allowing DuckDB (or other engines) to be plugged in later.

interface Staging : AutoCloseable {
    val executionId: UUID
    // Direct SQL for SQL nodes: runs block on one leased connection, owned throughout (§9.2).
    // block must not re-enter staging operations (never nest a lease), and nothing derived
    // from the Connection may escape the block. Session state is reset when the lease returns.
    suspend fun <T> withConnection(block: suspend (Connection) -> T): T

    // observer (149, §4.3): the drain's measured boundaries — the CREATE TABLE lease, each
    // batch's fetch / lease request / lease held / accepted batch — default NONE.
    suspend fun stage(resultSet: ResultSet, tableName: String, sourceDialect: Dialect,
                      observer: StageObserver = StageObserver.NONE): StageResult
    // The already-decoded twin of stage(): canonical columns + rows (composition's direct
    // delivery — a parent PIPELINE node's child rows). Same column-label validation
    // (§4.5: a malformed or case-insensitively duplicated label fails the node — labels
    // are never trusted, never sanitised), same partial-table rollback, same post-write
    // budget check; warnings always empty — the source-dialect mapping already happened
    // in the child's executor.
    suspend fun stageRows(tableName: String, columns: List<ColumnSchema>, rows: Sequence<List<Any?>>,
                          observer: StageObserver = StageObserver.NONE): StageResult
    // Runs block against the cursor with its connection leased for the WHOLE consumption
    // (§3.3/§9.2). The cursor is never handed out to be read after the lease is returned, so
    // nothing can interleave on the cursor's connection; other connections stay available.
    // block must fully consume (or abandon) the cursor before it returns; nothing derived
    // from it escapes.
    suspend fun <T> withQuery(sql: String, block: suspend (ResultSet) -> T): T
    suspend fun execute(sql: String): Long    // INSERT/UPDATE/DELETE/DDL against staging; returns row count

    suspend fun stats(): StagingStats         // current tables/rows/measured memory

    override fun close()                      // not suspend: called from finally, must not throw (§3.4)
}

data class StageResult(
    val tableName: String,
    val rowsStaged: Long,
    val columns: List<ColumnSchema>,
    val warnings: List<TypeMappingWarning> = emptyList()   // §8.2 warnings from the source→canonical mapping
)

data class StagingStats(
    val tableCount: Int,
    val totalRows: Long,
    val memoryUsedBytes: Long        // measured JVM heap, in-process (§8.2), not estimated
)

ColumnSchema is the canonical column descriptor from Type System §7.1.

10.1 Future: DuckDB staging

For analytical workloads (large joins, aggregations on wide tables), DuckDB would outperform H2. The interface above is designed so that DuckDbStaging could be a drop-in replacement. Differences:

• JDBC URL: jdbc:duckdb:memory:exec_{id}.
• Type mapping: similar but DuckDB has HUGEINT, native nested types.
• Parallelism: DuckDB is internally parallel (one connection parallelizes queries), so the lease model could be revisited — but only after verifying DuckDB's JDBC connection is documented thread-safe for concurrent statements.
• Memory accounting: DuckDB has its own memory_limit setting and duckdb_memory() view; MEMORY_USED() is H2-specific.

Marked as v2 candidate, not v1. Requesting an engine that is not on the classpath fails with pipeline.staging.engine_unavailable.

11. H2-Specific Concerns

11.1 SQL compatibility

H2 in MODE=PostgreSQL accepts most PG syntax: :: casts, LIMIT n OFFSET m, array literals (limited), RETURNING, JSON operators (partial). Templates authored for staging (source: "tempdb") should target H2's PG mode syntax, not raw PG syntax. Document this for template authors.

Differences from real PG to be aware of:

• Some PG functions not implemented (e.g., generate_series has limitations).
• Indexes: H2 supports but rarely needed for in-memory staging.
• Stored procedures: not supported in staging (we don't define any).

11.2 Functions available

H2 provides standard SQL functions: COUNT, SUM, AVG, MIN, MAX, COALESCE, CASE WHEN, JOINs (all kinds), window functions, common table expressions (CTEs). All available for staging templates. Note MEMORY_USED() is not used by the staging layer (it is admin-gated in this H2 version, and the operational user is non-admin) — accounting reads JVM heap in-process (§8.2). The admin-only host-reaching functions (FILE_READ/FILE_WRITE/CSVREAD/CSVWRITE/RUNSCRIPT/CREATE ALIAS/LINK_SCHEMA) are not available to author SQL — it runs as a non-admin user (§9.5).

11.3 Known gotchas

• Timestamp arithmetic: H2's PG mode handles INTERVAL differently from PG. Test before relying on it.
• Case sensitivity: H2's native folding is upper-case, PG's is lower-case — which is why the staging URL pins DATABASE_TO_LOWER=TRUE (§3.1): under it H2 lower-folds unquoted identifiers exactly like PG, and unquoted references to the staged lowercase tables and lowercase columns work. (Without the parameter, MODE=PostgreSQL alone left H2 upper-folding and every unquoted reference to a staged table failed 42S03 — the v1.10 fix.) Staged tables are created with quoted identifiers (§4.2), so a mixed-case source column keeps its exact case and template SQL must quote it to match. Template authors should prefer lowercase aliases in source SQL to avoid needing quotes downstream.
• NULL ordering: H2 defaults to NULLS FIRST for ASC; PG defaults to NULLS LAST. Use explicit NULLS FIRST/LAST in ORDER BY for predictable cross-engine behavior.
• H2 built-in table functions keep UPPERCASE result columns even under DATABASE_TO_LOWER=TRUE: SELECT x FROM SYSTEM_RANGE(1, n) fails 42122 because the function's column is X and the unquoted reference folds to x. A tempdb template using a built-in table function must quote the built-in's column (SELECT "X" FROM SYSTEM_RANGE(...)). Measured on the pinned driver (2.3.232); staged tables are unaffected — their columns are lowercase by §4.5.
• A :bind parameter inside a GROUP BY expression breaks expression matching: H2 refuses the query with Column "…" must be in the GROUP BY list (SQLState 90016) even when the GROUP BY expression is textually identical to the SELECT expression — the parameter marker defeats H2's expression-equality check. This bites the natural "classify then group" pattern (GROUP BY CASE WHEN COALESCE(w.prcp_mm,0) >= :rain_threshold_mm THEN ... END). Author it as a derived table: classify in an inner SELECT (parameters fine there — no grouping), then GROUP BY x.weather over the plain column. Measured on the pinned driver (2.3.232), 2026-09-04.
• DECIMAL/DECIMAL arithmetic collapses scale — never divide exact numerics without a DOUBLE cast: SUM(miles)/SUM(seconds)*3600 returned 0 (scale-0 division) where the true value was ~12 mph, 100.0 * tip / fare rounded to one decimal, and borough average fares truncated to integers. Verified empirically: DECIMAL(·,2)/DECIMAL(·,0)*3600 → 2448.00 where CAST(... AS DOUBLE) division → 2456.81 (correct). Every ratio in a tempdb template must cast its operands: CAST(SUM(x) AS DOUBLE) / NULLIF(CAST(SUM(y) AS DOUBLE), 0). Measured on the pinned driver (2.3.232), 2026-09-04.

These are documented in the authoring guide (future).

12. Testing

• Unit tests for H2EgressMapper (every canonical type → correct H2 DDL type string and java.sql.Types code) and H2IngressMapper (every H2 metadata shape → correct ColumnSchema).
• Identifier-safety tests: labels that are empty, 64+ chars, start with a digit, contain a space/quote/semicolon/--, or contain a DROP TABLE payload → all rejected with pipeline.staging.invalid_column_name; case-insensitive duplicates rejected; a valid mixed-case label round-trips with its case preserved. An injection-shaped label must not create, drop, or alter any object.
• Duplicate-table test: staging the same table name twice in one execution → pipeline.staging.table_already_exists, first table's rows intact.
• Integration tests for staging: stage a ResultSet from a mock source, query it back, verify round-trip (type fidelity + row count + value equality).
• Streaming tests: stage 1M rows with limited JVM heap; verify constant transfer memory.
• Concurrency tests (§9.2): two coroutines calling stage() on the same instance simultaneously complete correctly on distinct physical connections, and no physical connection ever has two callers inside it (a per-connection gauge). Real overlap is proved by an event, not a timing: two author blocks meet on a barrier inside a driver call, which only both can reach on two leases — a global lock reintroduced around SQL makes that guard red. The 108 half stays: the two source cursors' read windows must OVERLAP in wall-clock time, at capacity one too. Plus: a cursor blocked in result delivery does not stop an independent operation; a duplicate name raced by two stages has exactly one winner and an intact table; stats() under concurrent writers counts every completed stage once; a table committed by one connection is visible to a read on another.
• Pool tests (H2ConnectionPoolTest): the cap is never exceeded and waiters queue; capacity one makes progress under more ready tasks than dispatcher threads; a waiter cancelled in the queue leaves capacity for the next caller; a dirtied session (schema, search path, variable, local temporary table, time zone, timeouts, isolation, an uncommitted insert) is forcibly reused at capacity one and shown clean with the insert rolled back, not committed; a failed reset discards the connection and replaces it before closing when it is the last; close is idempotent, refuses new leases, never waits for an active lease, and the late return finishes cleanup exactly once; an opener failure returns its permit; no method returns a connection.
• Lifecycle tests beyond §3.4 (H2StagingPoolLifecycleTest): close against an open cursor; creation failure after the operational connection opened destroys the half-built database; a database shut down under a lease fails that lease and close() still does not throw; a partial table whose drop refuses keeps its name owned; a stage cancelled mid-drain at capacity one still rolls back; child rows are pulled between leases, never inside one; two executions cannot see each other's objects; every physical connection is the restricted user.
• Lifecycle tests: after close(), a new lease is refused AND a fresh connection to the same jdbc:h2:mem:exec_{id} URL finds the STAGING_EXEC user gone (SELECT COUNT(*) FROM INFORMATION_SCHEMA.USERS WHERE UPPER(USER_NAME)='STAGING_EXEC' = 0 — case-insensitively, because DATABASE_TO_LOWER=TRUE stores the unquoted-created user as staging_exec; the bare literal would count 0 while the user was alive, exactly the vacuous shape this test exists to rule out) — this is the regression test for DB_CLOSE_DELAY=-1 ever returning. It must key on something the §3.4 cleanup does not remove: an empty fresh database no longer distinguishes "destroyed" from "survived-but-emptied" now that cleanup drops the tables before closing, so the old "sees an empty database" assertion is satisfied even if the DB survived — the user (dropped only when the DB itself dies) is the falsifiable signal. Separately, prove the §3.4 enumerate+DROP TABLE belt actually runs: hold a second peer connection open to the same URL so the DB survives connection.close(), close() the instance, then assert through the peer that the staged tables are gone. Also: close() on a connection already broken does not throw.
• Memory-limit test: stage past a deliberately small max_memory_mb; assert pipeline.staging.memory_limit_exceeded and that the measured in-process JVM-heap reading (not an estimate) drove the decision — with the budget anchored to a measured baseline + headroom, since the reading is JVM-heap-wide (§8.2).
• Type round-trip tests for every canonical type:
  • Source value → staged → queried back → wire-encoded → asserted equal to source.
  • Covers BIGINTEGER, BIGDECIMAL precision, TIMESTAMP UTC normalization, etc.

13. Stability Promise

13.1 Frozen in v1

• The Staging interface and StagingFactory.create(executionId, engine) signature.
• The lifecycle: a bounded pool of non-admin (§9.5) connections, the first opened at execution start and at least one held for the execution, table drop (enumerate + DROP TABLE, §3.4) + close of every connection in finally, no GC reliance, no waiting on a lease still inside the driver.
• The one-owner-per-connection lease model and per-lease session state (§9.2).
• Identifier safety: column-name regex, duplicate rejection, unconditional double-quoting, no sanitizing.
• The canonical → H2 type mapping (per Type System §6).
• Table names = output.table from pipeline nodes (no prefixes).

13.2 Not frozen

• H2 mode (PostgreSQL today, could switch).
• Batch sizes and timeouts (configuration, per Configuration §3.3).
• The default and ceiling of max-connections (configuration; the model is frozen, the number is not — §9.4).
• Memory-polling granularity (per staging operation today; could tighten).
• The H2-specific class names (only the Staging interface is the contract).

14. Open Questions / Future Additions

Out of scope for v1:

• DuckDB as alternative staging: switch on per-pipeline or globally. Useful for analytical workloads.
• Hybrid staging: H2 for small state, DuckDB for large joins. Complex; only if profiling justifies.
• Spill-to-disk: when in-memory limit hit, allow H2 to spill to disk (with severe perf warning) rather than fail. Useful for exploratory queries on large data.
• Indexing hints: let templates declare CREATE INDEX for staging tables to speed up specific JOINs.
• Persistent staging for debugging: opt-in mode where staging is preserved for N minutes after execution so developers can inspect intermediate tables. Note this requires an explicit holder for the connection (or a bounded positive H2 close-delay plus a reaper) — the v1 lifecycle deliberately has no such holder.

Appendix A: Change Log