diff --git a/CHANGELOG.md b/CHANGELOG.md index 19fd655..9077b5c 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 5194eb0..addbaa1 100644 --- a/docs/benchmarks.md +++ b/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: diff --git a/docs/features.md b/docs/features.md index d14fc0b..ee09626 100644 --- a/docs/features.md +++ b/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 diff --git a/docs/limitations.md b/docs/limitations.md index 3cea82d..2c59a2d 100644 --- a/docs/limitations.md +++ b/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 diff --git a/docs/sql-reference.md b/docs/sql-reference.md index dd543fc..fe4d8fd 100644 --- a/docs/sql-reference.md +++ b/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 diff --git a/docs/user-guide.md b/docs/user-guide.md index 13fc016..8b0f332 100644 --- a/docs/user-guide.md +++ b/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 diff --git a/pgcolumnar--1.0-dev.sql b/pgcolumnar--1.0-dev.sql index be716fd..ef7305d 100644 --- a/pgcolumnar--1.0-dev.sql +++ b/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)'; diff --git a/src/columnar_parallel_copy.c b/src/columnar_parallel_copy.c index 757a1c6..e148b34 100644 --- a/src/columnar_parallel_copy.c +++ b/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);