Loading rows into a table that is already there

README.md

@@ -36,7 +36,7 @@ The rows are an array, so iterating them is `for (const row of rows)` and nothin
 
 ## What works today
 
-`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. Both module formats, typed separately.
+`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. An appender, for loading rows a batch at a time. Both module formats, typed separately.
 
 Build it with `npm run build`, and run the suite with `npm test`. Nothing is published yet, so `npm i zudb` is not a thing you can type at anybody's terminal, but everything it will do is built and installed on every run of the release workflow.
 
@@ -102,6 +102,28 @@ The statements are still the connection's, because the span is the connection's
 
 A block that ends well and forgets to commit loses its work, which is a loud kind of wrong and shows up the first time the code runs. The alternative was a block that failed and kept half of what it did, which is a quiet kind and shows up in production. Committing or rolling back twice is refused as a `ZuUsageError` rather than ignored, since the statements after the first end belong to no transaction of yours. Leaving the block of a transaction whose connection has already been closed says nothing, because a closed connection took the unwritten span with it and there is nothing left to undo.
 
+## Loading a lot of rows
+
+`INSERT` is the wrong shape for loading. Every row is parsed, bound, planned and committed, and the commit is the expensive part, so a million rows is a million commits and the load is spent on durability nobody asked for. An appender is the right shape: rows go into columns in memory, and a flush turns the whole buffer into one commit.
+
+```ts
+await using rows = await conn.appender("Person");
+for (const [id, name] of people) rows.appendRow([id, name]);
+await rows.flush();
+```
+
+A row is every column of the table, in the order the table declares them, and a column is a position rather than a name. Naming the columns per row would cost a lookup per value on the one path where per-value cost is the whole story, and a loader knows its own column order. `appendRows` takes an array of them, which is one check for the batch rather than one per row.
+
+`appendRow` is the one synchronous call in this client, and it is synchronous because it reaches nothing. It converts the values in front of it and pushes them onto a vector, bounded by the width of one row, with no file and no lock at the end of it. Making it a promise would put a microtask between the loop and a memcpy and allocate a million promises to describe work that had already finished. Being synchronous it throws rather than rejecting, with the same `ZuUsageError` everything else here rejects with, so `isZuError(caught)` recognizes it either way. Everything that touches the file, which is `flush`, `close` and the disposal, is a promise like the rest of the client.
+
+What is buffered is typed from the table's own columns, read when the appender opened, so a value that does not belong in a column is refused by the call that appended it rather than a million rows later by the flush that would have carried it. The message names the column and the position: `value 0 of this row is a string and column 'id' of 'Person' holds whole numbers`. A refused row is a row that never happened, so the columns that did take a value give it back and the appender is usable as soon as the caller has fixed the row. In a batch the refusal says which row it was and keeps the ones before it, since nothing here is a transaction until the flush.
+
+`await using rows` flushes. That is the opposite of what a transaction's disposal does, and the two differ because the question differs: a transaction that leaves its scope unfinished is a unit of work nobody completed, and a buffer that leaves its scope unwritten is a loader that read a million rows and threw them away. `discard()` is there for the caller who meant exactly that, and it answers how many rows it dropped.
+
+Two more things are worth knowing before a load. A flush issued while one is still running is refused rather than queued, and so is an append, because waiting for either would be the event loop waiting for a write to disk: `await` the flush. And rows an appender writes are not part of an open transaction, since it writes through the file rather than through the session, so a `ROLLBACK` after a flush does not take them back. A load and a transaction are two different things to reach for.
+
+A rel table has no property columns. A row of one is the two ends of an edge, as offsets into the tables it runs between, so `conn.appender("knows")` takes two columns and the flush checks that both rows are there before it writes anything. That check is here rather than the engine's, because the engine's comes after the write is durable.
+
 ## Asking for numbers instead of bigints
 
 `bigIntMode` says how INT64 is spelled on the way out. It goes on one statement, or on a connection for all of them, and a statement on a connection that named one may still name the other:

bench/append.mjs

@@ -0,0 +1,185 @@
+// What loading rows costs, and what it costs to load them the other way.
+//
+// The appender exists because `INSERT` is the wrong shape for a load:
+// every row is parsed, bound, planned and committed, and the commit is
+// the expensive part. So the first number here is the one to read the
+// rest against, and the ratio between it and the last is the whole
+// argument for the class.
+//
+// The other thing being measured is the boundary itself. `appendRow` is
+// the one synchronous call in this client, and what it does is convert a
+// value per column and push it onto a vector, so its number should be
+// tens of nanoseconds rather than hundreds. When it is not, something on
+// the way in started allocating.
+//
+// Run it against a release build, for the reason bench/query.mjs gives.
+//
+//   npm run build && npm run bench:append
+
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+
+import { connect } from 'zudb'
+
+const ROWS = Number(process.env.ZU_BENCH_ROWS ?? 100_000)
+const REPEATS = Number(process.env.ZU_BENCH_REPEATS ?? 5)
+// How many rows one `INSERT` carries in the batched case. Enough to
+// amortize the commit and small enough that the statement it builds is
+// one a parser can still be asked to read.
+const BATCH = 500
+// One statement per row is slow enough that measuring the whole table
+// that way would dominate the run, so that case is measured over a
+// smaller table and reported per row like the others.
+const SLOW = Number(process.env.ZU_BENCH_SLOW_ROWS ?? 2_000)
+
+const dir = await mkdtemp(join(tmpdir(), 'zu-bench-append-'))
+
+/// A database with the table declared and nothing else in it.
+///
+/// A fresh one per case, because a load into a table that already holds
+/// a million rows is not the same load as one into a table that holds
+/// two, and what is being compared is the way in rather than the size of
+/// what is already there.
+async function blank() {
+  const path = join(dir, `bench-${counter++}.zu1`)
+  const conn = await connect(path)
+  // The declaring insert is written with literals, because that is what
+  // tells the engine what each column holds.
+  await conn.exec("INSERT (p:person {id: 0, name: 'n0'})")
+  return conn
+}
+
+let counter = 0
+
+/// The fastest of `REPEATS` runs, in milliseconds, after one warmup.
+///
+/// The fastest for the reason bench/query.mjs gives: everything that
+/// makes a run slower than the work itself is something that happened to
+/// it rather than something about it.
+async function time(rows, run) {
+  await run(await blank())
+  let best = Infinity
+  for (let round = 0; round < REPEATS; round++) {
+    const conn = await blank()
+    const started = performance.now()
+    await run(conn)
+    best = Math.min(best, performance.now() - started)
+    conn.close()
+  }
+  return best
+}
+
+const cases = [
+  {
+    // One statement per row, which is a parse, a plan and a commit per
+    // row. The number every other line here is asking to be read
+    // against.
+    name: 'INSERT, one row each',
+    rows: SLOW,
+    run: async (conn) => {
+      for (let ix = 1; ix <= SLOW; ix++) {
+        await conn.exec(`INSERT (p:person {id: ${ix}, name: 'n${ix}'})`)
+      }
+    },
+  },
+  {
+    // The same statement carrying five hundred rows, which is what a
+    // loader without an appender ends up writing by hand.
+    name: 'INSERT, 500 rows each',
+    rows: ROWS,
+    run: async (conn) => {
+      for (let start = 1; start <= ROWS; start += BATCH) {
+        const parts = []
+        for (let ix = start; ix < Math.min(start + BATCH, ROWS + 1); ix++) {
+          parts.push(`(p${ix}:person {id: ${ix}, name: 'n${ix}'})`)
+        }
+        await conn.exec(`INSERT ${parts.join(', ')}`)
+      }
+    },
+  },
+  {
+    // Every row buffered and one commit at the end, which is the shape
+    // the class is for.
+    name: 'appender, one flush',
+    rows: ROWS,
+    run: async (conn) => {
+      const rows = await conn.appender('person')
+      for (let ix = 1; ix <= ROWS; ix++) rows.appendRow([BigInt(ix), `n${ix}`])
+      await rows.close()
+    },
+  },
+  {
+    // The same rows with a flush every ten thousand, which is what a
+    // loader that cannot hold the whole file in memory writes. The
+    // difference from the line above is what the extra commits cost.
+    name: 'appender, flush every 10k',
+    rows: ROWS,
+    run: async (conn) => {
+      const rows = await conn.appender('person')
+      for (let ix = 1; ix <= ROWS; ix++) {
+        rows.appendRow([BigInt(ix), `n${ix}`])
+        if (ix % 10_000 === 0) await rows.flush()
+      }
+      await rows.close()
+    },
+  },
+  {
+    // Rows handed over in arrays of a hundred, which is one boundary
+    // crossing per hundred rows rather than one per row. What it saves
+    // is the call and the checks around it, and what it costs is the
+    // arrays.
+    name: 'appender, appendRows(100)',
+    rows: ROWS,
+    run: async (conn) => {
+      const rows = await conn.appender('person')
+      let batch = []
+      for (let ix = 1; ix <= ROWS; ix++) {
+        batch.push([BigInt(ix), `n${ix}`])
+        if (batch.length === 100) {
+          rows.appendRows(batch)
+          batch = []
+        }
+      }
+      if (batch.length) rows.appendRows(batch)
+      await rows.close()
+    },
+  },
+  {
+    // A whole number rather than a `bigint`, which is what a caller
+    // writing row literals writes. It costs a check that the number is
+    // whole and saves whatever the runtime charges for a `bigint`.
+    name: 'appender, number ids',
+    rows: ROWS,
+    run: async (conn) => {
+      const rows = await conn.appender('person')
+      for (let ix = 1; ix <= ROWS; ix++) rows.appendRow([ix, `n${ix}`])
+      await rows.close()
+    },
+  },
+  {
+    // The buffers on their own, with the commit taken out of the
+    // measurement: everything is appended and then thrown away. What is
+    // left is the conversion and the push, which is what `appendRow`
+    // does and all it does.
+    name: 'appender, buffered only',
+    rows: ROWS,
+    run: async (conn) => {
+      const rows = await conn.appender('person')
+      for (let ix = 1; ix <= ROWS; ix++) rows.appendRow([BigInt(ix), `n${ix}`])
+      rows.discard()
+    },
+  },
+]
+
+console.log(`${ROWS} rows, fastest of ${REPEATS}`)
+for (const { name, rows, run } of cases) {
+  const ms = await time(rows, run)
+  const each = (ms * 1e6) / rows
+  const scale = rows === ROWS ? '' : ` (over ${rows})`
+  console.log(
+    `${name.padEnd(26)} ${ms.toFixed(2).padStart(9)} ms  ${Math.round(each).toString().padStart(8)} ns/row${scale}`,
+  )
+}
+
+await rm(dir, { recursive: true, force: true })

binding.cjs

@@ -700,6 +700,7 @@ if (!nativeBinding) {
 }
 
 module.exports = nativeBinding
+module.exports.Appender = nativeBinding.Appender
 module.exports.Connection = nativeBinding.Connection
 module.exports.Transaction = nativeBinding.Transaction
 module.exports.ZuCursor = nativeBinding.ZuCursor

binding.d.cts

@@ -123,6 +123,34 @@ export type ZuParam =
   | ZuParam[]
   | { [field: string]: ZuParam }
 
+/**
+ * A value an appender takes, which is narrower than what a statement
+ * takes.
+ *
+ * A column of an appender has one type, read from the table when the
+ * appender opened, and every value in it is that type. So there is no
+ * `null` here: a column that holds nulls cannot be appended to at all
+ * and the appender says so when it opens, and `undefined` in a row is a
+ * value the caller forgot rather than a null they meant. There are no
+ * lists and no objects either, because a property column holds a scalar.
+ *
+ * BYTES is a `Uint8Array`, which is the one type here that no statement
+ * parameter can be, and INT64 is a `bigint` or a whole `number` below
+ * 2^53. A number past that is refused rather than rounded, because past
+ * 2^53 a number no longer names one integer.
+ */
+export type ZuAppendValue =
+  | boolean
+  | number
+  | bigint
+  | string
+  | Uint8Array
+  | ZuDate
+  | ZuTime
+  | ZuTimestamp
+  | ZuDuration
+  | ZuTemporalValue
+
 /**
  * A walk through the graph: nodes and edges, alternating, a node at
  * each end.
@@ -333,6 +361,105 @@ export interface ZuError extends Error {
   /** The whole line `column` indexes into, for underlining it. */
   readonly excerpt?: string
 }
+/**
+ * Rows on their way into a table, buffered until they are flushed.
+ *
+ * Take one with `Connection.appender`, append rows to it, and close
+ * it. What is buffered is columnar and typed from the table's own
+ * columns, read when the appender opened, so a value that does not
+ * belong in a column is refused by the call that appended it rather
+ * than at the flush that would have carried it, and the message names
+ * the column it did not fit.
+ */
+export declare class Appender {
+  /** The table these rows are going into. */
+  get table(): string
+  /** Rows buffered and not yet written. */
+  get buffered(): number
+  /** Rows this appender has committed, across every flush. */
+  get committed(): number
+  /** Whether this appender has been closed. */
+  get closed(): boolean
+  /**
+   * Appends one row, which is one value per column of the table, in
+   * the order the table declares them.
+   *
+   * Synchronous, and the only synchronous call in this client: the
+   * values go into memory and nothing else happens, so this is a
+   * conversion and a push per column. Being synchronous it throws
+   * rather than rejecting, with the same `ZuUsageError` every other
+   * refusal here carries.
+   *
+   * A row of the wrong width, or with a value that does not fit the
+   * column, is refused with nothing of it kept, so the appender is
+   * still usable once the caller has fixed the row.
+   */
+  appendRow(row: readonly ZuAppendValue[]): void
+  /**
+   * Appends every row of an array of rows.
+   *
+   * The same thing in a loop, and worth a call of its own because it
+   * is one check and one lock for the batch rather than one per row.
+   * A row that is refused stops the call where it was refused and the
+   * rows before it stay buffered: nothing here is a transaction until
+   * the flush, and throwing away work the caller can keep would not
+   * make it one. What it answers is how many rows went in, which is
+   * where a caller who caught the refusal starts again.
+   */
+  appendRows(rows: readonly (readonly ZuAppendValue[])[]): number
+  /**
+   * Writes every buffered row and makes it readable, and answers how
+   * many rows this appender has committed in all.
+   *
+   * One commit, whatever the buffer holds: the values are sealed into
+   * the file, one frame naming them is synced to the log, and the
+   * fold that follows puts them where every query looks. On return
+   * the buffer is empty and the rows are there. A flush with nothing
+   * buffered touches no file, so a loader can flush on a timer
+   * without writing empty commits.
+   *
+   * A flush that fails keeps its rows, so that what did not go in is
+   * still there to be looked at and tried again.
+   */
+  flush(): Promise<number>
+  /**
+   * Flushes what is left and answers how many rows this appender
+   * committed in all.
+   *
+   * Closing twice is not an error and writes nothing the second
+   * time, because an `await using` that closed early would otherwise
+   * fail on the way out.
+   */
+  close(): Promise<number>
+  /**
+   * The close `await using` calls, which is the intended way to
+   * scope an appender.
+   *
+   * It flushes, whether the block ended well or badly, which is the
+   * opposite of what the disposal of a transaction here does and is
+   * the same answer the Python client gives. The two differ because
+   * the question differs: a transaction that leaves its scope
+   * unfinished is a unit of work nobody completed, and a buffer that
+   * leaves its scope unwritten is a loader that read a million rows
+   * and threw them away. A caller who wants the rows gone writes
+   * `discard()` and gets exactly that.
+   *
+   * It is also reachable as `Symbol.asyncDispose`, which is what
+   * `await using` actually looks for and which [`wire_disposal`] puts
+   * on every appender as it is made.
+   */
+  dispose(): Promise<number>
+  /**
+   * Throws away what is buffered and answers how many rows that was.
+   *
+   * The way out of a load that went wrong halfway. A caller who has
+   * noticed that the rows are wrong wants them gone, and closing
+   * would write them. Rows an earlier flush committed are committed,
+   * and this does not reach them.
+   */
+  discard(): number
+}
+
 /**
  * One connection to one database.
  *
@@ -384,6 +511,26 @@ export declare class Connection {
    * commit half of the work of a block that failed.
    */
   transaction(options?: ZuTransactionOptions | null): Promise<Transaction>
+  /**
+   * Opens an appender on `table` and hands it back.
+   *
+   * The bulk-load path. A load written as statements pays a commit
+   * per row, and an appender pays one per flush, which is the whole
+   * difference between loading a million rows in an afternoon and
+   * loading them in a minute.
+   *
+   * ```js
+   * await using rows = await conn.appender('person')
+   * for (const [id, name] of people) rows.appendRow([id, name])
+   * await rows.flush()
+   * ```
+   *
+   * The table has to exist, and its columns are read here, so a
+   * table nothing declares and a column of a type the ingest cannot
+   * carry are both refused at this call rather than at the flush a
+   * million rows later.
+   */
+  appender(table: string): Promise<Appender>
   /**
    * Runs one statement and gives back its rows.
    *