From 7e077a16322d0f783e29deced27d115503b04d74 Mon Sep 17 00:00:00 2001 From: Alexis Rico Date: Fri, 12 Jun 2026 14:29:16 +0200 Subject: [PATCH 1/3] feat: transparent INSERT into compressed partitions (DML P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the INSERT-side read-only limitation on compressed partitions. INSERTs (direct or parent-routed) land in the partition heap at native heap speed — no decompression, no segment rewrite. Every read path unions the compressed segments with the uncompressed heap tail: - DeltaXDecompress/DeltaXAppend gain a Phase 3 heap-tail scan (leader- only in parallel plans), with the full plan qual re-enabled when a tail exists (heap rows are never batch-filtered). - DeltaXCount/DeltaXMinMax fold loose heap rows into their metadata- derived results at exec time, so those pushdowns stay enabled. - DeltaXAgg bails to plain Agg over heap-tail-aware scans when any involved compressed partition has loose rows (its columnar accumulators cannot ingest row-form tuples); a begin-time guard errors on the stale-cached-plan race instead of dropping rows. - Top-N pushdown and pathkey claims are disabled at plan time when a tail exists; a [-4, flag] custom-private marker + relcache invalidation on the heap's empty->non-empty transition (sent by the leaf trigger's INSERT branch) catch stale cached plans. - INSERT ... ON CONFLICT stays rejected (conflict inference cannot see rows inside segments). UPDATE/DELETE stay rejected unchanged. deltax_compact_partition() folds loose rows into new segments appended to the existing companion tables (colstats/blooms/text_lengths/ valbitmap kept exact or dropped, never under-covering; catalog counters, column_minmax, HLL/ndistinct refreshed) and truncates the loose region; the background worker compacts automatically each cycle. Extracted from a longer perf session (DML P1, validated against a plain-PostgreSQL twin table); UPDATE/DELETE (P2) and tombstones (P2.5) follow in a separate PR. dev/docs/COMPRESSED_DML.md carries the full program design. --- dev/docs/COMPRESSED_DML.md | 555 +++++++++++++++++++ src/catalog.rs | 60 +- src/compress.rs | 935 ++++++++++++++++++++++++++++++++ src/lib.rs | 16 + src/scan/exec/agg/callbacks.rs | 28 + src/scan/exec/count_minmax.rs | 262 +++++++++ src/scan/exec/decompress.rs | 312 ++++++++++- src/scan/exec/segments.rs | 54 ++ src/scan/hook.rs | 202 ++++++- src/scan/json_extract.rs | 14 + src/scan/mod.rs | 16 + src/scan/path.rs | 29 + src/worker.rs | 23 + tests/test_compressed_insert.py | 371 +++++++++++++ tests/test_compression.py | 28 +- 15 files changed, 2870 insertions(+), 35 deletions(-) create mode 100644 dev/docs/COMPRESSED_DML.md create mode 100644 tests/test_compressed_insert.py diff --git a/dev/docs/COMPRESSED_DML.md b/dev/docs/COMPRESSED_DML.md new file mode 100644 index 0000000..316123c --- /dev/null +++ b/dev/docs/COMPRESSED_DML.md @@ -0,0 +1,555 @@ +# Transparent DML on Compressed Partitions + +Status: P1 (INSERT-only, §4) is implemented — transparent INSERT into +compressed partitions, heap-tail union on every read path +(`DeltaXDecompress`/`DeltaXAppend` Phase 3, exec-time folding in +`DeltaXCount`/`DeltaXMinMax`, plan-time gating + stale-plan guard for +`DeltaXAgg`/Top-N/pathkeys), `deltax_compact_partition()` + worker +compaction with valmap append. P1 deviations from this design: `DeltaXAgg` +bails to plain Agg over heap-tail-aware scans instead of ingesting the tail +(steady state in §4.3 is future work), Top-N disables instead of merging +heap candidates, and compaction takes an AccessExclusive lock + TRUNCATE +instead of `FOR UPDATE SKIP LOCKED` (§4.5 step 1 refinement deferred to +P3). Logical-replication origin filtering (§5.6) is not implemented yet. + +P2 (UPDATE/DELETE via option (a), segment-level decompose-on-write, §5) and +P2.5 (tombstone fast layer) are design-only on this branch; their +implementation is being extracted as a follow-up PR that stacks on P1. + +Provenance note: this document covers the whole DML program and was written +on a longer perf-session branch. Some infrastructure it references lives in +sibling PRs and may not be merged yet — dual-mode segment files +(`STORAGE_V2.md`, the `blob_file` catalog column), partition-level bloom +sentinels (`_segment_id = -1` rows, PERF #47), and per-value count +sidecars. Where those are absent the corresponding interaction rules (§4.4 +sentinel fold-in, §6 matrix rows) are vacuously satisfied; whichever PR +lands second must implement them — compaction in `compact_partition_impl` +carries an inline NOTE marking the sentinel fold-in obligation. + +Today compressed partitions are read-only: an ExecutorStart hook +(`deltax_executor_start`, `src/scan/hook.rs:4609`) rejects INSERT/UPDATE/DELETE +when a result relation is a compressed partition, and a belt-and-suspenders +row trigger (`deltax_reject_compressed_partition_dml`, `src/lib.rs:247`, +installed by `install_compressed_dml_trigger`, `src/catalog.rs:1057`) catches +tuple-routed inserts that arrive through the parent. To change one row the +user must `deltax_decompress_partition()` (full round-trip: restore all rows +to the heap, drop all six companion tables, unlink the blob file), modify, +and recompress the whole partition. + +This document designs **transparent DML**: the user runs plain +INSERT/UPDATE/DELETE and the extension does whatever is needed internally, +with full transactional semantics. + +## 1. The mechanics we are working with + +Facts about the current implementation that constrain the design: + +| Component | Current behavior | Relevance | +|---|---|---| +| Partition heap | `TRUNCATE`d at compress time (`compress.rs:980`); assumed empty forever after | The natural landing zone for new/decomposed rows | +| Scan path | `DeltaXDecompress` **replaces** the SeqScan (`path.rs:348` nulls the pathlist) and reads only companion tables; heap rows would be invisible | Must learn to union heap rows | +| `DeltaXAppend` | Bails if any uncompressed partition heap is non-empty (`hook.rs:4556`) — "any uncompressed data would be silently dropped" | Must include per-partition heap tails | +| `DeltaXCount` / `DeltaXMinMax` / `DeltaXAgg` | Computed purely from `_meta._row_count` / `_colstats` min-max / decoded segments | Wrong the moment heap rows exist or rows are logically deleted | +| Segment identity | `_meta._segment_id` is a `SERIAL` PK; one meta row per segment, sidecar rows in `_colstats`, `_blobs`, `_blooms`, `_text_lengths`, `_valbitmap` keyed `(_col_idx, _segment_id)` | A segment is fully described by ordinary, MVCC-visible heap tuples | +| Segment meta scan | Normal `heap_getnext` under `GetActiveSnapshot()` (`segments.rs:2030`) | Deleting a meta row transactionally hides the segment — MVCC for free | +| Blob bytes | TOAST rows in `_blobs`, optionally mirrored in an immutable write-once segment file (`blob_storage = 'dual'`, `segment_file.rs`); per-segment fallback to TOAST when a file index entry is missing | New segments cannot be appended to an existing file; orphaned file entries are harmless | +| Partition bloom sentinels | `_segment_id = -1` rows in `_blooms` (PERF #47). Probe runs as Phase 0pre and, on reject, **skips the colstats probes and meta scan entirely**. Correctness today relies on "companion tables are always built fresh + DML rejected + decompress drops the table" | New segments must be folded into (or invalidate) sentinels; sentinel reject must never skip the heap tail | +| Decompressed / blob caches | Keyed `(companion_oid, segment_id, col_idx)`; invalidation is "companion table is dropped+recreated so keys become unreachable, LRU ages them out" | Holds as long as segment ids are never reused within a companion table's lifetime (SERIAL guarantees this) | +| pg_statistic | Synthesized at compress time from colstats/HLL/valmap; partition has `autovacuum_enabled = off` so PG never overwrites them | Loose rows make stats slightly stale; refresh on compaction | +| Background worker | 60s cycle: drain default, premake, `auto_compress_partitions`, stats merge, auto-drop. Skips replicas. No recompress/merge path exists | The obvious home for compaction | + +## 2. Prior art: TimescaleDB + +TimescaleDB's compressed chunks are the closest production system and validate +the overall shape: + +- **INSERT (2.3+):** rows land in the original uncompressed chunk heap next to + the compressed companion table; the chunk is flagged "partially compressed"; + scans plan an Append of the decompress node plus a plain heap scan; a + recompression job folds loose rows back into batches (segmentwise + recompression decompresses only affected batches). +- **UPDATE/DELETE (2.11+): decompress-on-write.** Segmentby and orderby + min/max predicates locate candidate batches; matching batches are + decompressed into the chunk heap, the compressed batch tuple is **deleted**, + then the ordinary UPDATE/DELETE runs over row-form tuples. Granularity is + the whole ~1,000-row batch. +- **MVCC:** batches are ordinary heap tuples, so delete-batch + insert-rows + + DML is plain PostgreSQL transactionality. Concurrency serializes on the + batch tuple (REPEATABLE READ gets "could not serialize access" on + conflict). A GUC caps tuples decompressed per DML transaction (2.14). +- **Pain points & fixes:** write amplification (1-row update rewrites a + batch), dead-tuple bloat from decompress-then-delete (issue #6196), + tuple filtering during decompression (2.16, up to 500x faster DML), direct + whole-batch DELETE without decompression when predicates provably cover the + batch (2.17/2.21), per-batch bloom pruning for DML (2.27). + +Notably, Timescale did **not** ship delete-bitmaps; they doubled down on +decompose-on-write and spent subsequent releases shrinking how much gets +decomposed. That matches our analysis below. + +## 3. Design overview + +Two cooperating mechanisms, mirroring the storage's existing two-tier shape: + +1. **Loose-row region (INSERT, P1).** The partition heap — empty today — holds + newly inserted rows alongside the segments. Every scan node unions + "segments ∪ heap tail". The worker periodically compacts loose rows into + new segments. +2. **Decompose-on-write (UPDATE/DELETE, P2).** The ExecutorStart interceptor + locates the candidate segments via existing pruning machinery (minmax, + blooms, valbitmap), decompresses *just those segments* into the partition + heap, deletes their meta+sidecar rows, and lets PostgreSQL's normal DML run + over ordinary heap tuples. The worker recompacts later (P3 policy). + +The unifying invariant, which the whole scan/agg architecture already depends +on, is preserved: **a live segment's metadata is exact** — `_row_count`, +colstats min/max/sum, blooms, valbitmaps always describe exactly the rows in +that segment. Rows are either fully in a segment or fully in the heap; never +half-and-half. + +## 4. P1: INSERT into compressed partitions + +### 4.1 Write path + +Cheapest possible: drop the insert rejection (remove INSERT from the row +trigger and from the ExecutorStart check) and let tuple routing put the row in +the partition heap. Indexes defined on the partition are maintained normally +for loose rows. No companion table is touched at insert time. COPY and +multi-row INSERT work unchanged. + +Caveat to document: unique constraints are only enforced among loose rows and +pre-compression data that had index entries — segment rows are not in any +index, so a duplicate of a compressed row is not detected. (Timescale solved +this in 2.11/2.27 with decompress-on-conflict + bloom-assisted checks; that is +explicitly out of scope until P3+.) + +### 4.2 Scan path: the heap tail + +`DeltaXDecompress` gains a **Phase 3 heap tail**: after the last segment, open +the partition heap with `table_beginscan` under the active snapshot and emit +tuples through the existing `ExecQual`/`ExecProject` path (batch quals don't +apply; rows are already row-form — plain qual evaluation is fine for what is +expected to be a small tail). The pathlist replacement at `path.rs:348` stays; +the custom scan simply becomes responsible for both sources. `DeltaXAppend` +does the same per child instead of bailing. + +Doing the union *inside* the node (rather than planning an Append of +CustomScan + SeqScan) is deliberate: every aggregate pushdown node can then +reuse one helper, Top-N can merge heap candidates into its pass-1 candidate +set, and EXPLAIN keeps a single node with a `heap_rows=N` counter. + +Cheapness gate: `relation_heap_is_empty()` (`hook.rs:4507`) already exists and +is the right test — zero heap blocks means Phase 3 is free. Compaction ends +with `TRUNCATE` of the loose region (see 4.5), restoring the zero-block state. + +### 4.3 Aggregate shortcuts + +| Node | Change | +|---|---| +| `DeltaXCount` | `sum(_meta._row_count) + count(heap tail)` — heap counted at exec time under the query snapshot | +| `DeltaXMinMax` | fold heap-tail values into the metadata-derived min/max | +| `DeltaXAgg` | feed heap-tail rows through the existing accumulators after the segment loop (row-form ingest; no vectorization needed for a small tail) | +| Top-N two-pass | pass 1 also collects (heap, ctid, time) candidates; pass 2 fetches winners by ctid | +| `classify_meta_quals` fast paths (Phase 0a/0b) | unchanged for the segment side; heap tail is always scanned regardless of meta-qual classification | + +First implementation may instead *disable* a shortcut when the heap is +non-empty (correct, slower); the table above is the steady state. Disabling is +acceptable only for `DeltaXMinMax`/`DeltaXCount`; `DeltaXAgg` must learn the +heap tail immediately or parent-level aggregates silently regress to row paths +whenever any partition has one loose row. + +### 4.4 Bloom sentinel correctness + +New rule replacing PERF #47's "fresh-build-only" invariant: + +> **Sentinels cover live segments only. A sentinel reject skips the segment +> side (colstats probes + meta scan) but never the heap tail.** + +Phase 0pre therefore short-circuits to Phase 3 instead of returning empty. +Loose rows need no sentinel maintenance at insert time — they are always +scanned. + +When compaction creates new segments (4.5), each affected column's sentinel is +**folded into**, not rebuilt: sentinels store positions as `hash % 2^n` +(OR-halving fold), so inserting new value hashes at `hash % stored_size` +directly is exactly fold-compatible — no false negatives. If the updated +density exceeds `PARTITION_BLOOM_MAX_DENSITY` (0.6) the sentinel row is +deleted (saturated filters prune nothing; same self-pruning rule as build +time). If anything about the fold is in doubt, deleting the sentinel rows for +affected columns is always safe — false positives only. + +### 4.5 Compaction (worker) + +New worker step after `auto_compress_partitions`: for each compressed +partition whose heap is non-empty and crosses a threshold (P3 policy, §8), +in one transaction: + +1. Read loose rows `FOR UPDATE SKIP LOCKED` (rows under concurrent DML stay + loose until the next cycle), sorted by `segment_by + order_by`. +2. Reuse the existing `flush_segment` path to append new segments: fresh + `_segment_id`s from the SERIAL sequence, full sidecar rows (colstats, + blobs, blooms, text_lengths, valbitmap), **TOAST-only** — no blob-file + append (the file is write-once; per-segment TOAST fallback in + `segment_file.rs` already handles file-index misses). +3. Fold new values into partition bloom sentinels (4.4) and append any new + low-cardinality values to `column_valmap` — appending keeps every existing + valbitmap valid (missing trailing bits read as 0 = absent, which is + correct). If a column's value count outgrows the valmap limits, drop its + valbitmap rows + valmap entry. +4. Delete the compacted heap rows. +5. Refresh catalog `row_count`/sizes and re-run + `stats::analyze_partition_from_catalog` so pg_statistic reflects the new + segments; the worker's existing parent-stats merge picks it up. + +Compaction may produce undersized trailing segments; merging small segments is +a P3 concern, not a correctness one. + +## 5. P2: UPDATE / DELETE + +### 5.1 Options considered + +**(a) Segment-level decompose-on-write.** Locate candidate segments via +minmax + blooms + valbitmap; decompress only those into the partition heap; +delete their meta + sidecar rows; let the planned heap scan + ModifyTable +proceed normally. Worker recompacts later. + +**(b) Delete-bitmap sidecar.** A `_delmap(_segment_id, _bits)` companion row +per touched segment; scans AND the bitmap into the selection vector; DELETE = +set bits; UPDATE = set bits + insert new version into the heap. + +| | (a) decompose-on-write | (b) delete bitmap | +|---|---|---| +| MVCC / transactionality | **Free.** Meta-row delete + heap inserts + user DML are one transaction over ordinary tuples; rollback restores the segment untouched | Bitmap row is MVCC-visible, but the *combination* (bitmap version + heap row version) must be read consistently; UPDATE writes two places whose visibility must always agree | +| Scan-path cost | Zero for untouched partitions; touched segments become heap rows until recompaction | Every scan of every segment must probe `_delmap` forever (one PK probe + AND per segment); cost paid by all readers to benefit writers | +| Metadata-exactness invariant | **Preserved** — live segments stay exact | **Broken.** `_row_count`, min/max, `_sum`, `_nonnull_count`, ndistinct, valbitmaps, blooms, text-length sidecars all over-count when bits are set. min/max/sum cannot be "subtracted" without decompressing — so `DeltaXCount`, `DeltaXMinMax`, `DeltaXAgg`, Top-N, the Phase 0a/0b fast paths, and planner stats must all detect marks and bail or consult bitmaps. This is the entire performance architecture of the extension | +| Write amplification | High: 1-row update decomposes a 30,000-row segment (30x Timescale's batch size). Mitigations: candidate pruning usually hits few segments; whole-segment direct DELETE (5.4); future segment split | Low: one bitmap row write per touched segment | +| Concurrency | Serializes on the segment's meta row (delete vs delete); after decompose, normal per-tuple row locks | Serializes on the bitmap row — *every* DML pair touching the same segment conflicts, even on disjoint rows; read-modify-write merge of bits under contention | +| blob_file (storage v2) | File untouched; deleted segment's bytes become unreachable garbage, reclaimed at recompress (file is immutable, holes unsupported — by design) | Same (bitmaps live in a heap table), slight edge | +| Crash safety | WAL'd heap operations only | WAL'd heap operations only | +| Code surface | ExecutorStart interceptor + single-segment decompose (refactor of existing `decompress_partition_inner` loop) + sidecar deletes | New sidecar table + bitmap consult in segment load, batch-qual eval, all five custom nodes, agg pushdown, Top-N, count/minmax, stats synthesis, valbitmap semantics, EXPLAIN | +| Long-term | Matches Timescale's proven trajectory; improvements are localized (prune better, decompose less) | Bitmaps never go away; compaction must additionally rewrite marked segments to reclaim | + +**Recommendation: (a) decompose-on-write.** The deciding factor is the +metadata-exactness invariant: pg_deltax derives far more from per-segment +metadata than a row-store does (count/minmax/agg pushdown, Phase 0 fast +paths, sentinel pruning, synthesized pg_statistic), and option (b) poisons +every one of those paths with a "but check the delete bitmap" qualifier — a +permanent correctness tax on the read side, which is the side this extension +exists to make fast. Option (a) costs more per write but writes to compressed +partitions are, by the product's own definition (compress_after), rare and +cold. Timescale's history is direct evidence that (a) is shippable and that +its weaknesses (amplification) are fixable incrementally. + +### 5.2 Chosen flow + +In `deltax_executor_start`, replace the error path for CMD_UPDATE/CMD_DELETE +(and CMD_INSERT ... ON CONFLICT later) with: + +1. For each compressed result relation (leaf rels appear in + `resultRelations` for both direct and parent-targeted DML; runtime-pruned + leaves that survive are handled identically): +2. Extract pushable predicates from the plan's qual tree — reuse + `extract_segment_filters` + the batch-qual extraction walker against the + ModifyTable child scan's quals. +3. Run the existing Phase 0pre/0a/0b machinery to produce candidate segment + ids: sentinel probe, colstats minmax, per-segment blooms, valbitmap. No + pushable quals ⇒ all segments are candidates (document the perf cliff, add + a `pg_deltax.max_segments_decomposed_per_dml` guard GUC like Timescale's + 2.14 limit, default generous, 0 = unlimited). +4. Under `DML_BYPASS`, for each candidate segment: decompress all columns + (single-segment refactor of the `decompress_partition_inner` per-segment + loop) and insert the rows into the partition heap; then `DELETE FROM + _meta WHERE _segment_id = X` plus the sidecar rows in `_colstats`, + `_blobs`, `_blooms` (segment rows only, not sentinels), `_text_lengths`, + `_valbitmap`. +5. `CommandCounterIncrement()` + `UpdateActiveSnapshotCommandId()` so the + already-taken statement snapshot sees the decomposed rows (the same + mechanism Timescale uses before letting ModifyTable run). +6. Fall through to `standard_ExecutorStart`; the planned SeqScan/IndexScan + over the (formerly empty) heap now finds the rows; UPDATE/DELETE proceeds + with ordinary ctids, firing user triggers, RLS, and RETURNING normally. + +The row-level rejection trigger is removed entirely once this ships (the +interceptor is the single enforcement point; `DML_BYPASS` keeps internal +compress/decompress working during the transition). Prerequisite check: the +pathlist hook must never have replaced a DML target scan with a CustomScan +(CustomScan can't supply target-rel ctids) — today this is moot because DML +was rejected; P2 must add an explicit `root->parse->resultRelation` guard. + +Cross-partition UPDATE (row moves to another partition) works for free: PG +turns it into delete + routed insert, and the routed insert lands in the +target's loose region per P1. + +### 5.3 MVCC, transactionality, crash safety + +All effects — heap inserts of decomposed rows, meta/sidecar deletes, the user +DML itself — are ordinary WAL-logged heap operations in one transaction: + +- **Abort:** every tuple version vanishes; the segment's meta row was never + visibly deleted; concurrent readers never saw an intermediate state. The + blob/decomp caches stay valid because the segment's bytes never changed. +- **Concurrent readers:** a snapshot taken before the DML commits sees the + meta row (segment alive) and not the decomposed heap rows; after commit, the + reverse. No reader can see both (same-transaction visibility) or neither + (the meta delete and heap inserts commit atomically). +- **Crash:** standard WAL recovery; no extension-side state machine. The + immutable blob file is untouched by P2, so there is nothing non-WAL'd to + repair (orphaned file entries are dead weight until recompress). + +### 5.4 Direct whole-segment DELETE (fast path) + +When the statement is a DELETE and the pushed-down predicates provably cover a +candidate segment in full (e.g. time range `[min,max] ⊆` deleted range, with +no residual quals), skip decompose entirely: delete the meta + sidecar rows. +This is Timescale's 2.17/2.21 trick and removes both the amplification and the +dead-tuple bloat for the dominant bulk-retention pattern. `TRUNCATE`/`DROP` of +whole partitions continue to work as today. + +### 5.5 Locking and concurrency + +- Decompose serializes naturally on the `_meta` row delete. In READ COMMITTED, + a second transaction targeting the same segment blocks, then finds the meta + row gone (0 rows from its DELETE) — it must then re-run its candidate scan + (the rows are now loose heap rows; its planned heap scan handles them via + EvalPlanQual like any concurrent-update case). Implementation: perform the + meta delete *first* with `SPI` and treat "0 rows deleted" as "segment + already decomposed by someone else — skip its decompose step". + In REPEATABLE READ/SERIALIZABLE this surfaces as a serialization error, + matching Timescale and vanilla PG semantics. +- Compaction vs concurrent DML: compaction's `FOR UPDATE SKIP LOCKED` read + (4.5) means it never waits on in-flight writers and never compacts a row + mid-update. Compaction vs decompose on the same partition: compaction only + touches loose rows + *inserts* new segments; decompose only deletes + *existing* segments — disjoint except for sentinel-row updates, which both + sides do via short row-level UPDATE/DELETE. A per-partition advisory lock + (`pg_try_advisory_xact_lock`) around compaction keeps the worker from + stacking up behind itself; DML never takes it. + +### 5.6 Replicas + +- **Physical standbys:** every P1/P2 effect is WAL'd heap data; standbys see + consistent state. The blob file is not WAL'd (existing dual-mode caveat); + since decompose never edits the file and compaction writes TOAST-only, the + standby's TOAST fallback keeps working unchanged. The worker already skips + replicas, so no compaction runs there. +- **Logical replication:** loose-row INSERTs and user UPDATE/DELETEs replicate + as normal row changes — strictly better than today. Two hazards: + 1. *Decompose noise:* the decomposed-row inserts + meta deletes are internal. + Companion tables are typically not in the publication (Scenario 1 in + `tests/test_logical_replication.py`), so meta deletes don't leak; but the + decomposed heap-row INSERTs *do* replicate, followed by the user DML — + the subscriber (which holds the rows uncompressed or compressed on its + own schedule) would duplicate them. + 2. *Compaction deletes:* compaction's heap-row DELETEs would erase + subscriber rows — same hazard class as the compress-time TRUNCATE that + Scenario 1 already excludes from publication. + Mitigation for both: maintenance/decompose transactions set a replication + origin (`deltax_internal`); Scenario-1 subscriptions use `origin = NONE` + (PG16+) so origin-tagged changes are filtered while user DML (origin-less) + flows. Scenario 2 (companion schema replicated wholesale) is consistent + as-is — both sides see the same meta/sidecar/heap mutations — but the + decomposed-row inserts must then *not* be origin-filtered; the two scenarios + need distinct documented publication recipes, and + `test_logical_replication.py` grows cases for DML-on-compressed under each. + +## 6. Sidecar / cache interaction matrix + +| Component | INSERT (P1) | Decompose (P2) | Compaction (P3) | +|---|---|---|---| +| `_meta` row counts | untouched; `DeltaXCount` adds heap count at exec | rows deleted with segment — counts stay exact | new exact rows | +| `_colstats` min/max/sum | untouched | deleted with segment | new exact rows | +| Per-segment blooms / text-lengths / valbitmap | untouched | deleted with segment | new rows per new segment | +| Partition bloom sentinels | none (heap tail always scanned) | none (over-coverage ⇒ false positives only — safe) | fold-in new hashes; drop row if density > 0.6 | +| `column_valmap` / HLL / MCV catalog stats | stale w.r.t. loose rows (acceptable; planner-only) | stale (over-estimate; safe) | valmap appended (old bitmaps stay valid); stats re-synthesized | +| pg_statistic | slightly stale; partition autovacuum stays off | slightly stale | `analyze_partition_from_catalog` re-run + parent merge | +| Decomp/blob caches `(oid, seg_id, col)` | unaffected | dead ids unreachable, LRU evicts; **invariant: segment ids never reused** (SERIAL — holds) | new ids; no invalidation needed | +| `MAPPED_FILE_CACHE` / blob file | unaffected | file untouched; deleted segments = dead bytes until recompress | new segments TOAST-only; file untouched | +| Full recompress (`deltax_compress_partition` after decompress) | unchanged: still drops + recreates everything, rewrites blob file, rebuilds sentinels — the GC of last resort | | | + +## 7. Phasing + +**P1 — INSERT-only (~2–3 weeks).** Remove INSERT rejection; heap tail in +`DeltaXDecompress`/`DeltaXAppend`; extend `DeltaXCount`/`DeltaXMinMax`/ +`DeltaXAgg`/Top-N (or gate the first two); sentinel covers-segments-only rule; +worker compaction reusing `flush_segment` + sentinel fold-in + valmap append + +stats refresh; integration tests incl. logical replication recipes. Biggest +user value (late-arriving data into compressed history) for the smallest +mechanism. + +**P2 — UPDATE/DELETE via decompose-on-write (~3–4 weeks).** ExecutorStart +interceptor; qual reuse for candidate location; single-segment decompose +refactor; meta-delete-first concurrency protocol; CCI/snapshot dance; +`max_segments_decomposed_per_dml` GUC; whole-segment direct DELETE; remove the +rejection trigger; correctness suite (concurrent DML, rollback, RR/serializable, +RETURNING, triggers, cross-partition UPDATE). + +**P3 — recompaction policy (~2 weeks).** Trigger thresholds: loose rows ≥ +`segment_size`, or ≥ N% of partition row_count, or oldest loose row older than +a `compact_after` interval — whichever first. Churn guard: skip partitions +with write activity in the last cycle (hysteresis), so steady writers aren't +compacted into a stream of tiny segments that immediately re-fragment; merge +undersized segments opportunistically; blob-file garbage ratio metric + +recommendation to recompress. Later/optional: unique-constraint conflict +checking against segments (bloom-assisted), direct-compress on insert. + +## 8. Open questions + +- Should Phase 3 heap-tail rows flow through batch quals via a row-form shim + for very large tails (e.g. right after a big decompose), or is plain + `ExecQual` always fine? Measure before optimizing. +- Decompose emits up to 30k rows × N segments into the heap inside the user's + transaction — memory is bounded (per-segment batching exists in the + decompress path) but dead-tuple bloat after the DML commits argues for + compaction prioritizing recently decomposed partitions. +- `ON CONFLICT` / UPSERT semantics on compressed partitions: reject until P3+, + or silently treat segments as conflict-invisible? Rejecting is honest; + silent is wrong. Reject. +- Parallel-worker scans (`pg_deltax.parallel_workers`): heap tail should be + scanned by exactly one worker; simplest is leader-only Phase 3. + +--- + +## P2.5 — Tombstone fast layer (added 2026-06-12, product requirement) + +**Requirement (Alexis):** DML latency on compressed partitions must be +near-indistinguishable from normal Postgres. Decompose-on-write alone +cannot meet this: even a perfectly-pruned single-row UPDATE decodes a +~30K-row segment (~50–300 ms vs ~1 ms native). + +**Architecture:** synchronous tombstones-as-rows + asynchronous rewrite. + +- DELETE ⇒ insert `(segment_id, row_offset)` into a per-partition + `_tombstones` companion table (ordinary heap rows ⇒ MVCC, rollback, + replication all native). UPDATE ⇒ tombstone + heap-insert of the new + version. O(1) per affected row, single-digit ms. +- Scan paths: segments with zero tombstones (steady state) are + untouched — one indexed existence probe gates this. Tombstoned + segments filter rows by offset (exact) or bail to the row path. +- Metadata fast paths (count/minmax/agg/valcounts): bail or subtract + exact per-segment tombstone counts; never approximate. +- The P1/P3 compaction worker physically rewrites tombstone-bearing + segments (decompose → drop dead rows → recompress), restoring + pristine fast-path state; P2's decompose machinery is the tool. +- Why this differs from the REJECTED option (b) bitmap blobs: blobs + broke MVCC/exactness; tombstone *rows* are MVCC-native and the + zero-tombstone gate keeps the steady-state read path tax at one + index probe per scanned partition (cacheable alongside the heap + -emptiness check P1 added). + +Sequencing: P2 (decompose-on-write) lands first as the correctness +foundation + compactor tool; P2.5 then makes the synchronous path +fast; P3 policy work follows. + +### P2.5 implementation status (DELETE-only tombstones) + +Implemented in the follow-up DML P2/P2.5 PR for **DELETE only**; UPDATE +stays on P2 decompose-on-write. +Honesty check outcome: a tombstone-fast UPDATE must still materialize the +old row versions into the heap for the executor to apply SET expressions / +fire triggers / serve RETURNING, so its synchronous cost is dominated by +decoding **all** columns of the candidate segments and inserting rows — the +same decode the decompose performs. The clean incremental win there +(restore only the matching rows instead of the whole segment) is real but +bounded well below 10x for selective updates and adds a second restore +path; it is deferred. DELETE needs **no materialization at all** — only the +qual columns are decoded once to identify matching offsets — which is where +the order-of-magnitude win lives. + +**Write path** (`deltax_executor_start` → `claim_segments_for_tombstone` / +`insert_dml_tombstones`, offsets via `dml_matching_offsets`): + +- Eligibility (per statement): DELETE; no RETURNING; no user row DELETE + triggers (same observation-free guards as the §5.4 whole-segment drop); + the collected scan quals are PROVABLY the complete predicate AND every + qual was batch-extracted (`all_quals_handled` — the same exactness + contract production scans rely on when they null `ps.qual`); no BOOL + ordering quals (the shared evaluator degrades those to equality). + Anything else falls back to P2 decompose — never to skipping. +- Whole-segment drops (§5.4) still take precedence for AllPass segments; + tombstones cover the partially-matching candidates. +- Per surviving candidate: decode ONLY the qual columns (blob cache + applies), evaluate text quals explicitly + the shared + `evaluate_batch_quals` (NULL fails, SQL semantics), collect matching + offsets; segments with zero matches are left untouched (pruning false + positives now cost a qual-column decode, not a decompose). +- Concurrency: `SELECT ... FOR UPDATE` on the candidates' `_meta` rows + (ordered by segment_id) — concurrent decompose/compaction of those + segments waits for our commit and then excludes our tombstones; segments + already claimed by a concurrent decompose are absent and skipped (the P2 + "0 rows claimed" rule). Readers never block (no AccessExclusive on this + path). Tombstone inserts use `ON CONFLICT DO NOTHING`: a row concurrently + tombstoned elsewhere is skipped and not double-counted (READ COMMITTED + "already deleted" semantics; RR surfaces a serialization error). +- Command tag: inserted-tombstone count folds into `es_processed` via the + same ExecutorRun hook mechanism as whole-segment drops. +- Catalog `row_count` semantics (decision): it counts **live rows stored in + segments** and stays exact — decremented by the tombstone insert in the + same transaction (MVCC-rollback-safe), so the `DeltaXCount` catalog + shortcut and synthesized planner stats need no tombstone awareness. + +**Storage**: per-partition `_deltax_compressed._tombstones +(_segment_id int, _row_offset int, PRIMARY KEY(_segment_id, _row_offset))`. +Created EAGERLY (empty) at compress time so the OWNER/GRANT cascade covers +it (deviation from "lazily on first tombstone": permission-safety; an empty +table has zero blocks = zero read cost, which is what lazy bought), plus +`CREATE TABLE IF NOT EXISTS` at first tombstone for pre-P2.5 partitions. +Ordinary heap rows ⇒ MVCC / rollback / physical & logical replication all +native. + +**Read side** — steady-state gate is `companion_may_have_tombstones` +(syscache probe + relcache nblocks, the same physical-truth trick as P1's +`relation_heap_is_empty`); exact map loaded only when the gate fires, +under the scan's active snapshot (atomic with the `_meta` scan): + +- `load_segments_heap` (and the Phase 0a point-lookup path, and parallel + workers rebuilding segments from the DSM wire) attach per-segment sorted + offset lists to `SegmentData.tombstones`. +- Decode paths (`DeltaXDecompress`/`DeltaXAppend` main scan + all three + Top-N variants) seed their selection vectors with + `tombstone_preselection` — tombstoned rows are filtered exactly during + decode; Top-N/pathkeys stay enabled (removing rows preserves order). +- `DeltaXCount`: catalog path exact by construction (see above); meta-scan + fallback sums `live_row_count()` (exact: that path is only planned when + quals provably cover every row of surviving segments). +- `DeltaXMinMax` and `DeltaXAgg` (metadata fast paths, AllPass shortcuts, + valcounts GROUP BY) describe physical rows → they BAIL: plan-time gates + (`any_compressed_rte_heap_nonempty` extended + tombstone check before + `add_minmax_path`) push those queries to the row path while tombstones + exist; exec-time stale-plan guards error on the residual race (same + contract as P1's heap-tail guard; the tombstone writer fires a relcache + invalidation so cached plans replan). Compaction restores the pushdowns. +- Sentinels/blooms/valbitmaps/colstats/pg_statistic: tombstones only ever + REMOVE rows, so all existing structures over-cover → false positives + only → safe unchanged (§6 column for tombstones ≙ the Decompose column). + +**Drive-by fix uncovered by the P2.5 tests**: the per-segment / partition +bloom probes hashed the RAW PG-epoch datum for timestamp/date equality +constants while the build side hashes Unix-epoch-encoded values — every +segment was falsely bloom-rejected and `ts_col = const` (or `IN`) returned +zero rows on compressed data. `bloom_probe_encode` (segments.rs) now +converts probe constants into the build domain; `point_ts` joined the +twin-equality query set as the regression net. + +**Rewrite**: every rematerialization path excludes tombstoned rows and +consumes their tombstone rows — full `deltax_decompress_partition`, P2 +decompose (`restore_segment_rows(skip_offsets)`), and compaction. +`deltax_compact_partition()` + the worker (`auto_compact_partitions`, now +also triggered by a non-empty tombstones table) decompose tombstone-bearing +segments minus dead rows, recompress the survivors into fresh segments, and +`TRUNCATE` the tombstones table back to the zero-block steady state (also +clearing dead pages from rolled-back tombstone DML). + +Measured (local Docker arm64, one partition, 60k rows in 2×30k-row +segments with an int + text payload, `\timing`): + +| single-row DELETE | latency | +|------------------------------|----------| +| plain PostgreSQL (indexed) | 0.42 ms | +| P2.5 tombstone (first) | 0.78 ms | +| P2.5 tombstone (warm caches) | 0.53 ms | +| P2 decompose (same predicate, forced via RETURNING) | 91.2 ms | + +≈118–173x faster than decompose and within ~2x of native PostgreSQL — +the "near-indistinguishable" requirement. Guarded in CI by +`tests/test_compressed_dml.py::TestTombstoneDelete:: +test_single_row_delete_latency` (asserts >10x vs decompose). diff --git a/src/catalog.rs b/src/catalog.rs index 41fdd43..761fa0d 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -408,6 +408,31 @@ pub fn mark_partition_compressed( Ok(()) } +/// Add a compaction batch's row count and compressed size to a partition's +/// counters. Unlike `mark_partition_compressed` (full rebuild) this is +/// incremental — used when loose heap rows are folded into new segments. +pub fn bump_partition_compaction( + client: &mut SpiClient, + partition_id: i32, + added_rows: i64, + added_compressed_size: i64, +) -> spi::SpiResult<()> { + client.update( + "UPDATE deltax.deltax_partition + SET row_count = COALESCE(row_count, 0) + $1, + compressed_size = COALESCE(compressed_size, 0) + $2, + compressed_at = now() + WHERE id = $3", + None, + &[ + added_rows.into(), + added_compressed_size.into(), + partition_id.into(), + ], + )?; + Ok(()) +} + /// Write a pre-computed per-column ndistinct map (typically from /// HLL sketches merged across segments during compression) to the /// `deltax.deltax_partition.column_ndistinct` JSONB column. @@ -1016,11 +1041,44 @@ pub fn mark_partition_decompressed( Ok(()) } +/// Per-row helper for the compressed-partition DML trigger's INSERT branch. +/// +/// On the heap's empty → non-empty transition (the BEFORE INSERT trigger for +/// the first loose row still sees zero blocks) this sends a relcache +/// invalidation for the partition, so cached plans in *all* backends that +/// were built assuming an empty heap (metadata-only aggregate pushdowns, +/// pathkey-claiming sorted scans, Top-N pushdowns) are replanned against the +/// new loose-row state. Subsequent rows see `nblocks > 0` and do nothing, so +/// the per-row cost is one lseek. +#[pg_extern] +fn deltax_note_compressed_insert(relid: pg_sys::Oid) -> bool { + unsafe { + let rel = pg_sys::table_open(relid, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + let nblocks = + pg_sys::RelationGetNumberOfBlocksInFork(rel, pg_sys::ForkNumber::MAIN_FORKNUM); + pg_sys::table_close(rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + if nblocks == 0 { + pg_sys::CacheInvalidateRelcacheByRelid(relid); + } + } + true +} + +/// SQL-visible probe for the backend-local DML bypass flag, used by the +/// compressed-partition trigger to let internal maintenance DML (compaction's +/// delete of just-compacted loose rows) through while still rejecting user +/// UPDATE/DELETE. +#[pg_extern] +fn deltax_dml_bypass_active() -> bool { + crate::scan::dml_bypass_active() +} + /// Install the DML rejection trigger on a compressed leaf partition. /// /// Parent-table INSERTs are routed to the leaf after ExecutorStart, so the /// result-relation hook cannot reliably catch them. A row trigger on the leaf -/// fires after tuple routing and blocks both direct and routed DML. +/// fires after tuple routing; since P1 it lets INSERTs through (loose rows) +/// and blocks UPDATE/DELETE. pub fn install_compressed_dml_trigger( client: &mut SpiClient, schema: &str, diff --git a/src/compress.rs b/src/compress.rs index 146228f..574f6bb 100644 --- a/src/compress.rs +++ b/src/compress.rs @@ -4253,6 +4253,941 @@ pub fn auto_compress_partitions(client: &mut SpiClient<'_>, ht: &catalog::Deltat compressed } +// ============================================================================ +// P1 compaction: fold loose heap rows (transparent INSERTs into compressed +// partitions) into new segments appended to the existing companion tables. +// See dev/docs/COMPRESSED_DML.md §4.5. +// ============================================================================ + +/// Compact the loose heap rows of a compressed partition into new segments. +/// +/// Loose rows are INSERTs that arrived after the partition was compressed +/// (P1 transparent DML). They live in the partition heap and are unioned +/// with the segment data at scan time; compaction moves them into proper +/// segments so the metadata-only fast paths re-engage. +#[pg_extern] +fn deltax_compact_partition(partition: &str) -> String { + Spi::connect_mut(|client| compact_partition_impl(client, partition)) +} + +/// RAII guard for the backend-local DML/planner bypass flag. While held, +/// the set_rel_pathlist / upper-paths hooks plan queries against the plain +/// heap (no DeltaX custom scans) and the DML rejection (trigger + executor +/// hook) is disabled. Drop-reset so an error inside compaction can't leave +/// the backend planning every later query without DeltaX paths. +struct DmlBypassGuard; + +impl DmlBypassGuard { + fn new() -> Self { + crate::scan::set_dml_bypass(true); + DmlBypassGuard + } +} + +impl Drop for DmlBypassGuard { + fn drop(&mut self) { + crate::scan::set_dml_bypass(false); + } +} + +/// Rewrite a companion `CREATE TABLE` DDL into `CREATE TABLE IF NOT EXISTS`. +/// Compaction may need to create sidecar tables that the original +/// compression skipped (e.g. `_blooms` when the GUC was off, `_text_lengths` +/// when there were no text columns yet). +fn ddl_if_not_exists(ddl: &str) -> String { + ddl.replacen("CREATE TABLE ", "CREATE TABLE IF NOT EXISTS ", 1) +} + +/// True when the relation named by `fqn` (a quoted, schema-qualified name) +/// exists. +fn relation_exists(client: &SpiClient, fqn: &str) -> bool { + client + .select("SELECT to_regclass($1) IS NOT NULL", Some(1), &[fqn.into()]) + .ok() + .and_then(|r| r.first().get_one::().ok().flatten()) + .unwrap_or(false) +} + +pub(crate) fn compact_partition_impl(client: &mut SpiClient, partition: &str) -> String { + let (schema, part_table) = crate::partition::resolve_relation(client, partition); + let part_info = catalog::get_partition_by_name(client, &schema, &part_table) + .expect("failed to query partition") + .unwrap_or_else(|| { + pgrx::error!( + "pg_deltax: partition {}.{} not found in catalog", + schema, + part_table + ) + }); + + if !part_info.is_compressed { + return format!( + "Partition {}.{} is not compressed; nothing to compact", + schema, part_table + ); + } + + let ht = catalog::get_deltatable_by_id(client, part_info.deltatable_id) + .expect("failed to query deltatable") + .unwrap(); + + let part_fqn = crate::partition::fqn(&schema, &part_table); + + // Everything below reads the partition through plain heap plans (the + // custom scans would union the segments back in — double-counting) and + // deletes the compacted rows past the DML rejection. The guard resets + // the flag on every exit path, including errors. + let _bypass = DmlBypassGuard::new(); + + // Serialize against concurrent writers: the read → flush → delete cycle + // below requires a stable visible row set. The lock is acquired BEFORE + // the row-reading snapshot, so in READ COMMITTED every committed loose + // row is visible and TRUNCATE at the end cannot destroy unseen data. + client + .update( + &format!("LOCK TABLE {} IN ACCESS EXCLUSIVE MODE", part_fqn), + None, + &[], + ) + .expect("failed to lock partition for compaction"); + + let loose_rows: i64 = client + .select( + &format!("SELECT count(*) FROM ONLY {}", part_fqn), + None, + &[], + ) + .expect("failed to count loose rows") + .first() + .get_one::() + .ok() + .flatten() + .unwrap_or(0); + if loose_rows == 0 { + return format!( + "Partition {}.{} has no loose rows to compact", + schema, part_table + ); + } + + // Schema-drift guard: appending segments assumes the companion layout + // (col_idx assignment) still matches the live table shape. Refuse when + // the parent's column shape changed since compression — the safe path + // is a full decompress + recompress. + let cc_now = catalog::snapshot_compressed_columns( + client, + &ht.schema_name, + &ht.table_name, + &ht.segment_by, + ) + .expect("failed to snapshot column shape"); + let cc_now_val: serde_json::Value = + serde_json::from_str(&cc_now).unwrap_or(serde_json::Value::Null); + let shape_matches = + matches!(&part_info.compressed_columns, Some(stored) if *stored == cc_now_val); + if !shape_matches { + pgrx::error!( + "pg_deltax: partition {}.{} has loose rows but its column shape changed since compression; run deltax_decompress_partition() + deltax_compress_partition() instead", + schema, + part_table + ); + } + + let columns = get_column_metadata( + client, + &schema, + &part_table, + &ht.segment_by, + &ht.time_column, + ht.json_extract.as_ref(), + ); + if columns.is_empty() { + pgrx::error!("pg_deltax: no columns found for {}.{}", schema, part_table); + } + + let ddl = build_companion_ddl(&part_table, &columns); + if !relation_exists(client, &ddl.meta_fqn) { + pgrx::error!( + "pg_deltax: companion meta table missing for compressed partition {}.{}", + schema, + part_table + ); + } + + // New segment ids continue after the existing maximum (sentinel rows in + // _blooms use -1 and never appear in _meta). + let max_segment_id: i32 = client + .select( + &format!("SELECT COALESCE(MAX(_segment_id), 0) FROM {}", ddl.meta_fqn), + None, + &[], + ) + .expect("failed to read max segment id") + .first() + .get_one::() + .ok() + .flatten() + .unwrap_or(0); + let mut next_segment_id: i32 = max_segment_id + 1; + let first_new_segment_id = next_segment_id; + + let segment_size = ht.segment_size as usize; + let kinds: Vec = columns + .iter() + .map(|c| classify_column(&c.data_type, c.is_segment_by)) + .collect(); + + let select_cols = columns + .iter() + .zip(kinds.iter()) + .map(|(c, kind)| { + if c.is_segment_by || matches!(kind, ColumnKind::Text) { + format!("\"{}\"::text", c.name) + } else { + format!("\"{}\"", c.name) + } + }) + .collect::>() + .join(", "); + + let order_clause = if !ht.segment_by.is_empty() { + let mut order_parts = Vec::new(); + for s in &ht.segment_by { + order_parts.push(format!("\"{}\"", s)); + } + for o in &ht.order_by { + order_parts.push(format!("\"{}\"", o)); + } + format!(" ORDER BY {}", order_parts.join(", ")) + } else { + String::new() + }; + + let order_col_indices: Vec = ht + .order_by + .iter() + .filter_map(|name| columns.iter().position(|c| c.name == *name)) + .collect(); + let seg_col_indices: Vec = columns + .iter() + .enumerate() + .filter(|(_, c)| c.is_segment_by) + .map(|(i, _)| i) + .collect(); + + let cursor_sql = format!( + "DECLARE compact_cursor CURSOR FOR SELECT {} FROM ONLY {}{}", + select_cols, part_fqn, order_clause + ); + client + .update(&cursor_sql, None, &[]) + .expect("failed to declare compaction cursor"); + let batch_size = segment_size; + let fetch_sql = format!("FETCH {} FROM compact_cursor", batch_size); + + let mut typed_cols = init_typed_columns(&columns, &kinds); + let mut current_seg_values: Vec> = Vec::new(); + let mut rows_in_segment: usize = 0; + let mut total_compressed_size: i64 = 0; + let mut total_rows: i64 = 0; + let mut blob_buffer: Vec<(u16, i32, Vec)> = Vec::new(); + let mut bloom_buffer: Vec<(u16, i32, u8, Vec)> = Vec::new(); + let mut colstats_buffer: Vec = Vec::new(); + let mut text_length_buffer: Vec<(u16, i32, Vec)> = Vec::new(); + let mut valbitmap_value_buffer: Vec<(u16, i32, SegValueCounts)> = Vec::new(); + let num_nonseg_cols = columns.iter().filter(|c| !c.is_segment_by).count(); + // Batch-only HLL — merged into the persisted partition sketches below. + let mut batch_hll: Vec> = (0..num_nonseg_cols) + .map(|_| CardinalityEstimator::::new()) + .collect(); + let mut batch_topvals: TopVals = (0..num_nonseg_cols) + .map(|_| std::collections::HashMap::new()) + .collect(); + + loop { + let result = client + .select(&fetch_sql, None, &[]) + .expect("failed to fetch from compaction cursor"); + let fetched = result.len(); + if fetched == 0 { + break; + } + let tuptable_to_free = unsafe { pg_sys::SPI_tuptable }; + + for row in result { + if !seg_col_indices.is_empty() { + let row_seg_values: Vec> = seg_col_indices + .iter() + .map(|&i| { + row.get_datum_by_ordinal(i + 1) + .unwrap() + .value::() + .unwrap() + }) + .collect(); + if current_seg_values.is_empty() { + current_seg_values = row_seg_values; + } else if row_seg_values != current_seg_values { + if rows_in_segment > 0 { + total_compressed_size += flush_with_splitting( + client, + &ddl.meta_fqn, + &ddl.colstats_fqn, + &columns, + &typed_cols, + ¤t_seg_values, + rows_in_segment, + segment_size, + &mut next_segment_id, + &mut blob_buffer, + &mut bloom_buffer, + &mut colstats_buffer, + &mut text_length_buffer, + &mut valbitmap_value_buffer, + &mut batch_hll, + &mut batch_topvals, + ); + typed_cols = init_typed_columns(&columns, &kinds); + rows_in_segment = 0; + } + current_seg_values = row_seg_values; + } + } + + append_row_to_columns(&row, &columns, &kinds, &mut typed_cols); + rows_in_segment += 1; + total_rows += 1; + + if rows_in_segment >= segment_size { + if seg_col_indices.is_empty() { + sort_typed_columns(&mut typed_cols, &order_col_indices, rows_in_segment); + } + total_compressed_size += flush_with_splitting( + client, + &ddl.meta_fqn, + &ddl.colstats_fqn, + &columns, + &typed_cols, + ¤t_seg_values, + rows_in_segment, + segment_size, + &mut next_segment_id, + &mut blob_buffer, + &mut bloom_buffer, + &mut colstats_buffer, + &mut text_length_buffer, + &mut valbitmap_value_buffer, + &mut batch_hll, + &mut batch_topvals, + ); + typed_cols = init_typed_columns(&columns, &kinds); + rows_in_segment = 0; + } + } + + if !tuptable_to_free.is_null() { + unsafe { pg_sys::SPI_freetuptable(tuptable_to_free) }; + } + if fetched < batch_size { + break; + } + } + + if rows_in_segment > 0 { + if seg_col_indices.is_empty() { + sort_typed_columns(&mut typed_cols, &order_col_indices, rows_in_segment); + } + total_compressed_size += flush_with_splitting( + client, + &ddl.meta_fqn, + &ddl.colstats_fqn, + &columns, + &typed_cols, + ¤t_seg_values, + rows_in_segment, + segment_size, + &mut next_segment_id, + &mut blob_buffer, + &mut bloom_buffer, + &mut colstats_buffer, + &mut text_length_buffer, + &mut valbitmap_value_buffer, + &mut batch_hll, + &mut batch_topvals, + ); + } + + client + .update("CLOSE compact_cursor", None, &[]) + .expect("failed to close compaction cursor"); + + if total_rows == 0 { + return format!( + "Partition {}.{} has no loose rows to compact", + schema, part_table + ); + } + + // ---- Append sidecar rows to the TOAST-backed companion tables. ---- + if !colstats_buffer.is_empty() { + colstats_buffer.sort_by_key(|r| (r.col_idx, r.segment_id)); + for chunk in colstats_buffer.chunks(100) { + let values: Vec = chunk + .iter() + .map(|r| { + let min_str = r.min_val.map_or("NULL".to_string(), |v| v.to_string()); + let max_str = r.max_val.map_or("NULL".to_string(), |v| v.to_string()); + let sum_str = r.sum_val.as_deref().unwrap_or("NULL"); + format!( + "({}, {}, {}, {}, {}, {}, {}, {})", + r.col_idx, + r.segment_id, + min_str, + max_str, + sum_str, + r.nonnull_count, + r.nonzero_count, + r.ndistinct + ) + }) + .collect(); + let sql = format!( + "INSERT INTO {} (_col_idx, _segment_id, _min, _max, _sum, _nonnull_count, _nonzero_count, _ndistinct) VALUES {}", + ddl.colstats_fqn, + values.join(", ") + ); + client + .update(&sql, None, &[]) + .expect("failed to insert compaction colstats"); + } + } + + if !blob_buffer.is_empty() { + client + .update(&ddl_if_not_exists(&ddl.blobs_ddl), None, &[]) + .expect("failed to ensure blobs table"); + blob_buffer.sort_by_key(|&(col_idx, seg_id, _)| (col_idx, seg_id)); + for (col_idx, seg_id, blob) in blob_buffer { + use pgrx::datum::DatumWithOid; + let insert_sql = format!( + "INSERT INTO {} (_col_idx, _segment_id, _data) VALUES ($1, $2, $3)", + &ddl.blobs_fqn + ); + let args: Vec = vec![ + (col_idx as i16).into(), + seg_id.into(), + DatumWithOid::from(blob), + ]; + client + .update(&insert_sql, None, &args) + .expect("failed to insert compaction blob"); + } + } + + if !bloom_buffer.is_empty() { + client + .update(&ddl_if_not_exists(&ddl.blooms_ddl), None, &[]) + .expect("failed to ensure blooms table"); + bloom_buffer.sort_by_key(|&(col_idx, seg_id, _, _)| (col_idx, seg_id)); + for (col_idx, seg_id, num_hashes, bloom_bytes) in bloom_buffer { + use pgrx::datum::DatumWithOid; + let insert_sql = format!( + "INSERT INTO {} (_col_idx, _segment_id, _num_hashes, _data) VALUES ($1, $2, $3, $4)", + &ddl.blooms_fqn + ); + let args: Vec = vec![ + (col_idx as i16).into(), + seg_id.into(), + (num_hashes as i16).into(), + DatumWithOid::from(bloom_bytes), + ]; + client + .update(&insert_sql, None, &args) + .expect("failed to insert compaction bloom"); + } + } + + if !text_length_buffer.is_empty() { + client + .update(&ddl_if_not_exists(&ddl.text_lengths_ddl), None, &[]) + .expect("failed to ensure text_lengths table"); + text_length_buffer.sort_by_key(|&(col_idx, seg_id, _)| (col_idx, seg_id)); + for (col_idx, seg_id, blob) in text_length_buffer { + use pgrx::datum::DatumWithOid; + let insert_sql = format!( + "INSERT INTO {} (_col_idx, _segment_id, _data) VALUES ($1, $2, $3)", + &ddl.text_lengths_fqn + ); + let args: Vec = vec![ + (col_idx as i16).into(), + seg_id.into(), + DatumWithOid::from(blob), + ]; + client + .update(&insert_sql, None, &args) + .expect("failed to insert compaction text length sidecar"); + } + } + + // NOTE for partition-level bloom sentinels (PERF #47, not yet on this + // branch): compaction appends segments to a live companion table, so if + // `_segment_id = -1` sentinel rows ever exist they MUST be fold-merged + // with the batch's value hashes or deleted here — a sentinel that + // under-covers any live segment produces wrong query results. + + // ---- Valbitmap: keep every existing column's bitmap regime exact, or + // drop it (absence is safe — segments just stop being pruned on that + // column). ---- + let new_segment_ids: Vec = (first_new_segment_id..next_segment_id).collect(); + append_valbitmaps_for_compaction( + client, + part_info.id, + &ddl.valbitmap_fqn, + &columns, + &valbitmap_value_buffer, + &new_segment_ids, + ); + + // ---- Catalog refresh ---- + catalog::bump_partition_compaction(client, part_info.id, total_rows, total_compressed_size) + .expect("failed to bump partition counters"); + + // Partition-level [min,max] map (Phase -1 pruning) — recomputed from the + // full colstats table, which now includes the new segments. CORRECTNESS- + // critical: new values may extend the old ranges. + catalog::update_partition_column_minmax(client, part_info.id, &ddl.colstats_fqn, &columns) + .expect("failed to refresh partition column_minmax"); + + // ndistinct/HLL: merge the batch sketches into the persisted ones + // (planner-only; best-effort). + merge_batch_hll_into_catalog(client, part_info.id, &columns, &batch_hll); + + // ---- Remove the compacted loose rows. READ COMMITTED: TRUNCATE is safe + // (the AEL was taken before the reading snapshot, so the visible set is + // complete) and restores the zero-block state that re-enables every + // metadata fast path. Under REPEATABLE READ / SERIALIZABLE the reading + // snapshot may predate the lock, so DELETE only what we actually read + // (rows invisible to the snapshot survive for a later compaction). ---- + let read_committed = + unsafe { pg_sys::XactIsoLevel } < pg_sys::XACT_REPEATABLE_READ as std::ffi::c_int; + if read_committed { + client + .update(&format!("TRUNCATE ONLY {}", part_fqn), None, &[]) + .expect("failed to truncate compacted loose rows"); + } else { + client + .update(&format!("DELETE FROM ONLY {}", part_fqn), None, &[]) + .expect("failed to delete compacted loose rows"); + } + + // ANALYZE the touched companion tables and refresh pg_statistic / + // reltuples for the partition (best-effort). + let _ = client.update(&format!("ANALYZE {}", ddl.meta_fqn), None, &[]); + let _ = client.update(&format!("ANALYZE {}", ddl.colstats_fqn), None, &[]); + let msg = analyze_partition_impl_split(client, &schema, &part_table); + pgrx::debug1!("pg_deltax compact stats refresh: {}", msg); + + // Make the change visible to cached plans in every backend: TRUNCATE + // already swaps the relfilenode (relcache inval), but the DELETE path + // needs an explicit invalidation; sending it twice is harmless. + let part_oid: pg_sys::Oid = client + .select(&format!("SELECT '{}'::regclass::oid", part_fqn), None, &[]) + .ok() + .and_then(|r| r.first().get_one::().ok().flatten()) + .unwrap_or(pg_sys::InvalidOid); + if part_oid != pg_sys::InvalidOid { + unsafe { pg_sys::CacheInvalidateRelcacheByRelid(part_oid) }; + } + crate::scan::invalidate_compressed_cache(); + + format!( + "Compacted {}.{}: {} loose rows into {} new segment(s)", + schema, + part_table, + total_rows, + next_segment_id - first_new_segment_id, + ) +} + +/// Keep the per-segment value-presence bitmaps exact across a compaction. +/// +/// For every column present in the catalog `column_valmap`: +/// - if every new segment produced a complete value set and the union of +/// old + new values still fits the bitmap budget, append the novel +/// values to the END of the valmap (existing bitmaps stay valid — the +/// scan probes positions linearly and missing trailing bits read as 0) +/// and insert a bitmap row for each new segment; +/// - otherwise DROP the column's bitmap regime entirely (delete its +/// valbitmap rows + valmap/valcounts entries). Absence is safe; an +/// incomplete valmap is not (`prune_all` would wrongly prune segments +/// containing a new value). +fn append_valbitmaps_for_compaction( + client: &mut SpiClient, + partition_id: i32, + valbitmap_fqn: &str, + columns: &[ColumnMeta], + value_buffer: &[(u16, i32, SegValueCounts)], + new_segment_ids: &[i32], +) { + use std::collections::HashMap; + + // Load the current valmap / valcounts from the catalog. + let (valmap_text, valcounts_text): (Option, Option) = { + let row = client + .select( + "SELECT column_valmap::text, column_valcounts::text + FROM deltax.deltax_partition WHERE id = $1", + Some(1), + &[partition_id.into()], + ) + .expect("failed to read partition valmap"); + let first = row.first(); + (first.get(1).ok().flatten(), first.get(2).ok().flatten()) + }; + let mut valmap: HashMap> = valmap_text + .as_deref() + .and_then(|t| serde_json::from_str(t).ok()) + .unwrap_or_default(); + if valmap.is_empty() { + // No bitmap regime on any column — nothing to maintain. + return; + } + let mut valcounts: ColumnValcounts = valcounts_text + .as_deref() + .and_then(|t| serde_json::from_str::(t).ok()) + .map(|v| { + let mut out: ColumnValcounts = HashMap::new(); + if let serde_json::Value::Object(cols) = v { + for (name, counts) in cols { + if let serde_json::Value::Object(m) = counts { + let list: Vec<(String, i64)> = m + .into_iter() + .filter_map(|(k, c)| c.as_i64().map(|n| (k, n))) + .collect(); + out.insert(name, list); + } + } + } + out + }) + .unwrap_or_default(); + + // col_idx ↔ name mapping (non-segment-by enumeration order). + let mut idx_to_name: HashMap = HashMap::new(); + let mut name_to_idx: HashMap<&str, u16> = HashMap::new(); + { + let mut ci: u16 = 0; + for col in columns { + if col.is_segment_by { + continue; + } + idx_to_name.insert(ci, col.name.as_str()); + name_to_idx.insert(col.name.as_str(), ci); + ci += 1; + } + } + + let mut changed = false; + let col_names: Vec = valmap.keys().cloned().collect(); + for col_name in col_names { + let Some(&col_idx) = name_to_idx.get(col_name.as_str()) else { + // Column no longer exists in the layout (shouldn't happen — the + // shape guard upstream rejects drift). Drop defensively. + drop_valbitmap_column( + client, + valbitmap_fqn, + None, + &mut valmap, + &mut valcounts, + &col_name, + ); + changed = true; + continue; + }; + + // Collect this column's per-new-segment value sets. + let entries: Vec<&(u16, i32, SegValueCounts)> = value_buffer + .iter() + .filter(|(ci, _, _)| *ci == col_idx) + .collect(); + + // Completeness: every new segment must have produced a value set + // (compute_segment_valbitmap_values omits segments whose local + // distinct count overflowed — their values are unknown). + if entries.len() != new_segment_ids.len() { + drop_valbitmap_column( + client, + valbitmap_fqn, + Some(col_idx), + &mut valmap, + &mut valcounts, + &col_name, + ); + changed = true; + continue; + } + + // Union check + novel-value collection. + let existing = valmap.get(&col_name).cloned().unwrap_or_default(); + let existing_set: std::collections::HashSet<&str> = + existing.iter().map(|s| s.as_str()).collect(); + let mut novel: std::collections::BTreeSet = std::collections::BTreeSet::new(); + for (_, _, vals) in &entries { + for (v, _) in vals.iter() { + if !existing_set.contains(v.as_str()) { + novel.insert(v.clone()); + } + } + } + if existing.len() + novel.len() > VALBITMAP_MAX_DISTINCT { + drop_valbitmap_column( + client, + valbitmap_fqn, + Some(col_idx), + &mut valmap, + &mut valcounts, + &col_name, + ); + changed = true; + continue; + } + + // Append novel values (sorted among themselves) to the END of the + // valmap — existing bitmaps stay valid. + let mut updated = existing.clone(); + updated.extend(novel.iter().cloned()); + let bit_idx: HashMap<&str, u8> = updated + .iter() + .enumerate() + .map(|(i, v)| (v.as_str(), i as u8)) + .collect(); + + // Ensure the bitmap table exists (it might not when the original + // compress produced no bitmap rows but the catalog valmap exists — + // defensive; normally both exist together). + let n_bytes = updated.len().div_ceil(8); + for (_, seg_id, vals) in &entries { + let mut bits: Vec = vec![0; n_bytes.max(1)]; + for (v, _) in vals.iter() { + if let Some(&bi) = bit_idx.get(v.as_str()) { + bits[(bi / 8) as usize] |= 1u8 << (bi % 8); + } + } + use pgrx::datum::DatumWithOid; + let args: Vec = vec![ + (col_idx as i16).into(), + (*seg_id).into(), + DatumWithOid::from(bits), + ]; + client + .update( + &format!( + "INSERT INTO {} (_col_idx, _segment_id, _bits) VALUES ($1, $2, $3)", + valbitmap_fqn + ), + None, + &args, + ) + .expect("failed to insert compaction valbitmap row"); + } + + // Merge counts into valcounts (stats-only). + if !novel.is_empty() || !entries.is_empty() { + let counts = valcounts.entry(col_name.clone()).or_default(); + for (_, _, vals) in &entries { + for (v, c) in vals.iter() { + if let Some(slot) = counts.iter_mut().find(|(k, _)| k == v) { + slot.1 += *c as i64; + } else { + counts.push((v.clone(), *c as i64)); + } + } + } + } + if !novel.is_empty() { + valmap.insert(col_name.clone(), updated); + } + changed = true; + } + + if changed { + catalog::update_partition_column_valmap(client, partition_id, &valmap) + .expect("failed to update partition column_valmap"); + catalog::update_partition_column_valcounts(client, partition_id, &valcounts) + .expect("failed to update partition column_valcounts"); + } +} + +/// Remove one column's bitmap regime: its valbitmap rows (when the col_idx +/// is known and the table exists) plus the valmap/valcounts entries. +fn drop_valbitmap_column( + client: &mut SpiClient, + valbitmap_fqn: &str, + col_idx: Option, + valmap: &mut std::collections::HashMap>, + valcounts: &mut ColumnValcounts, + col_name: &str, +) { + if let Some(ci) = col_idx + && relation_exists(client, valbitmap_fqn) + { + client + .update( + &format!("DELETE FROM {} WHERE _col_idx = $1", valbitmap_fqn), + None, + &[(ci as i16).into()], + ) + .expect("failed to delete valbitmap rows"); + } + valmap.remove(col_name); + valcounts.remove(col_name); +} + +/// Merge the compaction batch's per-column HLL sketches into the persisted +/// partition sketches and refresh `column_ndistinct`. Planner-only data — +/// every failure path silently keeps the (slightly stale) existing values. +fn merge_batch_hll_into_catalog( + client: &mut SpiClient, + partition_id: i32, + columns: &[ColumnMeta], + batch_hll: &[CardinalityEstimator], +) { + let stored: Option = client + .select( + "SELECT column_hll::text FROM deltax.deltax_partition WHERE id = $1", + Some(1), + &[partition_id.into()], + ) + .ok() + .and_then(|r| r.first().get_one::().ok().flatten()); + let Some(stored) = stored else { return }; + let Ok(serde_json::Value::Object(map)) = serde_json::from_str::(&stored) + else { + return; + }; + + let nd_col_names: Vec<&str> = columns + .iter() + .filter(|c| !c.is_segment_by) + .map(|c| c.name.as_str()) + .collect(); + + let mut merged: std::collections::HashMap> = + std::collections::HashMap::new(); + for (name, val) in map { + if let Ok(sketch) = serde_json::from_value::>(val) { + merged.insert(name, sketch); + } + } + let mut nd_updates: std::collections::HashMap = std::collections::HashMap::new(); + for (name, sketch) in nd_col_names.iter().zip(batch_hll.iter()) { + if let Some(existing) = merged.get_mut(*name) { + existing.merge(sketch); + nd_updates.insert((*name).to_string(), existing.estimate() as i64); + } + } + if nd_updates.is_empty() { + return; + } + + // Refresh column_ndistinct: existing values for untouched columns, + // merged estimates for the rest. + let existing_nd: std::collections::HashMap = client + .select( + "SELECT column_ndistinct::text FROM deltax.deltax_partition WHERE id = $1", + Some(1), + &[partition_id.into()], + ) + .ok() + .and_then(|r| r.first().get_one::().ok().flatten()) + .and_then(|t| serde_json::from_str(&t).ok()) + .unwrap_or_default(); + let mut nd_final = existing_nd; + for (k, v) in nd_updates { + nd_final.insert(k, v); + } + let _ = catalog::update_partition_column_ndistinct_from_map(client, partition_id, &nd_final); + + // Re-serialize the merged sketches. + let names: Vec<&str> = merged.keys().map(|s| s.as_str()).collect(); + let sketches: Vec> = names + .iter() + .map(|n| merged.get(*n).cloned().unwrap()) + .collect(); + if let Some(json) = serialize_partition_hll(&names, &sketches) { + let _ = catalog::update_partition_column_hll(client, partition_id, &json); + } +} + +/// Worker entry point: compact every compressed partition of `ht` whose heap +/// has accumulated loose rows. Returns the number of partitions compacted. +/// Each partition runs inside a subtransaction (`PgTryBuilder`) so a lock +/// timeout or transient failure on one partition doesn't abort the worker's +/// whole maintenance cycle. +pub fn auto_compact_partitions(client: &mut SpiClient<'_>, ht: &catalog::DeltatableInfo) -> i32 { + let candidates = client + .select( + "SELECT p.schema_name, p.table_name + FROM deltax.deltax_partition p + WHERE p.deltatable_id = $1 + AND p.is_compressed + AND COALESCE( + pg_catalog.pg_relation_size( + pg_catalog.to_regclass( + pg_catalog.format('%I.%I', p.schema_name, p.table_name))), + 0) > 0", + None, + &[ht.id.into()], + ) + .expect("failed to query compaction candidates"); + + let mut targets: Vec<(String, String)> = Vec::new(); + for row in candidates { + let sch: Option = row.get(1).ok().flatten(); + let name: Option = row.get(2).ok().flatten(); + if let (Some(sch), Some(name)) = (sch, name) { + targets.push((sch, name)); + } + } + if targets.is_empty() { + return 0; + } + + // Don't let the AccessExclusive lock stall the worker behind long + // queries; skip locked partitions until the next cycle. + let _ = client.update("SET LOCAL lock_timeout = '5s'", None, &[]); + + let mut compacted = 0; + for (sch, name) in &targets { + let qualified = format!("{}.{}", sch, name); + // `&mut SpiClient` is not UnwindSafe; PgTryBuilder runs the closure + // in a subtransaction and rolls it back on error, which is exactly + // the recovery the SPI stack needs — assert the boundary. + let client_cell = std::panic::AssertUnwindSafe(&mut *client); + let ok = pgrx::PgTryBuilder::new(move || { + let client_cell = client_cell; + let msg = compact_partition_impl(client_cell.0, &qualified); + pgrx::log!("pg_deltax: {}", msg); + true + }) + .catch_others(|_e| { + pgrx::log!( + "pg_deltax: compaction of a partition was skipped (lock timeout or error); will retry next cycle" + ); + false + }) + .execute(); + if ok { + compacted += 1; + } + } + + let _ = client.update("SET LOCAL lock_timeout = 0", None, &[]); + + compacted +} + + /// Re-populate pg_class.reltuples + pg_statistic for an already-compressed /// partition. `stats::analyze_partition_from_catalog` reads the /// authoritative per-column distinct counts persisted at compression time diff --git a/src/lib.rs b/src/lib.rs index 912bf81..0352dc8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -209,6 +209,22 @@ RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN + -- P1 transparent DML: INSERTs land in the partition heap (the "loose + -- row" region) and are unioned with segment data at scan time. The + -- helper sends a relcache invalidation on the empty->non-empty + -- transition so cached plans that assumed an empty heap get replanned. + IF TG_OP = 'INSERT' THEN + PERFORM deltax.deltax_note_compressed_insert(TG_RELID); + RETURN NEW; + END IF; + -- Internal maintenance (compaction deleting just-compacted loose rows) + -- runs with the DML bypass flag set; let it through. + IF deltax.deltax_dml_bypass_active() THEN + IF TG_OP = 'DELETE' THEN + RETURN OLD; + END IF; + RETURN NEW; + END IF; RAISE EXCEPTION 'cannot % compressed partition "%.%", decompress it first', TG_OP, TG_TABLE_SCHEMA, diff --git a/src/scan/exec/agg/callbacks.rs b/src/scan/exec/agg/callbacks.rs index 9c7521c..7f32b7e 100644 --- a/src/scan/exec/agg/callbacks.rs +++ b/src/scan/exec/agg/callbacks.rs @@ -323,6 +323,31 @@ pub(crate) unsafe extern "C-unwind" fn create_agg_scan_state( } } +/// P1 heap-tail stale-plan guard. The planner never emits a DeltaXAgg path +/// when any involved compressed partition holds loose heap rows (transparent +/// INSERTs); the insert trigger sends a relcache invalidation on the +/// empty→non-empty transition so cached plans replan. This guard catches the +/// residual race (rows committed between plan revalidation and snapshot): +/// DeltaXAgg's columnar accumulators cannot ingest row-form heap tuples, so +/// silently proceeding would drop rows from the aggregate. Erroring is the +/// correct-by-construction fallback; a retry replans onto the slow path. +fn error_if_compressed_heap_tail(companion_oids: &[pg_sys::Oid]) { + for &oid in companion_oids { + let part_oid = super::super::segments::partition_oid_for_companion(oid); + if part_oid == pg_sys::InvalidOid { + pgrx::error!( + "pg_deltax: cannot resolve partition for companion OID {} (catalog inconsistency)", + u32::from(oid) + ); + } + if !unsafe { crate::scan::relation_heap_is_empty(part_oid) } { + pgrx::error!( + "pg_deltax: a compressed partition gained uncompressed rows after this plan was created; retry the query" + ); + } + } +} + /// BeginCustomScan callback for DeltaXAgg: decompress and aggregate. #[pg_guard] pub(crate) unsafe extern "C-unwind" fn begin_agg_scan( @@ -360,6 +385,7 @@ pub(crate) unsafe extern "C-unwind" fn begin_agg_scan( if plan.companion_oids.is_empty() { pgrx::error!("pg_deltax: DeltaXAgg has no companion tables"); } + error_if_compressed_heap_tail(&plan.companion_oids); reset_scan_buf_stats(); let ctx = build_agg_exec_context_from_plan(plan); let state = build_deferred_agg_state(ctx, /* is_worker */ false); @@ -384,6 +410,8 @@ pub(crate) unsafe extern "C-unwind" fn begin_agg_scan( pgrx::error!("pg_deltax: DeltaXAgg has no companion tables"); } + error_if_compressed_heap_tail(&plan.companion_oids); + let (meta, metadata_us) = load_agg_metadata_from_plan(&plan.companion_oids); // Fast path 1: answer from catalog metadata (no segment scan at all) diff --git a/src/scan/exec/count_minmax.rs b/src/scan/exec/count_minmax.rs index fc7d84e..e8efc6a 100644 --- a/src/scan/exec/count_minmax.rs +++ b/src/scan/exec/count_minmax.rs @@ -72,6 +72,120 @@ unsafe fn rehydrate_segment_filters( } } +/// P1 heap tail: scan the live heap tuples of every compressed partition +/// backing `companion_oids` whose heap is non-empty, evaluate the full +/// rehydrated qual list against each tuple, and invoke `f(slot)` for each +/// qualifying row. The metadata-only aggregates (DeltaXCount / DeltaXMinMax) +/// fold these loose rows into their catalog/colstats-derived results so +/// transparent INSERTs into compressed partitions stay visible. +/// +/// The qual list is the query's full WHERE clause (the planner only emits +/// these paths when every qual classified as time/segment bounds, which are +/// plain `Var op Const` expressions), evaluated with `ExecQual` against a +/// virtual slot holding the deformed heap tuple. Var attnos index the +/// partition layout positionally — guarded by a natts check against +/// `expected_natts` (`0` skips the check, for COUNT(*) which reads no +/// columns... but layout drift would still poison qual evaluation, so +/// callers should pass the companion column count whenever known). +unsafe fn for_each_heap_tail_row( + node: *mut pg_sys::CustomScanState, + companion_oids: &[pg_sys::Oid], + qual_bytes: &[u8], + expected_natts: usize, + mut f: impl FnMut(*mut pg_sys::TupleTableSlot), +) { + unsafe { + // Cheap gate: collect only partitions with a non-empty heap. + let mut heap_oids: Vec = Vec::new(); + for &oid in companion_oids { + let part_oid = super::segments::partition_oid_for_companion(oid); + if part_oid == pg_sys::InvalidOid { + pgrx::error!( + "pg_deltax: cannot resolve partition for companion OID {} (catalog inconsistency)", + u32::from(oid) + ); + } + if !crate::scan::relation_heap_is_empty(part_oid) { + heap_oids.push(part_oid); + } + } + if heap_oids.is_empty() { + return; + } + + // Build an ExprState for the full qual list. Vars with ordinary + // varnos read from econtext->ecxt_scantuple. + let qual_state: *mut pg_sys::ExprState = if qual_bytes.is_empty() { + std::ptr::null_mut() + } else { + let cstr = std::ffi::CString::new(qual_bytes.to_vec()).unwrap(); + let qual_list = pg_sys::stringToNode(cstr.as_ptr()) as *mut pg_sys::List; + pg_sys::ExecInitQual(qual_list, &mut (*node).ss.ps as *mut pg_sys::PlanState) + }; + let econtext = (*node).ss.ps.ps_ExprContext; + + let snapshot = pg_sys::GetActiveSnapshot(); + for &oid in &heap_oids { + let rel = pg_sys::table_open(oid, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + let tupdesc = (*rel).rd_att; + let natts = (*tupdesc).natts as usize; + if expected_natts != 0 && natts != expected_natts { + let name_ptr = pg_sys::get_rel_name((*rel).rd_id); + let name = if name_ptr.is_null() { + format!("OID {}", u32::from((*rel).rd_id)) + } else { + std::ffi::CStr::from_ptr(name_ptr) + .to_string_lossy() + .into_owned() + }; + pg_sys::table_close(rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + pgrx::error!( + "pg_deltax: compressed partition \"{}\" has uncompressed rows but its physical layout ({} attrs) does not match the compressed layout ({} cols); decompress and recompress it", + name, + natts, + expected_natts, + ); + } + let slot = pg_sys::MakeSingleTupleTableSlot(tupdesc, &pg_sys::TTSOpsVirtual); + let flags: u32 = pg_sys::ScanOptions::SO_TYPE_SEQSCAN + | pg_sys::ScanOptions::SO_ALLOW_STRAT + | pg_sys::ScanOptions::SO_ALLOW_SYNC + | pg_sys::ScanOptions::SO_ALLOW_PAGEMODE; + let scan = (*(*rel).rd_tableam).scan_begin.unwrap()( + rel, + snapshot, + 0, + std::ptr::null_mut(), + std::ptr::null_mut(), + flags, + ); + loop { + let tuple = pg_sys::heap_getnext(scan, pg_sys::ScanDirection::ForwardScanDirection); + if tuple.is_null() { + break; + } + pg_sys::ExecClearTuple(slot); + pg_sys::heap_deform_tuple(tuple, tupdesc, (*slot).tts_values, (*slot).tts_isnull); + pg_sys::ExecStoreVirtualTuple(slot); + if !qual_state.is_null() { + (*econtext).ecxt_scantuple = slot; + let passes = pg_sys::ExecQual(qual_state, econtext); + pg_sys::MemoryContextReset((*econtext).ecxt_per_tuple_memory); + if !passes { + continue; + } + } + f(slot); + } + if let Some(end) = (*(*rel).rd_tableam).scan_end { + end(scan); + } + pg_sys::ExecDropSingleTupleTableSlot(slot); + pg_sys::table_close(rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + } + } +} + /// State for DeltaXCount (COUNT(*) pushdown). pub(crate) struct CountScanState { pub(crate) total_count: i64, @@ -244,6 +358,33 @@ pub(super) unsafe extern "C-unwind" fn begin_count_scan( } }; + // P1 heap tail: add the loose rows sitting in the partition heaps + // (transparent INSERTs after compression), filtered through the + // full qual list. Zero-cost when every heap is empty. + let mut state = state; + { + let expected_natts = if qual_bytes.is_empty() { + 0 // COUNT(*) without quals reads no columns + } else { + super::segments::load_metadata_cached(companion_oids[0]) + .col_names + .len() + }; + let t_heap = Instant::now(); + let mut heap_rows: i64 = 0; + for_each_heap_tail_row( + node, + &companion_oids, + &qual_bytes, + expected_natts, + |_slot| { + heap_rows += 1; + }, + ); + state.total_count += heap_rows; + state.heap_scan_us += t_heap.elapsed().as_micros() as u64; + } + let state_box = Box::new(state); let state_ptr = Box::into_raw(state_box); (*node).custom_ps = state_ptr as *mut pg_sys::List; @@ -631,6 +772,127 @@ pub(super) unsafe extern "C-unwind" fn begin_minmax_scan( } total_segments += segs.len() as u64; } + + // P1 heap tail: fold loose heap rows (transparent INSERTs into + // compressed partitions) into the metadata-derived accumulators. + // Zero-cost when every involved partition heap is empty. + for_each_heap_tail_row( + node, + &companion_oids, + &qual_bytes, + meta.col_names.len(), + |slot| { + let values = (*slot).tts_values; + let nulls = (*slot).tts_isnull; + for (spec_idx, spec) in agg_specs.iter().enumerate() { + let attno = spec.varattno; + let (d, isnull) = if attno >= 1 { + ( + *values.add((attno - 1) as usize), + *nulls.add((attno - 1) as usize), + ) + } else { + (pg_sys::Datum::from(0usize), true) + }; + match &mut accs[spec_idx] { + Acc::MinMax { + datum, + null, + type_oid, + is_min, + } => { + if isnull { + continue; + } + let ct = if *type_oid != pg_sys::InvalidOid { + *type_oid + } else { + spec.col_type_oid + }; + if !matches!( + ct, + pg_sys::INT2OID + | pg_sys::INT4OID + | pg_sys::INT8OID + | pg_sys::FLOAT4OID + | pg_sys::FLOAT8OID + | pg_sys::DATEOID + | pg_sys::TIMESTAMPOID + | pg_sys::TIMESTAMPTZOID + ) { + pgrx::error!( + "pg_deltax: DeltaXMinMax cannot fold uncompressed rows for column type OID {}; run deltax_compact_partition() first", + u32::from(ct) + ); + } + if *type_oid == pg_sys::InvalidOid { + *type_oid = ct; + } + if *null { + *datum = d; + *null = false; + } else { + let cmp = compare_datums(d, *datum, ct); + let dominated = if *is_min { + cmp == std::cmp::Ordering::Less + } else { + cmp == std::cmp::Ordering::Greater + }; + if dominated { + *datum = d; + } + } + } + Acc::SumI128 { + acc, nonnull, seen, .. + } => { + if isnull { + continue; + } + let v: i128 = match spec.col_type_oid { + pg_sys::INT2OID => (d.value() as i16) as i128, + pg_sys::INT4OID => (d.value() as i32) as i128, + pg_sys::INT8OID => (d.value() as i64) as i128, + other => pgrx::error!( + "pg_deltax: DeltaXMinMax cannot fold uncompressed SUM rows for column type OID {}", + u32::from(other) + ), + }; + *acc = acc.saturating_add(v); + *nonnull = nonnull.saturating_add(1); + *seen = true; + } + Acc::SumF64 { + acc, nonnull, seen, .. + } => { + if isnull { + continue; + } + let v: f64 = match spec.col_type_oid { + pg_sys::FLOAT4OID => f32::from_bits(d.value() as u32) as f64, + pg_sys::FLOAT8OID => f64::from_bits(d.value() as u64), + other => pgrx::error!( + "pg_deltax: DeltaXMinMax cannot fold uncompressed SUM rows for column type OID {}", + u32::from(other) + ), + }; + *acc += v; + *nonnull = nonnull.saturating_add(1); + *seen = true; + } + Acc::Count { acc } => match spec.kind { + MetaAggKind::CountStar => *acc += 1, + MetaAggKind::CountCol => { + if !isnull { + *acc += 1; + } + } + _ => unreachable!(), + }, + } + } + }, + ); let heap_scan_us = t1.elapsed().as_micros() as u64; // Convert accumulators into the legacy `MinMaxResult` row shape diff --git a/src/scan/exec/decompress.rs b/src/scan/exec/decompress.rs index ced271e..4f28f0c 100644 --- a/src/scan/exec/decompress.rs +++ b/src/scan/exec/decompress.rs @@ -138,6 +138,65 @@ pub(crate) struct DecompressState { /// `ShutdownCustomScan` before the DSM is torn down. Empty on workers /// and in the serial path. Consumed by `ExplainCustomScan`. pub(crate) cached_worker_timings: Vec, + + /// P1 heap tail (Phase 3): loose rows INSERTed into compressed + /// partitions live in the partition heap and are emitted after the last + /// segment. `None` when every involved partition heap is empty (the + /// common case — zero overhead). Workers never set this; in parallel + /// DeltaXAppend only the leader emits the heap tail. + heap_tail: Option, +} + +/// State for the Phase 3 heap-tail scan: iterates the live tuples of one or +/// more partition heaps under the query snapshot. +struct HeapTailScan { + /// Partition heap OIDs left to scan (DeltaXDecompress: one; DeltaXAppend: + /// every compressed child whose heap had blocks at begin time). + oids: Vec, + /// Index of the relation currently being scanned (or next to open). + idx: usize, + /// Currently open relation (null when between relations). + rel: pg_sys::Relation, + /// Active heap scan on `rel` (null when between relations). + scan: pg_sys::TableScanDesc, + /// Query snapshot captured at BeginCustomScan. + snapshot: pg_sys::Snapshot, +} + +impl HeapTailScan { + fn new(oids: Vec, snapshot: pg_sys::Snapshot) -> Self { + HeapTailScan { + oids, + idx: 0, + rel: std::ptr::null_mut(), + scan: std::ptr::null_mut(), + snapshot, + } + } + + /// Close the currently open scan + relation, if any. + unsafe fn close_current(&mut self) { + unsafe { + if !self.scan.is_null() { + if let Some(end) = (*(*self.rel).rd_tableam).scan_end { + end(self.scan); + } + self.scan = std::ptr::null_mut(); + } + if !self.rel.is_null() { + pg_sys::table_close(self.rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + self.rel = std::ptr::null_mut(); + } + } + } + + /// Reset for rescan: close anything open and start over. + unsafe fn reset(&mut self) { + unsafe { + self.close_current(); + } + self.idx = 0; + } } /// Wall-clock timing for the decompress scan phases. @@ -201,6 +260,9 @@ pub(crate) struct ScanTiming { pub(crate) blob_cache_misses: u64, /// Bytes served from the blob cache (sum across hits). pub(crate) blob_cache_bytes_served: u64, + /// Rows emitted from the Phase 3 heap tail (loose rows inserted into + /// compressed partitions after compression). + pub(crate) heap_tail_rows: u64, } /// POD projection of `ScanTiming` for cross-process DSM aggregation. @@ -374,6 +436,11 @@ struct ParsedCustomPrivate { header: Vec, needed_indices: Vec, topn: Option, + /// `[-4, flag]` trailer: true when the planner claimed sorted output + /// (pathkeys) for this scan. The executor errors if the flag is set but + /// the partition heap has gained loose rows since planning (stale + /// cached plan) — appended heap-tail rows would break the sort promise. + pathkeys_claimed: bool, } struct TopNRaw { @@ -391,11 +458,13 @@ unsafe fn parse_custom_private(list: *mut pg_sys::List) -> ParsedCustomPrivate { let mut header = Vec::new(); let mut needed_indices = Vec::new(); let mut topn: Option = None; + let mut pathkeys_claimed = false; if list.is_null() { return ParsedCustomPrivate { header, needed_indices, topn, + pathkeys_claimed, }; } let len = unsafe { (*list).length }; @@ -408,6 +477,14 @@ unsafe fn parse_custom_private(list: *mut pg_sys::List) -> ParsedCustomPrivate { i += 1; continue; } + if val == -4 && sec >= 1 { + // Two-int pathkeys-claimed marker; section state is unchanged. + if i + 1 < len { + pathkeys_claimed = unsafe { pg_sys::list_nth_int(list, i + 1) } != 0; + } + i += 2; + continue; + } if val == -2 && sec == 1 { let read = |off: i32| -> Option { if i + off < len { @@ -443,6 +520,7 @@ unsafe fn parse_custom_private(list: *mut pg_sys::List) -> ParsedCustomPrivate { header, needed_indices, topn, + pathkeys_claimed, } } @@ -523,7 +601,7 @@ pub(crate) unsafe extern "C-unwind" fn create_custom_scan_state( #[pg_guard] pub(super) unsafe extern "C-unwind" fn begin_custom_scan( node: *mut pg_sys::CustomScanState, - _estate: *mut pg_sys::EState, + estate: *mut pg_sys::EState, _eflags: i32, ) { unsafe { @@ -540,7 +618,7 @@ pub(super) unsafe extern "C-unwind" fn begin_custom_scan( let companion_oid = pg_sys::Oid::from(parsed.header[0] as u32); let needed_indices = parsed.needed_indices; let ( - topn_limit, + mut topn_limit, topn_ascending, topn_nulls_first, topn_multi_col_sort, @@ -556,14 +634,58 @@ pub(super) unsafe extern "C-unwind" fn begin_custom_scan( None => (0i64, true, false, false, 0i32), }; + // P1 heap tail (Phase 3): the scan relation (the partition heap, + // opened by PG core since scanrelid > 0) may hold loose rows + // inserted after compression. Detect that here; exec then unions + // the heap tail after the segments. + let scan_rel = (*node).ss.ss_currentRelation; + let heap_tail_oids: Vec = if !scan_rel.is_null() + && pg_sys::RelationGetNumberOfBlocksInFork(scan_rel, pg_sys::ForkNumber::MAIN_FORKNUM) + > 0 + { + if parsed.pathkeys_claimed { + pgrx::error!( + "pg_deltax: compressed partition \"{}\" gained uncompressed rows after this plan was created (sorted scan); retry the query", + relation_name(scan_rel), + ); + } + // Heap-tail rows disable Top-N pushdown — the pruned candidate + // set would not include loose rows. The plan-time gate already + // does this for fresh plans; this covers stale cached plans. + topn_limit = 0; + vec![(*scan_rel).rd_id] + } else { + Vec::new() + }; + // Load metadata (cached), then load segment data via direct heap scan // (plan_qual is passed so batch qual columns are included in needed_cols) let plan_qual = (*(*node).ss.ps.plan).qual; let mut state = load_decompress_state(companion_oid, &needed_indices, plan_qual, topn_limit); - // If all plan quals are handled by batch eval, skip PG's per-row ExecQual - if state.all_quals_batch_handled { + if !heap_tail_oids.is_empty() { + // Schema guard: heap tuples are deformed positionally into the + // scan slot, which requires a 1:1 physical layout match with the + // companion's column list (no json_extract synthetic columns, + // no dropped columns). + let rel_natts = (*(*scan_rel).rd_att).natts as usize; + if rel_natts != state.col_names.len() { + pgrx::error!( + "pg_deltax: compressed partition \"{}\" has uncompressed rows but its physical layout ({} attrs) does not match the compressed layout ({} cols); decompress and recompress it", + relation_name(scan_rel), + rel_natts, + state.col_names.len(), + ); + } + state.heap_tail = Some(HeapTailScan::new(heap_tail_oids, (*estate).es_snapshot)); + } + + // If all plan quals are handled by batch eval, skip PG's per-row + // ExecQual — except when a heap tail exists: heap-tail rows are not + // batch-filtered, so the full qual must run per row (segment rows + // get double-filtered, which is correct, just slightly slower). + if state.all_quals_batch_handled && state.heap_tail.is_none() { (*node).ss.ps.qual = std::ptr::null_mut(); } @@ -677,6 +799,7 @@ fn make_worker_stub_state() -> DecompressState { wire_view: None, is_worker_stub: true, cached_worker_timings: Vec::new(), + heap_tail: None, } } @@ -684,7 +807,7 @@ fn make_worker_stub_state() -> DecompressState { #[pg_guard] pub(super) unsafe extern "C-unwind" fn begin_deltax_append( node: *mut pg_sys::CustomScanState, - _estate: *mut pg_sys::EState, + estate: *mut pg_sys::EState, _eflags: i32, ) { unsafe { @@ -711,7 +834,7 @@ pub(super) unsafe extern "C-unwind" fn begin_deltax_append( .collect(); let needed_indices = parsed.needed_indices; let ( - topn_limit, + mut topn_limit, topn_ascending, topn_nulls_first, topn_multi_col_sort, @@ -731,6 +854,33 @@ pub(super) unsafe extern "C-unwind" fn begin_deltax_append( pgrx::error!("pg_deltax: DeltaXAppend has no companion tables"); } + // P1 heap tail (Phase 3): collect the compressed children whose + // partition heap holds loose rows (transparent INSERTs after + // compression). Their live tuples are emitted by the leader after + // the segment cursor is exhausted. + let heap_tail_oids: Vec = { + let mut v = Vec::new(); + for &oid in &companion_oids { + let part_oid = super::segments::partition_oid_for_companion(oid); + if part_oid == pg_sys::InvalidOid { + pgrx::error!( + "pg_deltax: cannot resolve partition for companion OID {} (catalog inconsistency)", + u32::from(oid) + ); + } + if !crate::scan::relation_heap_is_empty(part_oid) { + v.push(part_oid); + } + } + v + }; + if !heap_tail_oids.is_empty() { + // Loose rows would be invisible to the Top-N candidate pass; + // fall back to the full scan (upper Sort/Limit stays correct — + // DeltaXAppend never claims pathkeys). + topn_limit = 0; + } + // Load metadata for first companion table (cached per-backend). let t0 = Instant::now(); let meta = super::segments::load_metadata_cached(companion_oids[0]); @@ -746,8 +896,10 @@ pub(super) unsafe extern "C-unwind" fn begin_deltax_append( (*plan_qual).length as usize }; let all_quals_batch_handled = handled_count > 0 && handled_count == nquals; - if all_quals_batch_handled { - // All quals are handled by batch eval — skip PG's per-row ExecQual + if all_quals_batch_handled && heap_tail_oids.is_empty() { + // All quals are handled by batch eval — skip PG's per-row + // ExecQual. With a heap tail the full qual must keep running: + // heap-tail rows are never batch-filtered. (*node).ss.ps.qual = std::ptr::null_mut(); } @@ -940,6 +1092,7 @@ pub(super) unsafe extern "C-unwind" fn begin_deltax_append( wire_view: None, is_worker_stub: false, cached_worker_timings: Vec::new(), + heap_tail: None, }; // Create per-segment memory context @@ -957,6 +1110,10 @@ pub(super) unsafe extern "C-unwind" fn begin_deltax_append( .segments_data .sort_by_key(|s| s.min_time.unwrap_or(i64::MAX)); + if !heap_tail_oids.is_empty() { + state.heap_tail = Some(HeapTailScan::new(heap_tail_oids, (*estate).es_snapshot)); + } + let state_box = Box::new(state); let state_ptr = Box::into_raw(state_box); (*node).custom_ps = state_ptr as *mut pg_sys::List; @@ -1135,6 +1292,7 @@ fn load_decompress_state( wire_view: None, is_worker_stub: false, cached_worker_timings: Vec::new(), + heap_tail: None, } } @@ -3389,6 +3547,14 @@ pub(super) unsafe extern "C-unwind" fn exec_custom_scan( return slot; } if !load_next_segment(state, instrument) { + // Phase 3: heap tail — live tuples sitting in the partition + // heap (loose rows inserted after compression). Emitted by + // the leader only, after every segment is consumed. + if let Some(slot) = + try_emit_heap_tail_row(state, scan_slot, econtext, qual, proj_info) + { + return slot; + } pg_sys::ExecClearTuple(scan_slot); return scan_slot; } @@ -3396,6 +3562,125 @@ pub(super) unsafe extern "C-unwind" fn exec_custom_scan( } } +/// Phase 3: emit the next qualifying live tuple from the partition heap(s). +/// +/// Active only when `state.heap_tail` was armed at begin time (some involved +/// partition heap had blocks). Tuples are read with a plain heap scan under +/// the query snapshot, deformed positionally into the scan slot (layout +/// verified at begin), filtered through the full plan qual (`ExecQual` — heap +/// rows never went through batch quals), and projected like segment rows. +/// +/// Parallel DeltaXAppend: the heap tail is leader-only; workers return None +/// immediately (their stub state has `heap_tail = None` anyway). +/// +/// Returns `Some(slot)` for each qualifying row, `None` when every heap is +/// exhausted (end of scan). +unsafe fn try_emit_heap_tail_row( + state: &mut DecompressState, + scan_slot: *mut pg_sys::TupleTableSlot, + econtext: *mut pg_sys::ExprContext, + qual: *mut pg_sys::ExprState, + proj_info: *mut pg_sys::ProjectionInfo, +) -> Option<*mut pg_sys::TupleTableSlot> { + unsafe { + // Belt-and-suspenders: never emit the heap tail from a worker. + if pg_sys::ParallelWorkerNumber >= 0 { + return None; + } + // Disjoint field borrows: the heap-tail cursor and the timing + // counters are updated independently inside the loop. + let ncols = state.col_names.len(); + let timing = &mut state.timing; + let ht = state.heap_tail.as_mut()?; + loop { + if ht.scan.is_null() { + if ht.idx >= ht.oids.len() { + return None; + } + let oid = ht.oids[ht.idx]; + let rel = pg_sys::table_open(oid, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + let rel_natts = (*(*rel).rd_att).natts as usize; + if rel_natts != ncols { + let name = relation_name(rel); + pg_sys::table_close(rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); + pgrx::error!( + "pg_deltax: compressed partition \"{}\" has uncompressed rows but its physical layout ({} attrs) does not match the compressed layout ({} cols); decompress and recompress it", + name, + rel_natts, + ncols, + ); + } + let flags: u32 = pg_sys::ScanOptions::SO_TYPE_SEQSCAN + | pg_sys::ScanOptions::SO_ALLOW_STRAT + | pg_sys::ScanOptions::SO_ALLOW_SYNC + | pg_sys::ScanOptions::SO_ALLOW_PAGEMODE; + let scan = (*(*rel).rd_tableam).scan_begin.unwrap()( + rel, + ht.snapshot, + 0, + std::ptr::null_mut(), + std::ptr::null_mut(), + flags, + ); + ht.rel = rel; + ht.scan = scan; + } + + let tuple = pg_sys::heap_getnext(ht.scan, pg_sys::ScanDirection::ForwardScanDirection); + if tuple.is_null() { + ht.close_current(); + ht.idx += 1; + continue; + } + + // Deform positionally into the scan slot (1:1 layout verified + // above). Pass-by-reference datums point into the scan's pinned + // page, which stays valid until the next heap_getnext — i.e. + // until the next call, matching the slot contract. + pg_sys::ExecClearTuple(scan_slot); + pg_sys::heap_deform_tuple( + tuple, + (*ht.rel).rd_att, + (*scan_slot).tts_values, + (*scan_slot).tts_isnull, + ); + pg_sys::ExecStoreVirtualTuple(scan_slot); + (*econtext).ecxt_scantuple = scan_slot; + + // Full plan qual — heap rows are never batch-filtered. + if !qual.is_null() && !exec_qual(qual, econtext) { + pg_sys::MemoryContextReset((*econtext).ecxt_per_tuple_memory); + timing.rows_filtered += 1; + continue; + } + + let result = if !proj_info.is_null() { + exec_project(proj_info) + } else { + scan_slot + }; + timing.rows_emitted += 1; + timing.heap_tail_rows += 1; + return Some(result); + } + } +} + +/// Best-effort relation name for error messages. +unsafe fn relation_name(rel: pg_sys::Relation) -> String { + unsafe { + if rel.is_null() { + return "?".to_string(); + } + let p = pg_sys::get_rel_name((*rel).rd_id); + if p.is_null() { + format!("OID {}", u32::from((*rel).rd_id)) + } else { + std::ffi::CStr::from_ptr(p).to_string_lossy().into_owned() + } + } +} + /// Handle Top-N execution paths for queries with ORDER BY + LIMIT. /// /// Active for queries like `SELECT * FROM t ORDER BY ts DESC LIMIT 10` where @@ -3911,7 +4196,10 @@ pub(super) unsafe extern "C-unwind" fn end_custom_scan(node: *mut pg_sys::Custom unsafe { let state_ptr = (*node).custom_ps as *mut DecompressState; if !state_ptr.is_null() { - let state = Box::from_raw(state_ptr); + let mut state = Box::from_raw(state_ptr); + if let Some(ht) = state.heap_tail.as_mut() { + ht.close_current(); + } // Emit timing summary at LOG level (visible with SET client_min_messages = log) // All timers are non-overlapping: @@ -3951,6 +4239,9 @@ pub(super) unsafe extern "C-unwind" fn end_custom_scan(node: *mut pg_sys::Custom t.topn_limit, t.topn_candidates, ); + if t.heap_tail_rows > 0 { + pgrx::log!("pg_deltax heap_tail: rows={}", t.heap_tail_rows); + } if !state.segment_mcxt.is_null() { pg_sys::MemoryContextDelete(state.segment_mcxt); @@ -3973,6 +4264,9 @@ pub(super) unsafe extern "C-unwind" fn rescan_custom_scan(node: *mut pg_sys::Cus state.topn_buffer.clear(); state.topn_cursor = 0; state.topn_done = false; + if let Some(ht) = state.heap_tail.as_mut() { + ht.reset(); + } } } diff --git a/src/scan/exec/segments.rs b/src/scan/exec/segments.rs index e71340e..558c71c 100644 --- a/src/scan/exec/segments.rs +++ b/src/scan/exec/segments.rs @@ -292,6 +292,59 @@ unsafe fn sibling_table_oid(meta_oid: pg_sys::Oid, suffix: &str) -> pg_sys::Oid } } +thread_local! { + /// Backend-local cache: companion (`_meta`) table OID → partition heap + /// OID. Reverse of `check_compressed_partition`. Cleared wholesale by + /// `metadata_relcache_callback` like `METADATA_CACHE`. + static PARTITION_OID_CACHE: RefCell> = + RefCell::new(HashMap::new()); +} + +/// Resolve a companion (`_meta`) table OID back to the partition heap OID, +/// via the `deltax.deltax_partition` catalog (the partition lives in the +/// user's schema, not in `_deltax_compressed`, so `sibling_table_oid` can't +/// be used). Returns `InvalidOid` when the partition can't be found. +/// Cached per-backend; only successful lookups are cached. +pub(crate) fn partition_oid_for_companion(companion_oid: pg_sys::Oid) -> pg_sys::Oid { + if let Some(oid) = PARTITION_OID_CACHE.with(|c| c.borrow().get(&companion_oid).copied()) { + return oid; + } + ensure_metadata_callback_registered(); + let partition_name = unsafe { + let name_ptr = pg_sys::get_rel_name(companion_oid); + if name_ptr.is_null() { + return pg_sys::InvalidOid; + } + let meta_name = std::ffi::CStr::from_ptr(name_ptr) + .to_string_lossy() + .into_owned(); + match meta_name.strip_suffix("_meta") { + Some(p) => p.to_string(), + None => return pg_sys::InvalidOid, + } + }; + let oid = pgrx::Spi::connect(|client| { + client + .select( + "SELECT c.oid + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN deltax.deltax_partition p + ON p.schema_name = n.nspname AND p.table_name = c.relname + WHERE p.table_name = $1", + Some(1), + &[partition_name.into()], + ) + .ok() + .and_then(|r| r.first().get_one::().ok().flatten()) + .unwrap_or(pg_sys::InvalidOid) + }); + if oid != pg_sys::InvalidOid { + PARTITION_OID_CACHE.with(|c| c.borrow_mut().insert(companion_oid, oid)); + } + oid +} + /// Locate the primary-key btree index on a heap relation by walking /// `RelationGetIndexList` and matching on `indisprimary`. Returns /// `InvalidOid` if the relation has no PK (e.g. blobs/blooms tables in @@ -1038,6 +1091,7 @@ unsafe extern "C-unwind" fn metadata_relcache_callback(_arg: pg_sys::Datum, _rel // and this avoids tracking dependencies between MetadataInfo entries and // every catalog row they read (parent table pg_attribute, deltax catalog). METADATA_CACHE.with(|c| c.borrow_mut().clear()); + PARTITION_OID_CACHE.with(|c| c.borrow_mut().clear()); } fn ensure_metadata_callback_registered() { diff --git a/src/scan/hook.rs b/src/scan/hook.rs index 8232a00..15bbdea 100644 --- a/src/scan/hook.rs +++ b/src/scan/hook.rs @@ -32,6 +32,11 @@ thread_local! { /// When true, the ExecutorStart hook skips the DML-on-compressed check. /// Used by internal operations like deltax_decompress_partition. static DML_BYPASS: std::cell::Cell = const { std::cell::Cell::new(false) }; + + /// Whether the planner-cache relcache callback has been registered in + /// this backend (one-shot, like the metadata-cache callback in + /// `segments.rs`). + static HOOK_CB_REGISTERED: std::cell::Cell = const { std::cell::Cell::new(false) }; } pub fn invalidate_compressed_cache() { @@ -42,6 +47,35 @@ pub fn invalidate_compressed_cache() { super::exec::segments::invalidate_colstats_cache(); } +/// Relcache-invalidation callback: conservatively wipe every planner-side +/// thread-local cache on any relcache invalidation. This is what makes +/// cross-backend state changes (a partition gaining its first loose row, a +/// compaction bumping catalog row counts, compress/decompress in another +/// session) visible here — the writer sends a relcache inval for the +/// partition and every backend drops its cached row counts / valmaps / +/// companion lookups. Mirrors `segments::metadata_relcache_callback`. +#[pg_guard] +unsafe extern "C-unwind" fn hook_caches_relcache_callback( + _arg: pg_sys::Datum, + _relid: pg_sys::Oid, +) { + invalidate_compressed_cache(); +} + +fn ensure_hook_callback_registered() { + HOOK_CB_REGISTERED.with(|c| { + if !c.get() { + unsafe { + pg_sys::CacheRegisterRelcacheCallback( + Some(hook_caches_relcache_callback), + pg_sys::Datum::from(0u32), + ); + } + c.set(true); + } + }); +} + /// Look up the companion OID for a partition's heap OID, using /// `COMPRESSED_CACHE` to amortise the catalog probe across the planner's /// repeated calls on the same query. @@ -52,6 +86,7 @@ unsafe fn cached_companion_for_rel(rel_oid: pg_sys::Oid) -> pg_sys::Oid { { return oid; } + ensure_hook_callback_registered(); let oid = unsafe { check_compressed_partition(rel_oid) }; COMPRESSED_CACHE.with(|c| c.borrow_mut().insert(rel_oid, oid)); oid @@ -123,6 +158,12 @@ pub(crate) fn set_dml_bypass(bypass: bool) { DML_BYPASS.with(|flag| flag.set(bypass)); } +/// Read the DML bypass flag (probed by the compressed-partition trigger via +/// `deltax.deltax_dml_bypass_active()`). +pub(crate) fn dml_bypass_active() -> bool { + DML_BYPASS.with(|flag| flag.get()) +} + /// Get the time column's attribute number for a deltatable parent table. /// Returns None if the table is not a deltax deltatable. /// @@ -572,6 +613,13 @@ pub unsafe extern "C-unwind" fn deltax_set_rel_pathlist( } let _profile = super::plan_profile::scope("set_rel_pathlist"); + // Internal maintenance (compaction reading the loose heap rows of a + // compressed partition) must see the plain heap, not the + // segments ∪ heap union the custom scans produce. + if DML_BYPASS.with(|flag| flag.get()) { + return; + } + // Only handle regular tables if (*rte).rtekind != pg_sys::RTEKind::RTE_RELATION { return; @@ -585,8 +633,13 @@ pub unsafe extern "C-unwind" fn deltax_set_rel_pathlist( // Check if this is the parent of a partitioned table (for DeltaXAppend) if (*rel).reloptkind == pg_sys::RelOptKind::RELOPT_BASEREL && (*rte).inh - && let Some(companion_oids) = collect_compressed_children(root, rti) + && let Some((companion_oids, any_heap_tail)) = collect_compressed_children(root, rti) { + // P1 heap tail: when any compressed child has loose heap rows, + // Top-N pushdown must be disabled — the pruned candidate set + // would not include the loose rows. The exec-time heap-tail + // union (Phase 3) keeps the plain scan correct. + let effective_limit = if any_heap_tail { 0 } else { effective_limit }; // For Top-N, validate ORDER BY is a simple column reference. // Works for any column (time, text, numeric). let (append_topn_limit, append_sort_col_attno) = if effective_limit > 0 { @@ -676,6 +729,14 @@ pub unsafe extern "C-unwind" fn deltax_set_rel_pathlist( return; } + // P1 heap tail: with loose rows in the partition heap the scan's + // output is segments ∪ heap tail. The heap tail is appended after + // the (time-ordered) segment rows, so the node can neither claim + // sorted output (pathkeys) nor push Top-N down. The plain path + + // upper Sort/Limit stays correct. + let heap_tail = !relation_heap_is_empty(rel_oid); + let effective_limit = if heap_tail { 0 } else { effective_limit }; + // For child partitions, check if we can advertise sorted output // and whether Top-N is valid. let parent_oid_opt = if (*rel).reloptkind == pg_sys::RelOptKind::RELOPT_OTHER_MEMBER_REL { @@ -689,7 +750,7 @@ pub unsafe extern "C-unwind" fn deltax_set_rel_pathlist( .unwrap_or(false); // Pathkeys for sorted output: only when no segment_by and time pathkey matches - let (pathkeys, _sort_ascending) = if !has_segby { + let (pathkeys, _sort_ascending) = if !has_segby && !heap_tail { time_col_attno_opt .map(|attno| check_time_pathkey(root, rel, attno)) .unwrap_or((std::ptr::null_mut(), true)) @@ -1941,6 +2002,10 @@ pub unsafe extern "C-unwind" fn deltax_create_upper_paths( if stage != pg_sys::UpperRelationKind::UPPERREL_GROUP_AGG { return; } + // Internal maintenance queries plan against the plain heap. + if DML_BYPASS.with(|flag| flag.get()) { + return; + } let _profile = super::plan_profile::scope("create_upper_paths"); let parse = (*root).parse; @@ -2626,6 +2691,16 @@ pub unsafe extern "C-unwind" fn deltax_create_upper_paths( // DeltaXAgg path: SUM/AVG/COUNT/COUNT(DISTINCT) ± GROUP BY ± WHERE ± HAVING // ===================================================================== + // P1 heap tail: DeltaXAgg's columnar accumulators only ingest + // segment data — loose heap rows (transparent INSERTs into + // compressed partitions) would be silently dropped. Bail to the + // plain Agg over DeltaXDecompress/DeltaXAppend scans, which union + // the heap tail at exec time. (DeltaXCount/DeltaXMinMax above stay + // enabled: their executors fold the heap tail in.) + if any_compressed_rte_heap_nonempty(root) { + return; + } + // Verify all WHERE quals are batch-pushable. Each qual must be // extractable by extract_batch_quals at execution time, otherwise the // filter is silently dropped and AggScan produces wrong results. @@ -4504,7 +4579,7 @@ unsafe fn extract_oids_from_custom_path( /// ANALYZE having run. A drained-but-not-vacuumed partition keeps its blocks, /// so this is conservative (reports non-empty) — which is the safe direction /// for the DeltaXAppend correctness check. -unsafe fn relation_heap_is_empty(oid: pg_sys::Oid) -> bool { +pub(crate) unsafe fn relation_heap_is_empty(oid: pg_sys::Oid) -> bool { unsafe { let rel = pg_sys::table_open(oid, pg_sys::AccessShareLock as pg_sys::LOCKMODE); let nblocks = @@ -4514,10 +4589,35 @@ unsafe fn relation_heap_is_empty(oid: pg_sys::Oid) -> bool { } } +/// True when any relation RTE in the query is a compressed partition whose +/// heap holds loose rows (P1 transparent INSERT). Conservative gate for the +/// DeltaXAgg pushdown, which cannot ingest row-form heap tuples. +unsafe fn any_compressed_rte_heap_nonempty(root: *mut pg_sys::PlannerInfo) -> bool { + unsafe { + let array_size = (*root).simple_rel_array_size; + for rti in 1..array_size { + let rte = *(*root).simple_rte_array.add(rti as usize); + if rte.is_null() + || (*rte).rtekind != pg_sys::RTEKind::RTE_RELATION + || (*rte).relid == pg_sys::InvalidOid + { + continue; + } + let relid = (*rte).relid; + if cached_companion_for_rel(relid) != pg_sys::InvalidOid + && !relation_heap_is_empty(relid) + { + return true; + } + } + false + } +} + unsafe fn collect_compressed_children( root: *mut pg_sys::PlannerInfo, parent_rti: pg_sys::Index, -) -> Option> { +) -> Option<(Vec, bool)> { unsafe { let _profile = super::plan_profile::scope("collect_compressed_children"); let list = (*root).append_rel_list; @@ -4527,6 +4627,7 @@ unsafe fn collect_compressed_children( let len = (*list).length; let mut companion_oids: Vec = Vec::new(); + let mut any_heap_tail = false; for i in 0..len { let node = pg_sys::list_nth(list, i) as *const pg_sys::AppendRelInfo; @@ -4546,6 +4647,12 @@ unsafe fn collect_compressed_children( let companion_oid = cached_companion_for_rel(child_oid); if companion_oid != pg_sys::InvalidOid { + // P1 heap tail: compressed children may hold loose rows in + // their heap. DeltaXAppend scans those at exec time + // (Phase 3), but the caller must disable Top-N pushdown. + if !relation_heap_is_empty(child_oid) { + any_heap_tail = true; + } companion_oids.push(companion_oid); } else { // Not compressed — DeltaXAppend reads only compressed @@ -4568,7 +4675,7 @@ unsafe fn collect_compressed_children( if companion_oids.is_empty() { None } else { - Some(companion_oids) + Some((companion_oids, any_heap_tail)) } } } @@ -4645,6 +4752,24 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( let rtable = (*planned_stmt).rtable; let n = (*result_relations).length; + // P1 transparent INSERT: plain INSERTs into compressed + // partitions are allowed — rows land in the partition heap + // (the loose-row region) and every read path unions them with + // the segment data. INSERT ... ON CONFLICT stays rejected: + // segment rows have no index entries, so conflict inference + // against compressed data would be silently wrong. + let insert_on_conflict = operation == pg_sys::CmdType::CMD_INSERT && { + let plan_tree = (*planned_stmt).planTree; + !plan_tree.is_null() + && (*plan_tree).type_ == pg_sys::NodeTag::T_ModifyTable + && (*(plan_tree as *mut pg_sys::ModifyTable)).onConflictAction + != pg_sys::OnConflictAction::ONCONFLICT_NONE + }; + if operation == pg_sys::CmdType::CMD_INSERT && !insert_on_conflict { + call_prev_executor_start(query_desc, eflags); + return; + } + for i in 0..n { // resultRelations is an IntList of 1-based RTE indices let rti = (*(*result_relations).elements.add(i as usize)).int_value; @@ -4660,13 +4785,22 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( let relid = (*rte).relid; let companion_oid = cached_companion_for_rel(relid); - - if companion_oid != pg_sys::InvalidOid { - let op_name = match operation { - pg_sys::CmdType::CMD_INSERT => "INSERT into", - pg_sys::CmdType::CMD_UPDATE => "UPDATE", - pg_sys::CmdType::CMD_DELETE => "DELETE from", - _ => "modify", + // For INSERT ... ON CONFLICT, the named target is usually + // the partitioned parent (tuple routing picks the leaf + // later), so also reject when the parent has at least one + // compressed partition. + let blocked = companion_oid != pg_sys::InvalidOid + || (insert_on_conflict && parent_has_compressed_partition(relid)); + + if blocked { + let op_name = if insert_on_conflict { + "INSERT ... ON CONFLICT into" + } else { + match operation { + pg_sys::CmdType::CMD_UPDATE => "UPDATE", + pg_sys::CmdType::CMD_DELETE => "DELETE from", + _ => "modify", + } }; let rel_name_ptr = pg_sys::get_rel_name(relid); let rel_name = if rel_name_ptr.is_null() { @@ -4676,11 +4810,19 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( .to_string_lossy() .into_owned() }; - pgrx::error!( - "cannot {} compressed partition \"{}\", decompress it first", - op_name, - rel_name, - ); + if insert_on_conflict { + pgrx::error!( + "cannot {} \"{}\": ON CONFLICT cannot see rows in compressed partitions, decompress them first", + op_name, + rel_name, + ); + } else { + pgrx::error!( + "cannot {} compressed partition \"{}\", decompress it first", + op_name, + rel_name, + ); + } } } } @@ -4689,6 +4831,32 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( } } +/// True when `relid` is a deltax-managed parent table with at least one +/// compressed partition. Used only on the (rare) INSERT ... ON CONFLICT +/// path, so the SPI probe is acceptable. +unsafe fn parent_has_compressed_partition(relid: pg_sys::Oid) -> bool { + if !crate::catalog::catalog_present() { + return false; + } + Spi::connect(|client| { + client + .select( + "SELECT 1 + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN deltax.deltax_deltatable h + ON h.schema_name = n.nspname AND h.table_name = c.relname + JOIN deltax.deltax_partition p ON p.deltatable_id = h.id + WHERE c.oid = $1 AND p.is_compressed + LIMIT 1", + Some(1), + &[relid.into()], + ) + .map(|r| !r.is_empty()) + .unwrap_or(false) + }) +} + /// Chain to the previous ExecutorStart hook or call standard_ExecutorStart. unsafe fn call_prev_executor_start(query_desc: *mut pg_sys::QueryDesc, eflags: c_int) { unsafe { diff --git a/src/scan/json_extract.rs b/src/scan/json_extract.rs index 5a34d0b..e911d92 100644 --- a/src/scan/json_extract.rs +++ b/src/scan/json_extract.rs @@ -2107,8 +2107,22 @@ unsafe fn rebuild_cscan_custom_private(cscan: *mut pg_sys::CustomScan, reference InSynth, } let mut phase = Phase::BeforeMinus1; + // `[-4, flag]` (pathkeys-claimed marker) is a two-int section + // that may appear after the cols or top-n sections; preserve it + // verbatim wherever it sits. + let mut copy_next = false; for i in 0..(*cp).length { let v = pg_sys::list_nth_int(cp, i); + if copy_next { + new_cp = pg_sys::lappend_int(new_cp, v); + copy_next = false; + continue; + } + if v == -4 && phase != Phase::BeforeMinus1 { + new_cp = pg_sys::lappend_int(new_cp, v); + copy_next = true; + continue; + } match phase { Phase::BeforeMinus1 => { new_cp = pg_sys::lappend_int(new_cp, v); diff --git a/src/scan/mod.rs b/src/scan/mod.rs index e5c26d6..7682b21 100644 --- a/src/scan/mod.rs +++ b/src/scan/mod.rs @@ -61,6 +61,22 @@ pub(crate) fn set_dml_bypass(bypass: bool) { hook::set_dml_bypass(bypass); } +/// Read the backend-local DML bypass flag (used by the compressed-partition +/// trigger via `deltax.deltax_dml_bypass_active()`). +pub(crate) fn dml_bypass_active() -> bool { + hook::dml_bypass_active() +} + +/// True if the relation's main fork has zero blocks (no loose heap rows +/// possible). Conservative in the non-empty direction: deleted-but-not- +/// vacuumed rows keep their blocks. +/// +/// # Safety +/// Must be called with a valid transaction (opens the relation). +pub(crate) unsafe fn relation_heap_is_empty(oid: pg_sys::Oid) -> bool { + unsafe { hook::relation_heap_is_empty(oid) } +} + /// Look up the companion OID for a compressed partition's relation OID, /// or `InvalidOid` if it's not a pg_deltax-managed compressed table. /// diff --git a/src/scan/path.rs b/src/scan/path.rs index 6323ae1..53c3e5e 100644 --- a/src/scan/path.rs +++ b/src/scan/path.rs @@ -251,6 +251,15 @@ static CUSTOM_SCAN_METHODS: SyncStatic = // sort_col_attno is 1-based PG attribute number of the ORDER BY column. thread_local! { static TOPN_INFO: std::cell::Cell<(i64, bool, bool, i32, bool)> = const { std::cell::Cell::new((0, true, false, 0, false)) }; + + /// Per-companion "this path claims sorted output (pathkeys)" flags, keyed + /// by companion OID so concurrent paths for different partitions in one + /// query don't clobber each other. Written by `add_decompress_path`, + /// read by `plan_custom_path`, serialized as the `[-4, flag]` trailer so + /// the executor can detect the stale-plan hazard: a plan that promised + /// sorted output but whose partition heap has since gained loose rows. + static PATHKEYS_CLAIMED: std::cell::RefCell> = + std::cell::RefCell::new(std::collections::HashMap::new()); } /// Register every CustomScanMethods struct with PG's name-keyed registry. @@ -328,6 +337,13 @@ pub unsafe fn add_decompress_path( (*cpath).custom_restrictinfo = std::ptr::null_mut(); (*cpath).methods = &CUSTOM_PATH_METHODS.0; + // Record whether this path claims sorted output, keyed by companion + // OID (consumed in plan_custom_path). + PATHKEYS_CLAIMED.with(|m| { + m.borrow_mut() + .insert(u32::from(companion_oid), !pathkeys.is_null()) + }); + // Store Top-N info for plan_custom_path. if effective_limit > 0 { TOPN_INFO.with(|cell| { @@ -603,6 +619,19 @@ pub unsafe extern "C-unwind" fn plan_custom_path( private_list = pg_sys::lappend_int(private_list, if nulls_first { 1 } else { 0 }); } + // Append pathkeys-claimed marker: [-4, flag]. The executor errors if + // the flag is set but the partition heap has gained loose rows since + // planning (stale cached plan) — appended heap-tail rows would break + // the promised sort order. + let pathkeys_claimed = PATHKEYS_CLAIMED.with(|m| { + m.borrow() + .get(&u32::from(companion_oid)) + .copied() + .unwrap_or(false) + }); + private_list = pg_sys::lappend_int(private_list, -4); + private_list = pg_sys::lappend_int(private_list, if pathkeys_claimed { 1 } else { 0 }); + (*cscan).custom_private = private_list; (*cscan).custom_scan_tlist = std::ptr::null_mut(); (*cscan).custom_plans = std::ptr::null_mut(); diff --git a/src/worker.rs b/src/worker.rs index c4ef567..c59beda 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -225,6 +225,29 @@ pub extern "C-unwind" fn deltax_worker_main(arg: pg_sys::Datum) { } } + // Compact loose heap rows (transparent INSERTs into + // compressed partitions) into new segments so the + // metadata-only fast paths re-engage. + let compacted = crate::compress::auto_compact_partitions(client, ht); + if compacted > 0 { + log!( + "pg_deltax: compacted loose rows in {} partition(s) for {}.{}", + compacted, + ht.schema_name, + ht.table_name + ); + if let Err(e) = + crate::stats::write_table_stats(client, &ht.schema_name, &ht.table_name) + { + log!( + "pg_deltax: failed to refresh parent stats for {}.{}: {:?}", + ht.schema_name, + ht.table_name, + e + ); + } + } + // Auto-drop expired partitions (retention policy) let dropped = partition::auto_drop_partitions(client, ht); if dropped > 0 { diff --git a/tests/test_compressed_insert.py b/tests/test_compressed_insert.py new file mode 100644 index 0000000..ec5bf74 --- /dev/null +++ b/tests/test_compressed_insert.py @@ -0,0 +1,371 @@ +"""Integration tests for P1 transparent INSERT into compressed partitions. + +Covers dev/docs/COMPRESSED_DML.md §4 (P1): + - INSERT into a compressed partition succeeds and lands in the partition + heap (the "loose row" region). + - Every read path unions the heap tail with the segment data: plain + scans, COUNT(*), MIN/MAX/SUM pushdowns, GROUP BY aggregates, point + lookups (partition bloom sentinels must never hide heap rows), and + ORDER BY ... LIMIT. + - UPDATE / DELETE stay rejected. + - INSERT ... ON CONFLICT stays rejected. + - deltax_compact_partition() folds loose rows into new segments; results + are unchanged afterwards and the heap is empty again. + +Correctness is asserted by comparing against an identical plain-PostgreSQL +twin table after every step. +""" + +import pytest + +MOCK_NOW = "2025-01-15 12:00:00+00" +BASE_TS = "2025-01-15 00:00:00+00" +LATE_TS = "2025-01-15 06:30:00+00" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def setup_tables(conn, segment_by=True): + """Create a deltax table + a plain twin, load identical data, compress.""" + conn.execute(f"SET pg_deltax.mock_now = '{MOCK_NOW}'") + for name in ("events", "events_plain"): + conn.execute(f""" + CREATE TABLE {name} ( + ts TIMESTAMPTZ NOT NULL, + device_id TEXT NOT NULL, + val INT, + temperature DOUBLE PRECISION + ) + """) + conn.execute( + "SELECT deltax.deltax_create_table('events', 'ts', '1 day'::interval)" + ) + if segment_by: + conn.execute( + "SELECT deltax.deltax_enable_compression('events', " + "segment_by => ARRAY['device_id'], order_by => ARRAY['ts'])" + ) + else: + conn.execute( + "SELECT deltax.deltax_enable_compression('events', " + "segment_by => ARRAY[]::text[], order_by => ARRAY['ts'])" + ) + conn.commit() + + rows = [] + for d in range(4): + for p in range(200): + rows.append( + f"('{BASE_TS}'::timestamptz + interval '{p} minutes', " + f"'device-{d}', {d * 1000 + p}, {20.0 + d}::float8 + {p}::float8 / 4)" + ) + for name in ("events", "events_plain"): + conn.execute( + f"INSERT INTO {name} (ts, device_id, val, temperature) VALUES " + + ", ".join(rows) + ) + conn.commit() + + +def compress_all(conn, table="events"): + parts = [ + r[0] + for r in conn.execute( + f"SELECT partition_name FROM deltax.deltax_partition_info('{table}') " + "WHERE partition_name NOT LIKE '%default%'" + ).fetchall() + ] + compressed = [] + for p in parts: + result = conn.execute( + f"SELECT deltax.deltax_compress_partition('{p}')" + ).fetchone()[0] + if "Compressed" in result: + compressed.append(p) + conn.commit() + assert compressed, "expected at least one compressed partition" + return compressed + + +def data_partition(conn, table="events"): + """The compressed partition holding BASE_TS.""" + return conn.execute( + f"SELECT partition_name FROM deltax.deltax_partition_info('{table}') " + "WHERE is_compressed ORDER BY partition_name LIMIT 1" + ).fetchone()[0] + + +def insert_late_rows(conn, n=25): + """Insert late-arriving rows into BOTH tables (routed into the + already-compressed partition for `events`).""" + rows = [] + for p in range(n): + rows.append( + f"('{LATE_TS}'::timestamptz + interval '{p} seconds', " + f"'device-late', {90000 + p}, 99.5::float8 + {p})" + ) + for name in ("events", "events_plain"): + conn.execute( + f"INSERT INTO {name} (ts, device_id, val, temperature) VALUES " + + ", ".join(rows) + ) + conn.commit() + + +def assert_tables_match(conn): + """Full result-set equality between the deltax table and the twin.""" + q = "SELECT ts, device_id, val, temperature FROM {} ORDER BY ts, device_id, val" + got = conn.execute(q.format("events")).fetchall() + want = conn.execute(q.format("events_plain")).fetchall() + assert got == want + + +QUERIES = [ + # (description, query template) + ("count_star", "SELECT count(*) FROM {}"), + ("count_where_time", + f"SELECT count(*) FROM {{}} WHERE ts >= '{BASE_TS}'::timestamptz"), + ("min_max_ts", "SELECT min(ts), max(ts) FROM {}"), + ("min_max_sum_val", "SELECT min(val), max(val), sum(val), count(val) FROM {}"), + ("group_by_device", + "SELECT device_id, count(*), sum(val) FROM {} GROUP BY device_id ORDER BY device_id"), + ("point_lookup_new_val", "SELECT count(*) FROM {} WHERE val = 90003"), + ("point_lookup_old_val", "SELECT count(*) FROM {} WHERE val = 1005"), + ("point_lookup_absent_val", "SELECT count(*) FROM {} WHERE val = 123456789"), + ("text_eq_new_value", + "SELECT count(*), coalesce(sum(val), 0) FROM {} WHERE device_id = 'device-late'"), + ("topn_ts_desc", + "SELECT ts, device_id, val FROM {} ORDER BY ts DESC, device_id, val LIMIT 7"), + ("topn_ts_asc", + "SELECT ts, device_id, val FROM {} ORDER BY ts ASC, device_id, val LIMIT 7"), + ("time_range", + f"SELECT count(*), coalesce(sum(val), 0) FROM {{}} " + f"WHERE ts BETWEEN '{LATE_TS}'::timestamptz AND " + f"'{LATE_TS}'::timestamptz + interval '10 seconds'"), +] + + +def assert_queries_match(conn): + for desc, q in QUERIES: + got = conn.execute(q.format("events")).fetchall() + want = conn.execute(q.format("events_plain")).fetchall() + assert got == want, f"{desc}: deltax={got} plain={want}" + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestCompressedInsert: + def test_insert_into_compressed_succeeds(self, db): + setup_tables(db) + compress_all(db) + part = data_partition(db) + + insert_late_rows(db) + + # Loose rows are in the partition heap, not yet in segments. (A + # direct count goes through the decompress union, so probe the heap + # physically.) + loose_bytes = db.execute( + f"SELECT pg_relation_size('{part}')" + ).fetchone()[0] + assert loose_bytes > 0 + total = db.execute(f"SELECT count(*) FROM {part}").fetchone()[0] + assert total == 800 + 25 + + # Direct insert into the partition (not via the parent) also works. + db.execute( + f"INSERT INTO {part} (ts, device_id, val, temperature) " + f"VALUES ('{LATE_TS}'::timestamptz + interval '1 hour', 'device-direct', 95000, 1.5)" + ) + db.execute( + "INSERT INTO events_plain (ts, device_id, val, temperature) " + f"VALUES ('{LATE_TS}'::timestamptz + interval '1 hour', 'device-direct', 95000, 1.5)" + ) + db.commit() + assert_tables_match(db) + + def test_read_paths_see_heap_tail(self, db): + setup_tables(db) + compress_all(db) + insert_late_rows(db) + + assert_tables_match(db) + assert_queries_match(db) + + def test_read_paths_see_heap_tail_no_segment_by(self, db): + # Without segment_by the decompress path can claim sorted output + # (pathkeys) — with loose rows it must not, and ORDER BY queries + # must stay correct via an explicit Sort. + setup_tables(db, segment_by=False) + compress_all(db) + insert_late_rows(db) + + assert_tables_match(db) + assert_queries_match(db) + + def test_count_pushdown_still_used_with_heap_tail(self, db): + setup_tables(db) + compress_all(db) + insert_late_rows(db) + + plan = "\n".join( + r[0] for r in db.execute("EXPLAIN SELECT count(*) FROM events").fetchall() + ) + # DeltaXCount folds the heap tail at exec time, so it stays enabled. + assert "DeltaXCount" in plan + got = db.execute("SELECT count(*) FROM events").fetchone()[0] + want = db.execute("SELECT count(*) FROM events_plain").fetchone()[0] + assert got == want + + def test_agg_pushdown_bails_with_heap_tail(self, db): + setup_tables(db) + compress_all(db) + + q = "SELECT device_id, sum(val) FROM events GROUP BY device_id" + plan_before = "\n".join(r[0] for r in db.execute(f"EXPLAIN {q}").fetchall()) + assert "DeltaXAgg" in plan_before + + insert_late_rows(db) + plan_after = "\n".join(r[0] for r in db.execute(f"EXPLAIN {q}").fetchall()) + # The columnar agg cannot ingest loose rows — planner must bail to a + # plain Agg over scans that union the heap tail. + assert "DeltaXAgg" not in plan_after + + def test_update_delete_still_rejected(self, db): + setup_tables(db) + compress_all(db) + part = data_partition(db) + insert_late_rows(db) + + for stmt in ( + f"UPDATE events SET val = 0 WHERE ts >= '{BASE_TS}'::timestamptz", + f"UPDATE {part} SET val = 0", + f"DELETE FROM events WHERE ts >= '{BASE_TS}'::timestamptz", + f"DELETE FROM {part}", + ): + with pytest.raises(Exception) as exc: + db.execute(stmt) + db.rollback() + db.execute(f"SET pg_deltax.mock_now = '{MOCK_NOW}'") + db.commit() + assert "compressed partition" in str(exc.value) + + def test_insert_on_conflict_rejected(self, db): + setup_tables(db) + compress_all(db) + + # ON CONFLICT DO NOTHING without a conflict target needs no unique + # index, but still goes through the ON CONFLICT machinery — which + # cannot see rows inside segments, so it must be rejected. + with pytest.raises(Exception) as exc: + db.execute( + "INSERT INTO events (ts, device_id, val, temperature) " + f"VALUES ('{LATE_TS}', 'x', 1, 1.0) " + "ON CONFLICT DO NOTHING" + ) + db.rollback() + assert "ON CONFLICT" in str(exc.value) + + def test_compact_partition(self, db): + setup_tables(db) + compress_all(db) + part = data_partition(db) + insert_late_rows(db) + + segs_before = db.execute( + f'SELECT count(*) FROM _deltax_compressed."{part.split(".")[-1]}_meta" ' + "WHERE _segment_id > 0" + ).fetchone()[0] + rc_before = db.execute( + "SELECT row_count FROM deltax.deltax_partition WHERE table_name = " + f"'{part.split('.')[-1]}'" + ).fetchone()[0] + + result = db.execute( + f"SELECT deltax.deltax_compact_partition('{part}')" + ).fetchone()[0] + db.commit() + assert "Compacted" in result + assert "25 loose rows" in result + + # Loose region is empty again (compaction truncates it); rows live + # in new segments. + loose_bytes = db.execute( + f"SELECT pg_relation_size('{part}')" + ).fetchone()[0] + assert loose_bytes == 0 + segs_after = db.execute( + f'SELECT count(*) FROM _deltax_compressed."{part.split(".")[-1]}_meta" ' + "WHERE _segment_id > 0" + ).fetchone()[0] + assert segs_after > segs_before + rc_after = db.execute( + "SELECT row_count FROM deltax.deltax_partition WHERE table_name = " + f"'{part.split('.')[-1]}'" + ).fetchone()[0] + assert rc_after == rc_before + 25 + + # Results unchanged after compaction — including point lookups for + # values that only ever existed as loose rows (the partition bloom + # sentinel must have been folded or dropped, never under-covering). + assert_tables_match(db) + assert_queries_match(db) + + def test_compact_then_insert_again(self, db): + setup_tables(db) + compress_all(db) + part = data_partition(db) + + insert_late_rows(db, n=10) + db.execute(f"SELECT deltax.deltax_compact_partition('{part}')") + db.commit() + + # Second wave of loose rows after a compaction. + rows = [ + f"('{LATE_TS}'::timestamptz + interval '{p + 100} seconds', " + f"'device-wave2', {97000 + p}, 7.25)" + for p in range(5) + ] + for name in ("events", "events_plain"): + db.execute( + f"INSERT INTO {name} (ts, device_id, val, temperature) VALUES " + + ", ".join(rows) + ) + db.commit() + + assert_tables_match(db) + assert_queries_match(db) + + result = db.execute( + f"SELECT deltax.deltax_compact_partition('{part}')" + ).fetchone()[0] + db.commit() + assert "Compacted" in result + assert_tables_match(db) + assert_queries_match(db) + + def test_compact_noop_without_loose_rows(self, db): + setup_tables(db) + compress_all(db) + part = data_partition(db) + result = db.execute( + f"SELECT deltax.deltax_compact_partition('{part}')" + ).fetchone()[0] + db.commit() + assert "no loose rows" in result + + def test_compact_uncompressed_partition_is_noop(self, db): + setup_tables(db) + part = db.execute( + "SELECT partition_name FROM deltax.deltax_partition_info('events') " + "WHERE partition_name NOT LIKE '%default%' LIMIT 1" + ).fetchone()[0] + result = db.execute( + f"SELECT deltax.deltax_compact_partition('{part}')" + ).fetchone()[0] + db.commit() + assert "not compressed" in result diff --git a/tests/test_compression.py b/tests/test_compression.py index 66c10ba..38c1cf1 100644 --- a/tests/test_compression.py +++ b/tests/test_compression.py @@ -2629,7 +2629,8 @@ def test_minmax_explain_analyze(self, db): # --------------------------------------------------------------------------- class TestDMLBlocking: - """Verify that INSERT/UPDATE/DELETE on compressed partitions raise errors.""" + """Verify DML semantics on compressed partitions: UPDATE/DELETE raise + errors; INSERT succeeds transparently (P1 heap-tail).""" def _setup_and_compress(self, db): """Create table, insert data, compress a partition. Return partition name.""" @@ -2654,14 +2655,25 @@ def _setup_and_compress(self, db): db.commit() return part_name - def test_insert_blocked_on_compressed(self, db): - """INSERT into a compressed partition raises an error.""" + def test_insert_into_compressed_lands_in_heap_tail(self, db): + """INSERT into a compressed partition succeeds (P1 transparent + INSERT): the row lands in the partition heap and is immediately + visible through the decompress union.""" part_name = self._setup_and_compress(db) - with pytest.raises(Exception, match="cannot INSERT into compressed partition"): - db.execute( - f"INSERT INTO \"{part_name}\" (ts, device_id, temperature, pressure, status) " - f"VALUES ('2025-01-15 06:00:00+00', 'dev-new', 99.0, 1000.0, true)" - ) + db.execute( + f"INSERT INTO \"{part_name}\" (ts, device_id, temperature, pressure, status) " + f"VALUES ('2025-01-15 06:00:00+00', 'dev-new', 99.0, 1000.0, true)" + ) + db.commit() + count = db.execute( + f"SELECT count(*) FROM \"{part_name}\" WHERE device_id = 'dev-new'" + ).fetchone()[0] + assert count == 1 + # The loose row physically sits in the partition heap. + loose_bytes = db.execute( + f"SELECT pg_relation_size('{part_name}')" + ).fetchone()[0] + assert loose_bytes > 0 def test_update_blocked_on_compressed(self, db): """UPDATE on a compressed partition raises an error.""" From d46524a5f994924af14c7ebc118439ab2a18bb4c Mon Sep 17 00:00:00 2001 From: Alexis Rico Date: Fri, 12 Jun 2026 15:29:47 +0200 Subject: [PATCH 2/3] fix: harden DML P1 correctness guards (review pass) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent correctness/cleanup pass over the P1 transparent-INSERT work: - Close the data-modifying-CTE bypass of the INSERT ... ON CONFLICT rejection: a wCTE hides the ModifyTable in PlannedStmt.subplans under a top-level CMD_SELECT, so the ExecutorStart hook never saw it. Conflict inference against empty leaf indexes would silently miss conflicts with segment rows (duplicate inserts under ON CONFLICT DO NOTHING). New integration test covers it. - Guard the leader-only Phase 3 heap tail against parallel_leader_participation=off: Gather never drives the leader's plan copy when workers launch, so the tail would be silently dropped. The begin-time guard errors with a remedy instead. - Re-validate the partition catalog row under the AccessExclusive lock in deltax_compact_partition(): a concurrent decompress committing between the unlocked is_compressed check and the lock grant would have made compaction treat the fully-restored heap as loose rows and truncate real data. - Self-heal dead loose-row pages in the loose_rows == 0 compaction branch: a REPEATABLE READ compaction deletes (not truncates) and autovacuum is off on compressed partitions, so the heap would otherwise stay at nonzero blocks forever — pinning scans on the heap-tail path and re-taking the AEL every worker cycle. - Reject json_extract tables in compaction with a clear remedy (synthetic columns have no heap presence — the cursor would fail with a raw SQL error) and skip them in worker auto-compaction to avoid an error loop. - Disambiguate partition_oid_for_companion() by is_compressed: two deltatables with the same table name in different schemas both register same-named partitions in the catalog; the reverse lookup could return the wrong schema's heap and union foreign rows into the scan. - Use the executor snapshot (es_snapshot) instead of GetActiveSnapshot() for the DeltaXCount/DeltaXMinMax heap-tail fold, matching the DeltaXDecompress/Append Phase 3 scan. - Tighten the Phase 3 deform guard to also check the scan-slot tupdesc: for DeltaXAppend the slot is the parent's layout, and a dropped column on either side would silently misalign the positional deform. --- src/compress.rs | 63 ++++++++++++++++++++++++++++++++- src/scan/exec/count_minmax.rs | 7 ++-- src/scan/exec/decompress.rs | 21 +++++++++-- src/scan/exec/segments.rs | 6 +++- src/scan/hook.rs | 58 +++++++++++++++++++++++------- tests/test_compressed_insert.py | 18 ++++++++++ 6 files changed, 153 insertions(+), 20 deletions(-) diff --git a/src/compress.rs b/src/compress.rs index 574f6bb..4c89a2c 100644 --- a/src/compress.rs +++ b/src/compress.rs @@ -4351,6 +4351,27 @@ pub(crate) fn compact_partition_impl(client: &mut SpiClient, partition: &str) -> ) .expect("failed to lock partition for compaction"); + // Re-read the catalog row now that the lock is held: a concurrent + // decompress could have flipped is_compressed between the unlocked + // check above and the lock grant — compacting then would treat the + // fully-restored heap as "loose rows" and truncate real data. The + // re-read also pins compressed_columns for the shape guard below. + let part_info = catalog::get_partition_by_name(client, &schema, &part_table) + .expect("failed to re-query partition") + .unwrap_or_else(|| { + pgrx::error!( + "pg_deltax: partition {}.{} disappeared from catalog during compaction", + schema, + part_table + ) + }); + if !part_info.is_compressed { + return format!( + "Partition {}.{} is not compressed; nothing to compact", + schema, part_table + ); + } + let loose_rows: i64 = client .select( &format!("SELECT count(*) FROM ONLY {}", part_fqn), @@ -4364,6 +4385,30 @@ pub(crate) fn compact_partition_impl(client: &mut SpiClient, partition: &str) -> .flatten() .unwrap_or(0); if loose_rows == 0 { + // A prior REPEATABLE READ compaction deletes (rather than truncates) + // the rows it compacted, and autovacuum is disabled on compressed + // partitions — so dead rows would keep the heap at nonzero blocks + // forever, pinning every scan on the slower heap-tail path and + // re-locking here each worker cycle. With the AEL held and zero + // visible rows, truncating away the dead tuples is the same MVCC + // tradeoff as the compress-time TRUNCATE. + let read_committed = + unsafe { pg_sys::XactIsoLevel } < pg_sys::XACT_REPEATABLE_READ as std::ffi::c_int; + let heap_bytes: i64 = client + .select( + &format!("SELECT pg_relation_size('{}'::regclass)", part_fqn), + Some(1), + &[], + ) + .ok() + .and_then(|r| r.first().get_one::().ok().flatten()) + .unwrap_or(0); + if read_committed && heap_bytes > 0 { + client + .update(&format!("TRUNCATE ONLY {}", part_fqn), None, &[]) + .expect("failed to truncate dead loose-row region"); + crate::scan::invalidate_compressed_cache(); + } return format!( "Partition {}.{} has no loose rows to compact", schema, part_table @@ -4404,6 +4449,17 @@ pub(crate) fn compact_partition_impl(client: &mut SpiClient, partition: &str) -> if columns.is_empty() { pgrx::error!("pg_deltax: no columns found for {}.{}", schema, part_table); } + // Synthetic json_extract columns don't exist in the partition heap, so + // the loose-row cursor below cannot read them (and the companion layout + // expects them). Bail with the supported remedy instead of a raw SQL + // error from the cursor. + if columns.iter().any(|c| c.extracted.is_some()) { + pgrx::error!( + "pg_deltax: partition {}.{} uses json_extract; compacting loose rows is not supported — run deltax_decompress_partition() + deltax_compress_partition() instead", + schema, + part_table + ); + } let ddl = build_companion_ddl(&part_table, &columns); if !relation_exists(client, &ddl.meta_fqn) { @@ -5125,6 +5181,12 @@ fn merge_batch_hll_into_catalog( /// timeout or transient failure on one partition doesn't abort the worker's /// whole maintenance cycle. pub fn auto_compact_partitions(client: &mut SpiClient<'_>, ht: &catalog::DeltatableInfo) -> i32 { + // json_extract deltatables can't be compacted (synthetic columns have no + // heap presence — compact_partition_impl rejects them); skip instead of + // error-logging every worker cycle. + if ht.json_extract.is_some() && crate::get_json_extract_mode() != crate::JsonExtractMode::None { + return 0; + } let candidates = client .select( "SELECT p.schema_name, p.table_name @@ -5187,7 +5249,6 @@ pub fn auto_compact_partitions(client: &mut SpiClient<'_>, ht: &catalog::Deltata compacted } - /// Re-populate pg_class.reltuples + pg_statistic for an already-compressed /// partition. `stats::analyze_partition_from_catalog` reads the /// authoritative per-column distinct counts persisted at compression time diff --git a/src/scan/exec/count_minmax.rs b/src/scan/exec/count_minmax.rs index e8efc6a..803755d 100644 --- a/src/scan/exec/count_minmax.rs +++ b/src/scan/exec/count_minmax.rs @@ -124,7 +124,9 @@ unsafe fn for_each_heap_tail_row( }; let econtext = (*node).ss.ps.ps_ExprContext; - let snapshot = pg_sys::GetActiveSnapshot(); + // The executor's query snapshot — same visibility as the rest of + // the plan (segment meta rows, sibling scans). + let snapshot = (*(*node).ss.ps.state).es_snapshot; for &oid in &heap_oids { let rel = pg_sys::table_open(oid, pg_sys::AccessShareLock as pg_sys::LOCKMODE); let tupdesc = (*rel).rd_att; @@ -278,7 +280,7 @@ pub(super) unsafe extern "C-unwind" fn begin_count_scan( } } - let state = if catalog_hit { + let mut state = if catalog_hit { // Approximate segment count from pg_class.reltuples (one row per segment). let mut total_segments: u64 = 0; for &oid in &companion_oids { @@ -361,7 +363,6 @@ pub(super) unsafe extern "C-unwind" fn begin_count_scan( // P1 heap tail: add the loose rows sitting in the partition heaps // (transparent INSERTs after compression), filtered through the // full qual list. Zero-cost when every heap is empty. - let mut state = state; { let expected_natts = if qual_bytes.is_empty() { 0 // COUNT(*) without quals reads no columns diff --git a/src/scan/exec/decompress.rs b/src/scan/exec/decompress.rs index 4f28f0c..f11034f 100644 --- a/src/scan/exec/decompress.rs +++ b/src/scan/exec/decompress.rs @@ -879,6 +879,17 @@ pub(super) unsafe extern "C-unwind" fn begin_deltax_append( // fall back to the full scan (upper Sort/Limit stays correct — // DeltaXAppend never claims pathkeys). topn_limit = 0; + // The heap tail is emitted by the leader's plan copy only. With + // parallel_leader_participation=off and workers launched, Gather + // never drives the leader's copy, so the tail would be silently + // dropped. Error instead — never wrong results. + if (*(*node).ss.ps.plan).parallel_aware && !pg_sys::parallel_leader_participation { + pgrx::error!( + "pg_deltax: a compressed partition holds uncompressed rows, which a parallel \ + scan can only return with parallel_leader_participation=on; re-enable it or \ + run deltax_compact_partition() and retry" + ); + } } // Load metadata for first companion table (cached per-backend). @@ -3600,14 +3611,20 @@ unsafe fn try_emit_heap_tail_row( let oid = ht.oids[ht.idx]; let rel = pg_sys::table_open(oid, pg_sys::AccessShareLock as pg_sys::LOCKMODE); let rel_natts = (*(*rel).rd_att).natts as usize; - if rel_natts != ncols { + // The slot tupdesc is the partition's for DeltaXDecompress + // but the PARENT's for DeltaXAppend — plan Vars index it, so + // all three layouts must agree before deforming positionally + // (a dropped column on one side would silently misalign). + let slot_natts = (*(*scan_slot).tts_tupleDescriptor).natts as usize; + if rel_natts != ncols || slot_natts != rel_natts { let name = relation_name(rel); pg_sys::table_close(rel, pg_sys::AccessShareLock as pg_sys::LOCKMODE); pgrx::error!( - "pg_deltax: compressed partition \"{}\" has uncompressed rows but its physical layout ({} attrs) does not match the compressed layout ({} cols); decompress and recompress it", + "pg_deltax: compressed partition \"{}\" has uncompressed rows but its physical layout ({} attrs) does not match the compressed layout ({} cols, {} scan attrs); decompress and recompress it", name, rel_natts, ncols, + slot_natts, ); } let flags: u32 = pg_sys::ScanOptions::SO_TYPE_SEQSCAN diff --git a/src/scan/exec/segments.rs b/src/scan/exec/segments.rs index 558c71c..df27a82 100644 --- a/src/scan/exec/segments.rs +++ b/src/scan/exec/segments.rs @@ -323,6 +323,10 @@ pub(crate) fn partition_oid_for_companion(companion_oid: pg_sys::Oid) -> pg_sys: None => return pg_sys::InvalidOid, } }; + // `is_compressed` disambiguates same-named partitions across schemas: + // companions live in the single `_deltax_compressed` namespace, so at + // most one partition of a given name can be compressed (and it is the + // one this companion belongs to). let oid = pgrx::Spi::connect(|client| { client .select( @@ -331,7 +335,7 @@ pub(crate) fn partition_oid_for_companion(companion_oid: pg_sys::Oid) -> pg_sys: JOIN pg_namespace n ON n.oid = c.relnamespace JOIN deltax.deltax_partition p ON p.schema_name = n.nspname AND p.table_name = c.relname - WHERE p.table_name = $1", + WHERE p.table_name = $1 AND p.is_compressed", Some(1), &[partition_name.into()], ) diff --git a/src/scan/hook.rs b/src/scan/hook.rs index 15bbdea..4811174 100644 --- a/src/scan/hook.rs +++ b/src/scan/hook.rs @@ -4725,12 +4725,17 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( ) { unsafe { let operation = (*query_desc).operation; + let planned_stmt = (*query_desc).plannedstmt; - // Only check DML commands - if operation != pg_sys::CmdType::CMD_INSERT - && operation != pg_sys::CmdType::CMD_UPDATE - && operation != pg_sys::CmdType::CMD_DELETE - { + // Only check DML commands — plus statements whose data-modifying + // CTEs hide an INSERT ... ON CONFLICT (their top-level operation + // is CMD_SELECT, but the conflict inference is just as blind to + // segment rows as a top-level one). + let is_dml = operation == pg_sys::CmdType::CMD_INSERT + || operation == pg_sys::CmdType::CMD_UPDATE + || operation == pg_sys::CmdType::CMD_DELETE; + let insert_on_conflict = stmt_has_on_conflict(planned_stmt); + if !is_dml && !insert_on_conflict { call_prev_executor_start(query_desc, eflags); return; } @@ -4741,7 +4746,6 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( return; } - let planned_stmt = (*query_desc).plannedstmt; if planned_stmt.is_null() { call_prev_executor_start(query_desc, eflags); return; @@ -4758,13 +4762,6 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( // the segment data. INSERT ... ON CONFLICT stays rejected: // segment rows have no index entries, so conflict inference // against compressed data would be silently wrong. - let insert_on_conflict = operation == pg_sys::CmdType::CMD_INSERT && { - let plan_tree = (*planned_stmt).planTree; - !plan_tree.is_null() - && (*plan_tree).type_ == pg_sys::NodeTag::T_ModifyTable - && (*(plan_tree as *mut pg_sys::ModifyTable)).onConflictAction - != pg_sys::OnConflictAction::ONCONFLICT_NONE - }; if operation == pg_sys::CmdType::CMD_INSERT && !insert_on_conflict { call_prev_executor_start(query_desc, eflags); return; @@ -4831,6 +4828,41 @@ pub unsafe extern "C-unwind" fn deltax_executor_start( } } +/// True when any ModifyTable in the statement uses ON CONFLICT — the +/// top-level plan, or a data-modifying CTE (those are planned into +/// `PlannedStmt.subplans`, so a wrapping SELECT/UPDATE would otherwise +/// smuggle the conflict inference past the top-level `operation` check). +unsafe fn stmt_has_on_conflict(planned_stmt: *mut pg_sys::PlannedStmt) -> bool { + unsafe fn is_on_conflict_modifytable(plan: *mut pg_sys::Plan) -> bool { + unsafe { + !plan.is_null() + && (*plan).type_ == pg_sys::NodeTag::T_ModifyTable + && (*(plan as *mut pg_sys::ModifyTable)).onConflictAction + != pg_sys::OnConflictAction::ONCONFLICT_NONE + } + } + unsafe { + if planned_stmt.is_null() { + return false; + } + if is_on_conflict_modifytable((*planned_stmt).planTree) { + return true; + } + if (*planned_stmt).hasModifyingCTE { + let subplans = (*planned_stmt).subplans; + if !subplans.is_null() { + for i in 0..(*subplans).length { + if is_on_conflict_modifytable(pg_sys::list_nth(subplans, i) as *mut pg_sys::Plan) + { + return true; + } + } + } + } + false + } +} + /// True when `relid` is a deltax-managed parent table with at least one /// compressed partition. Used only on the (rare) INSERT ... ON CONFLICT /// path, so the SPI probe is acceptable. diff --git a/tests/test_compressed_insert.py b/tests/test_compressed_insert.py index ec5bf74..e90ccc8 100644 --- a/tests/test_compressed_insert.py +++ b/tests/test_compressed_insert.py @@ -270,6 +270,24 @@ def test_insert_on_conflict_rejected(self, db): db.rollback() assert "ON CONFLICT" in str(exc.value) + def test_insert_on_conflict_in_cte_rejected(self, db): + # A data-modifying CTE hides the INSERT under a top-level SELECT; + # the ON CONFLICT rejection must still fire (conflict inference is + # just as blind to segment rows as a top-level INSERT). + setup_tables(db) + compress_all(db) + + with pytest.raises(Exception) as exc: + db.execute( + "WITH ins AS (" + " INSERT INTO events (ts, device_id, val, temperature) " + f" VALUES ('{LATE_TS}', 'x', 1, 1.0) " + " ON CONFLICT DO NOTHING RETURNING 1" + ") SELECT count(*) FROM ins" + ) + db.rollback() + assert "ON CONFLICT" in str(exc.value) + def test_compact_partition(self, db): setup_tables(db) compress_all(db) From a7fd85d20806cc33bd6646cfc59ef70082c9faf2 Mon Sep 17 00:00:00 2001 From: Alexis Rico Date: Fri, 12 Jun 2026 16:29:50 +0200 Subject: [PATCH 3/3] test: update RTABench correctness test for P1 transparent INSERT MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pre-existing correctness test asserted the P0 contract (parent-routed INSERT into a compressed partition raises). DML P1 intentionally accepts plain INSERTs — rows land in the partition heap and are unioned with segment data at scan time. Update the test to insert into both the plain and deltax tables, verify the loose row is visible through the deltax read path (count + total parity), and keep the join-equality check. UPDATE/DELETE rejection coverage is unchanged. --- tests/correctness/test_rtabench_joins.py | 64 ++++++++++++++---------- 1 file changed, 38 insertions(+), 26 deletions(-) diff --git a/tests/correctness/test_rtabench_joins.py b/tests/correctness/test_rtabench_joins.py index 0c4c1e9..7617765 100644 --- a/tests/correctness/test_rtabench_joins.py +++ b/tests/correctness/test_rtabench_joins.py @@ -82,44 +82,56 @@ def _rtabench_case(name: str): ("copy_csv_mixed_s25", "copy_csv", 25), ), ) -def test_rtabench_compressed_partition_rejects_parent_routed_insert( +def test_rtabench_compressed_partition_accepts_parent_routed_insert( db, layout_name, load_path, segment_size, ): + # P1 transparent DML: a plain INSERT routed through the parent into a + # compressed partition succeeds — the row lands in the partition heap + # (the "loose row" region) and every read path unions it with the + # segment data. plain_table, deltax_table = create_rtabench_synthetic_pair( db, deltax_table=f"rtabench_events_{layout_name}", load_path=load_path, segment_size=segment_size, ) - with pytest.raises(Exception, match="compressed partition"): - db.execute( - f""" - INSERT INTO {deltax_table} ( - order_id, - counter, - event_created, - event_type, - satisfaction, - processor, - backup_processor, - event_payload - ) - VALUES ( - 181, - 9001, - '2024-05-04 12:00:00+00', - 'Delivered', - 4.5, - 'proc-a', - 'proc-b', - '{{}}'::jsonb - ) - """ + insert_template = """ + INSERT INTO {table} ( + order_id, + counter, + event_created, + event_type, + satisfaction, + processor, + backup_processor, + event_payload ) - db.rollback() + VALUES ( + 181, + 9001, + '2024-05-04 12:00:00+00', + 'Delivered', + 4.5, + 'proc-a', + 'proc-b', + '{{}}'::jsonb + ) + """ + db.execute(insert_template.format(table=deltax_table)) + db.execute(insert_template.format(table=plain_table)) + db.commit() + + # The loose row must be visible through the deltax read path. + visible = db.execute( + f"SELECT count(*) FROM {deltax_table} WHERE counter = 9001" + ).fetchone()[0] + assert visible == 1 + deltax_total = db.execute(f"SELECT count(*) FROM {deltax_table}").fetchone()[0] + plain_total = db.execute(f"SELECT count(*) FROM {plain_table}").fetchone()[0] + assert deltax_total == plain_total assert_query_case( db,