docs: document parallel_copy + full bench benchmark (#300)

CHANGELOG.md

@@ -9,6 +9,21 @@ unreleased. For the forward-looking plan see
 
 ## [Unreleased]
 
+### Added
+
+- `pgcolumnar.parallel_copy(target, path [, workers])` loads a COPY text file into
+  a columnar table with several background workers at once, and returns the row
+  count (#300). Each worker runs core `COPY` over a byte range of the file, so
+  parse and write behavior match `COPY FROM`. The load is atomic through two-phase
+  commit: every worker prepares its transaction, and a coordinator commits them
+  together only when all succeeded, so any failure rolls the whole load back. The
+  target is a single columnar table, where any record-aligned split is correct, or
+  a RANGE-partitioned table with columnar partitions, where the file must be sorted
+  ascending by the partition key and the key type must be numeric or a date/time
+  type. The columnar encode step is CPU bound, so the load scales with worker
+  count up to the physical core count. It landed in two parts, partition-parallel
+  (#323) and single-table (#324). See docs/user-guide.md and docs/benchmarks.md.
+
 ### Changed
 
 - The C standard flag for PostgreSQL 19 is probed rather than hardcoded (#294).

docs/benchmarks.md

@@ -178,6 +178,52 @@ Compression `none` against `zstd`, for the columnar table only: 40 MB against
 5.95 MB. The scan latency does not change, at 0.52 ms against 0.52 ms. The
 encoded stream is already small, and the aggregates do not read it.
 
+## Parallel bulk ingest
+
+`pgcolumnar.parallel_copy` loads a text file with several background workers at
+once. The columnar encode step is CPU bound, so the load speeds up with the worker
+count, up to the physical core count. The bench host has 8 physical cores and 16
+hardware threads.
+
+Method: PostgreSQL 18.4, non-assert, on the bench with 16 vCPU and 62 GB. The
+source file is a 20,000,000-row TSBS cpu slice of 21 columns, sorted by time. Each
+figure is the median of three interleaved rounds, with the file warm in the page
+cache. The baseline is one server-side `COPY`.
+
+Single columnar table, 20,000,000 rows:
+
+| workers | seconds | speedup |
+| --- | --- | --- |
+| 1 (COPY) | 129.8 | 1.00x |
+| 2 | 67.4 | 1.93x |
+| 4 | 36.1 | 3.60x |
+| 8 | 20.6 | 6.29x |
+| 16 | 18.9 | 6.87x |
+
+One worker matches a plain `COPY` at 130.4 s, so the coordinator and the two-phase
+commit add little. The result is the same data every time. All runs load
+20,000,000 rows with an identical `sum(usage_user)`. On-disk size varies by 0.03%
+across worker counts, because the byte split moves a few stripe boundaries.
+
+A 100,000,000-row load shows the same effect at scale. One `COPY` takes 644.1 s;
+`parallel_copy` with 16 workers takes 92.8 s, a 6.94x speedup. Both produce 2.67 GB
+on disk, within 0.004%. The row counts match. The float `sum` matches to nine
+figures and differs in the last, because parallel summation adds in a different
+order.
+
+RANGE-partitioned table, 20,000,000 rows, 24 hourly partitions:
+
+| workers | seconds | speedup |
+| --- | --- | --- |
+| 1 (COPY) | 134.0 | 1.00x |
+| 8 | 29.1 | 4.61x |
+| 16 | 25.4 | 5.27x |
+
+The partitioned path routes each row to its partition and gives each worker a
+distinct partition set. That routing costs a little more than the single-table
+split, so the speedup is lower. It still cuts a two-minute load to under 30
+seconds.
+
 ## Import and export
 
 Export, 6,000,000 rows, 5 columns:

docs/features.md

@@ -134,6 +134,24 @@ coverage.
 - `pgcolumnar.stats(table)` reports per-row-group row counts, deleted-row counts,
   chunk counts, and byte sizes.
 
+## Parallel bulk ingest
+
+- `pgcolumnar.parallel_copy(target, path [, workers])` loads a COPY text file with
+  several background workers at once, as one atomic operation. It returns the row
+  count. The columnar encode step is CPU bound, so more workers give a large load
+  speedup up to the physical core count.
+- Each worker runs core `COPY` over a byte range of the file, so parse and write
+  behavior match `COPY FROM` exactly. There is no second parser to keep correct.
+- The load is atomic. Each worker prepares its transaction, and a coordinator
+  commits them together only when every worker succeeded. Any failure rolls the
+  whole load back, and the target keeps its earlier contents.
+- The target is a single columnar table or a RANGE-partitioned table with
+  columnar partitions. A single table needs no row order. A partitioned target
+  needs the file sorted ascending by the partition key, whose type must be
+  numeric or a date/time type.
+- See the [SQL reference](sql-reference.md#pgcolumnarparallel_copytarget-regclass-filename-text-workers-int-default-null-returns-bigint)
+  and [Benchmarks](benchmarks.md#parallel-bulk-ingest).
+
 ## Interoperability
 
 - Export to Arrow and Parquet: `pgcolumnar.export_arrow(table, path)` and

docs/limitations.md

@@ -183,7 +183,40 @@ the same rows. There is no import-specific overhead; the cost is the columnar
 write path either way.
 
 Work on load throughput is tracked in
-[issue #155](https://github.com/jdatcmd/pgcolumnar/issues/155).
+[issue #155](https://github.com/jdatcmd/pgcolumnar/issues/155). One realized lever
+is [`pgcolumnar.parallel_copy`](#parallel-bulk-ingest). It loads a text file
+across several cores at once.
+
+## Parallel bulk ingest
+
+`pgcolumnar.parallel_copy` loads a text file with several workers at once. It has
+these constraints.
+
+- The file must use COPY text format. The function does not accept CSV or binary
+  format.
+- The target is a single columnar table or a RANGE-partitioned table with
+  columnar partitions. The function rejects any other target, such as a heap
+  table.
+- For a partitioned target the file must be sorted ascending by the partition
+  key. The key type must be numeric or a date/time type. The function reports an
+  error for a text key and for an unsorted file.
+- The caller needs membership in the `pg_read_server_files` role and INSERT on
+  the target. The file is read on the server host.
+- Set `max_prepared_transactions` above the worker count. The load prepares one
+  transaction per worker, and the function errors up front when the setting is
+  too low.
+- The speedup is bounded by the physical core count. The columnar encode step is
+  CPU bound, so workers past the physical cores add little.
+- The load commits on its own. It runs in background workers, so it is not part
+  of the calling transaction. A `ROLLBACK` in the caller does not undo the loaded
+  rows. The atomicity is across the workers, not with the caller.
+- The load is atomic through two-phase commit. A coordinator crash during the
+  final commit step can leave some ranges committed and some prepared. This is
+  the ordinary two-phase-commit in-doubt case. A DBA resolves it from
+  `pg_prepared_xacts`.
+
+See the [SQL reference](sql-reference.md#pgcolumnarparallel_copytarget-regclass-filename-text-workers-int-default-null-returns-bigint)
+and [Benchmarks](benchmarks.md#parallel-bulk-ingest).
 
 ## Planner statistics
 

docs/sql-reference.md

@@ -326,6 +326,44 @@ SELECT pgcolumnar.import_parquet('events_copy', '/tmp/events.parquet');
 SELECT pgcolumnar.import_parquet('events_copy', '/data/events/');
 ```
 
+### pgcolumnar.parallel_copy(target regclass, filename text, workers int DEFAULT NULL) returns bigint
+
+Loads a text file into a columnar table with several background workers at once,
+as one atomic operation. Returns the number of rows loaded. The caller needs
+membership in the `pg_read_server_files` role, which superusers hold, and INSERT
+on the target. The file uses COPY text format. Each worker runs core `COPY` over
+a byte range of the file, so parse and write behavior match `COPY FROM` exactly.
+
+The target may be one of two kinds:
+
+- A single columnar table. The workers write the one table together. Any
+  record-aligned split of the file is correct, so the file needs no ordering.
+- A RANGE-partitioned table whose partitions are columnar. Each worker loads a
+  distinct set of partitions. The file must be sorted ascending by the partition
+  key. The key column may sit anywhere in the row, and its type must be numeric
+  or a date/time type. The function reports an error when the file is not sorted.
+
+The load is atomic. Each worker prepares its transaction rather than committing,
+and a coordinator commits them together only if every worker succeeded. A bad
+row, a full disk, or a constraint failure in any range rolls the whole load back.
+The target keeps its earlier contents. The load runs in background workers, so it
+commits on its own. It is not part of the calling transaction, and a caller
+`ROLLBACK` does not undo it. Set `max_prepared_transactions` above the worker
+count, because the load prepares one transaction per worker.
+
+When `workers` is omitted the function derives a value from the target. For a
+partitioned target it lowers `workers` to the partition count when the count is
+smaller.
+
+```sql
+-- single columnar table, any row order
+CREATE TABLE events (id bigint, ts timestamptz, val double precision) USING pgcolumnar;
+SELECT pgcolumnar.parallel_copy('events', '/data/events.txt', 8);   -- returns row count
+
+-- RANGE-partitioned target, file sorted ascending by the partition key
+SELECT pgcolumnar.parallel_copy('events_by_day', '/data/events_sorted.txt', 8);
+```
+
 ## Reading external Parquet
 
 These read a server-side Parquet file in place, without importing it. They

docs/user-guide.md

@@ -63,6 +63,28 @@ SELECT g, g % 1000, (random() * 100)::numeric(10,2), 'sale',
 FROM generate_series(1, 1000000) g;
 ```
 
+### Parallel bulk load
+
+A single `COPY` uses one core, and the columnar encode step is the largest part
+of a load. To use more cores, load a text file with
+[`pgcolumnar.parallel_copy`](sql-reference.md#pgcolumnarparallel_copytarget-regclass-filename-text-workers-int-default-null-returns-bigint).
+It fans the file across several background workers and returns the row count. The
+load is atomic, so a failure in any part rolls the whole load back. It commits on
+its own, so a `ROLLBACK` in the caller does not undo it.
+
+```sql
+CREATE TABLE events (id bigint, ts timestamptz, val double precision) USING pgcolumnar;
+SELECT pgcolumnar.parallel_copy('events', '/data/events.txt', 8);
+```
+
+The target is either a single columnar table or a RANGE-partitioned table with
+columnar partitions. A single table needs no row order. A partitioned target
+needs the file sorted ascending by the partition key, and the key type must be
+numeric or a date/time type. Set `max_prepared_transactions` above the worker
+count first, because the load prepares one transaction per worker. On a machine
+with many cores this loads several times faster than one `COPY`. See
+[Benchmarks](benchmarks.md#parallel-bulk-ingest) for measured numbers.
+
 ## Query
 
 Queries need no special syntax. The planner adds columnar scan and aggregate

pgcolumnar--1.0-dev.sql

@@ -934,4 +934,4 @@ CREATE FUNCTION pgcolumnar.parallel_copy(target regclass, filename text,
 	AS 'MODULE_PATHNAME', 'columnar_parallel_copy';
 
 COMMENT ON FUNCTION pgcolumnar.parallel_copy(regclass, text, int)
-	IS 'atomic parallel bulk load of a sorted text file into a RANGE-partitioned columnar table, one distinct partition set per worker (#300)';
+	IS 'atomic parallel bulk load of a COPY text file into a columnar table using background workers: a single columnar table (any row order), or a RANGE-partitioned columnar table sorted by the partition key with one distinct partition set per worker (#300)';

src/columnar_parallel_copy.c

@@ -1261,12 +1261,20 @@ columnar_parallel_copy(PG_FUNCTION_ARGS)
 		workers = PCOPY_MAX_WORKERS;
 
 	/*
-	 * The target must be a RANGE-partitioned table: each worker loads a distinct
-	 * partition (distinct storage id), the only shape pgcolumnar allows a parallel
-	 * AND atomic bulk load. Concurrent writers to one non-partitioned table
-	 * serialize on the per-storage write lock and, under 2PC, deadlock; single-table
-	 * parallel load is a planned columnar-core enhancement. Compute partition-
-	 * aligned byte ranges here (this may lower `workers` to the partition count).
+	 * The target is one of two shapes that pgcolumnar can load in parallel AND
+	 * atomically:
+	 *   - a RANGE-partitioned table: each worker loads a distinct partition
+	 *     (distinct storage id), so there is nothing to serialize on. Compute
+	 *     partition-aligned byte ranges (this may lower `workers` to the partition
+	 *     count), which requires the file sorted by the partition key.
+	 *   - a single columnar table: the loaders write the one storage concurrently
+	 *     via columnar_bulk_parallel_writer (see below), so a naive record-aligned
+	 *     byte split is enough and the file needs no ordering.
+	 * Any other target (e.g. a heap, or a partitioned table with non-columnar
+	 * partitions) is rejected. A naive split of one non-partitioned columnar table
+	 * WITHOUT that opt-in would serialize on the per-storage write lock and, under
+	 * 2PC, deadlock -- which is why the single-table path pre-creates the storage
+	 * row and the loaders skip that lock.
 	 */
 	{
 		Relation	target = table_open(relid, AccessShareLock);