You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/08/19 05:38:50 UTC
(pinot) 01/01: Add codec pipeline integration tests and design doc
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch xiangfu0/codex/codec-stack/07-integration-docs
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 88e1d409026c7f75afec2e25837a9508e912d461
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Tue Aug 18 22:28:29 2026 -0700
Add codec pipeline integration tests and design doc
---
docs/design/codec-pipeline-v7.md | 511 +++++++++++++++++++++
.../tests/custom/CodecPipelineIntegrationTest.java | 426 +++++++++++++++++
2 files changed, 937 insertions(+)
diff --git a/docs/design/codec-pipeline-v7.md b/docs/design/codec-pipeline-v7.md
new file mode 100644
index 00000000000..09a84132d9b
--- /dev/null
+++ b/docs/design/codec-pipeline-v7.md
@@ -0,0 +1,511 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# Codec Pipeline Framework — Forward Index Format V7
+
+**Status**: Implemented (stacked PR series splitting PR [#18229](https://github.com/apache/pinot/pull/18229))
+**Module**: `pinot-segment-spi/codec`, `pinot-segment-local/io/codec`, `pinot-segment-local/.../FixedByteChunkForwardIndexWriterV7`, `FixedByteChunkSVForwardIndexReaderV7`
+**Scope (v1)**: A single legacy-compatible compression invocation uses the existing raw
+forward-index format. Specs that cannot map to one legacy `ChunkCompressionType` (any transform,
+multiple compression stages, or non-default compression options) use V7 for single-value INT/LONG.
+**Wire-format identifier**: version 7 plus the explicit codec-pipeline header magic
+
+> **Note on placement**: this document lives under `docs/design/` for now; the final location is
+> subject to reviewer preference (e.g. the Pinot documentation site or a wiki page instead of the
+> source tree).
+
+---
+
+## 1. Goals and non-goals
+
+### Goals
+1. Replace the closed `FieldConfig.CompressionCodec` enum with an extensible **codec pipeline DSL** so new codecs (or new combinations) can be added without churning the public enum.
+2. Introduce a **multi-stage pipeline** that lets users compose a transform (e.g. DELTA, DELTADELTA) with a compression codec (e.g. LZ4, ZSTD) — covering use cases the single-codec enum cannot express cleanly.
+3. Embed the **canonical codec spec** in V7 pipeline segment headers so readers can decode without out-of-band configuration; V7 segments are self-describing.
+4. Preserve full **backward compatibility**: legacy `compressionCodec` continues to work; `codecSpec` is opt-in per column.
+5. Provide a clear **migration path** from the legacy enum to the DSL.
+
+### Non-goals (v1)
+- Variable-width or multi-value V7 pipeline storage. Only single-stage compression specs that map to a legacy `ChunkCompressionType` use the existing raw writer formats (and therefore already work for any SV/MV fixed/var-byte column); all V7-requiring specs are fixed-byte SV INT/LONG only.
+- Streaming / dictionary-encoded indexes. `codecSpec` applies to **raw** forward indexes only.
+- A SQL-visible codec function. The DSL lives in table config (`fieldConfigList[].indexes.forward.codecSpec`).
+- Plugin-registered codecs. v1 ships with 8 built-in codecs; plugin registration is a v2 follow-up.
+
+---
+
+## 2. DSL grammar
+
+A codec spec is a comma-separated list of codec invocations, evaluated left-to-right on encode:
+
+```
+spec ::= invocation ("," invocation)*
+invocation ::= NAME | NAME "(" args ")"
+args ::= ε | arg ("," arg)*
+arg ::= [0-9]+ (* unsigned integer; signs rejected *)
+NAME ::= [A-Za-z_][A-Za-z0-9_]* (* ASCII only, locale-stable *)
+```
+
+- Case-insensitive lookup (`zstd(3)` ≡ `ZSTD(3)`).
+- The name `CODEC` is **permanently reserved** and may not be used as a codec name. An earlier
+ draft of this DSL wrapped multi-stage specs in a `CODEC(...)` call; that wrapper was removed in
+ favor of the plain comma-separated list, and the parser rejects it with a precise error
+ (`CODEC(...) wrapper is not supported; list codec invocations directly`).
+- Whitespace is allowed between tokens; not allowed inside identifiers or numbers.
+- A pipeline must contain at least one stage.
+- To bound parse work for table configs and untrusted segment headers, a spec is limited to 64 KiB,
+ 32 stages, 128 characters per identifier, 16 arguments per stage, and 32 digits per argument.
+- A pipeline is a chain of the form **N typed-layout-preserving transforms → at most one packing transform → N compressions**:
+ - **Typed-layout-preserving transforms** (`DELTA`, `DELTADELTA`) map a column-typed value array to a same-width value array (header-less passthrough; element type comes from the column context, value count from the buffer length). Any number may be chained, so `DELTA,DELTADELTA,LZ4` is valid.
+ - **Packing transforms** (`T64`, `GORILLA`) emit a bit-packed, self-framed byte stream that is no longer a typed value array, so a packing transform must be the **last** transform — only compression stages may follow it (e.g. `DELTA,T64,LZ4`).
+ - **Compression** stages (`LZ4`, `ZSTD`, `SNAPPY`, `GZIP`) are byte→byte; any number may follow the transforms (e.g. `DELTA,LZ4,ZSTD(3)`).
+ - The validator enforces this by tracking a "typed-value domain": a `TRANSFORM` may only appear while still in the typed domain (it cannot follow a packing transform or a compression stage). It runs at table-config validation time so bad configs do not reach ZooKeeper.
+- **Evaluation order:** stages run **left-to-right on encode** and **right-to-left on decode**. For example, `DELTA,T64,LZ4` gives `encode(x) = LZ4.encode(T64.encode(DELTA.encode(x)))` and `decode(y) = DELTA.decode(T64.decode(LZ4.decode(y)))`.
+- **Canonical form:** the parser normalizes names to upper case and strips whitespace;
+ `CodecPipeline.toDslString()` / `CodecPipelineExecutor.getCanonicalSpec()` emit the canonical
+ comma-joined spec (e.g. `" delta , zstd( 3 ) "` canonicalizes to `DELTA,ZSTD(3)`). The canonical
+ form is what gets stored in `ForwardIndexConfig` and embedded in V7 segment headers.
+
+### Examples
+
+| DSL | Stages | Notes |
+|----------------------------|---------------------------------|-----------------------------------------|
+| `LZ4` | LZ4 (compression) | Single-stage compression |
+| `ZSTD(3)` | ZSTD level 3 (compression) | Compression with level argument |
+| `SNAPPY` | Snappy (compression) | |
+| `GZIP` | GZIP / DEFLATE (compression) | Slower than LZ4/ZSTD |
+| `DELTA` | DELTA (transform) | INT/LONG only, no compression |
+| `DELTADELTA` | DELTADELTA (transform) | Second-order delta; good for timestamps |
+| `DELTA,LZ4` | DELTA → LZ4 | Common timestamp pipeline |
+| `DELTA,ZSTD(3)` | DELTA → ZSTD level 3 | Better ratio than LZ4 at higher CPU |
+| `DELTADELTA,LZ4` | DELTADELTA → LZ4 | Best for monotonic timestamps |
+| `T64` | T64 bit-pack (transform) | Frame-of-reference bit-packing on 64-value blocks; INT/LONG only |
+| `T64,LZ4` | T64 → LZ4 | Bit-pack then byte-compress |
+| `GORILLA` | Gorilla XOR (transform) | XOR-delta bit-stream; INT/LONG only |
+| `GORILLA,ZSTD(3)` | Gorilla → ZSTD level 3 | XOR + entropy coding |
+| `DELTA,DELTADELTA,LZ4` | DELTA → DELTADELTA → LZ4 | Chained typed-layout transforms + compression |
+| `DELTA,T64,LZ4` | DELTA → T64 bit-pack → LZ4 | Delta then frame-of-reference pack then compress |
+| `DELTA,LZ4,ZSTD(3)` | DELTA → LZ4 → ZSTD level 3 | Chained compressions after a transform |
+
+---
+
+## 3. Architecture
+
+```
+ pinot-segment-spi/codec/ (interfaces + AST)
+ +------------------------------+
+ | CodecKind (TRANSFORM, |
+ | COMPRESSION) |
+ | CodecOptions (marker) |
+ | CodecContext (DataType ctx) |
+ | CodecInvocation (name, args) |
+ | CodecPipeline (List<Inv>) |
+ | CodecSpecParser (recursive |
+ | descent → CodecPipeline) |
+ | CodecDefinition (parseOpts, |
+ | canonicalize, kind) |
+ | ChunkCodecHandler |
+ | extends CodecDefinition: |
+ | encode(opts, ctx, src) |
+ | decode(opts, ctx, src) |
+ | decodeInto(opts, ctx, |
+ | src, dst) |
+ | maxEncodedSize(opts, n) |
+ | requiresDirectDstBuffer() |
+ +-------------|----------------+
+ |
+ v
+ pinot-segment-local/io/codec/ (concrete handlers + runtime)
+ +------------------------------------------------------------+
+ | DeltaCodecDefinition (TRANSFORM, INT/LONG) |
+ | DeltaDeltaCodecDefinition (TRANSFORM, INT/LONG) |
+ | extends BaseDeltaCodecDefinition (shared scaffold) |
+ | T64CodecDefinition (packing TRANSFORM, INT/LONG)|
+ | GorillaCodecDefinition (packing TRANSFORM, INT/LONG)|
+ | Lz4CodecDefinition (COMPRESSION, lazy native) |
+ | ZstdCodecDefinition (COMPRESSION, level 1..22) |
+ | SnappyCodecDefinition (COMPRESSION) |
+ | GzipCodecDefinition (COMPRESSION, direct |
+ | ByteBuffer I/O) |
+ | |
+ | CodecRegistry (immutable DEFAULT; |
+ | mutable @VisibleForTesting)|
+ | CodecPipelineValidator (structural rules: |
+ | N typed-layout transforms, |
+ | ≤1 packing transform, |
+ | N compressions) |
+ | CodecPipelineExecutor (binds pipeline → handlers; |
+ | thread-safe; |
+ | encode/decode; |
+ | getCanonicalSpec) |
+ | CodecSpecUtils (legacy-compatibility |
+ | classification) |
+ | CodecBufferUtils (toDirectBuffer helpers) |
+ +------------|-----------------------------------------------+
+ |
+ v
+ +-----------------------------------------------------------+
+ | FixedByteChunkForwardIndexWriterV7 |
+ | - writes self-describing V7 segments (header embeds |
+ | canonicalSpec) |
+ | FixedByteChunkSVForwardIndexReaderV7 |
+ | - dispatched by factory on version=7 + format magic |
+ | - reads canonicalSpec from header → builds executor |
+ | - validates header bounds, monotonic chunk offsets |
+ | ForwardIndexCreatorFactory |
+ | - one legacy-compatible compression → existing writers |
+ | - every other codecSpec → V7 fixed-byte writer |
+ +-----------------------------------------------------------+
+```
+
+**Module placement rationale**: SPI types are pure interfaces + AST + parser (no JNI, no runtime state). Concrete codec handlers and registry/executor live in `pinot-segment-local` because they reference native libraries (LZ4/ZSTD/Snappy) and PinotDataBuffer — both of which are local-module dependencies.
+
+---
+
+## 4. On-disk format (V7)
+
+The codec-pipeline V7 file is **self-describing**: every reader can determine the codec spec from the
+file header alone. There is no out-of-band configuration required at read time. Version 7 was already
+a valid legacy fixed-byte writer version, so the version field alone is not a format discriminator.
+Pipeline V7 adds the explicit `0xC0DEC0DE` magic immediately after the version. The reader factory
+uses only that stable marker to select the pipeline reader, whose constructor then validates every
+remaining field; other fixed-byte versions `>= 4`, including legacy V7, continue to use the legacy
+reader.
+
+```
+┌─────────────────────────────────────────────────────────────────────┐
+│ FILE HEADER │
+├─────────────────────────────────────────────────────────────────────┤
+│ Offset Field Size Notes │
+│ ─────── ──────────────── ───── ───────────────────────────────── │
+│ 0 version int(4) = 7 │
+│ 4 formatMagic int(4) = 0xC0DEC0DE │
+│ 8 numChunks int(4) ≥ 0 │
+│ 12 numDocsPerChunk int(4) power of 2, ≥ 1 │
+│ 16 sizeOfEntry int(4) 4 (INT) or 8 (LONG) │
+│ 20 totalDocs int(4) ≥ 0 │
+│ 24 codecSpecLength int(4) > 0 │
+│ 28 dataHeaderStart int(4) = 32 + codecSpecLength │
+│ 32 codecSpec[] byte[] UTF-8 canonical DSL │
+│ X chunkOffsets[] long[numChunks] absolute offsets, monotonic│
+├─────────────────────────────────────────────────────────────────────┤
+│ DATA SECTION (one entry per chunk) │
+├─────────────────────────────────────────────────────────────────────┤
+│ Y encodedSize int(4) │
+│ Y+4 decodedSize int(4) │
+│ Y+8 payload byte[encodedSize] (codec output) │
+└─────────────────────────────────────────────────────────────────────┘
+
+X = 32 + codecSpecLength
+Y = chunkOffsets[i]
+```
+
+### Reader-side validation (corruption defense)
+
+| Check | Failure |
+|-------------------------------------------------------------------|--------------------------------------|
+| `version == 7` and `formatMagic == 0xC0DEC0DE` | `IllegalArgumentException` |
+| `numChunks ≥ 0` | `IllegalArgumentException` |
+| `numDocsPerChunk` is a power of two | `IllegalArgumentException` |
+| `sizeOfEntry == storedType.size()` | `IllegalArgumentException` |
+| `totalDocs ≥ 0` | `IllegalArgumentException` |
+| `0 < codecSpecLength ≤ 64 KiB` and `dataHeaderStart == 32 + length` | `IllegalArgumentException` |
+| `dataHeaderStart` and chunk-offset table fit in buffer | `IllegalArgumentException` |
+| `chunkOffsets[]` strictly monotonic and leave room for a header | `IllegalArgumentException` |
+| Per-chunk `encodedSize ≥ 0` and fits before next chunk offset | `IllegalStateException` ("corrupt") |
+| `decodedSize` exactly matches rows in that chunk × entry size | `IllegalStateException` ("corrupt") |
+| Decoded chunks stay within 64 MiB, each encoded/intermediate bound within 128 MiB, and cumulative stage-output bounds within 256 MiB | `IOException` / `IllegalArgumentException` |
+| Each codec consumes a complete, valid frame (including GZIP checksum/trailer) | `IOException` ("corrupt") |
+
+---
+
+## 5. Codec catalog (built-in)
+
+| Codec | `CodecKind` | Args | Wire format | Notes |
+|--------------|--------------|--------------|----------------------------------------------------------------------|--------------------------------------------------------------------------|
+| `DELTA` | TRANSFORM | none | `[first:N][delta_i:N for i=1..count-1]` (header-less passthrough; type from column ctx, count from length) | Typed-layout-preserving (chainable); two's-complement wrap intentional, symmetric on decode (locked by tests) |
+| `DELTADELTA` | TRANSFORM | none | `[first:N][firstDelta:N][dod_i:N for i=2..]` (header-less passthrough) | Typed-layout-preserving (chainable); same wrap semantics |
+| `T64` | TRANSFORM | none | `[flag:1B][count:4B]` + per-64-value block `[baseline:N][bitWidth:1B][packed:ceil(bitWidth*64/8)B]` | Frame-of-reference + bit-packing on fixed 64-value blocks |
+| `GORILLA` | TRANSFORM | none | `[flag:1B][count:4B][first:N][bit-stream]` | XOR-delta with MSB-first bit-stream, reusing previous leading/width window when it fits |
+| `LZ4` | COMPRESSION | none | LZ4 length-prefixed | `LZ4Factory.fastestInstance()` lazy init via inner holder class |
+| `ZSTD` | COMPRESSION | `level` (int) | Zstd frame with embedded decompressedSize | Levels 1–22; default 3. `ZSTANDARD` accepted as an alias, canonicalized to `ZSTD` |
+| `SNAPPY` | COMPRESSION | none | xerial Snappy | JNI requires direct buffers (handled internally) |
+| `GZIP` | COMPRESSION | none | DEFLATE payload + 4-byte decompressed-size footer | ThreadLocal `Deflater`/`Inflater`; direct `ByteBuffer` input and output |
+
+**Frozen on-disk names**: All codec `NAME` constants (DELTA, DELTADELTA, T64, GORILLA, LZ4, ZSTD, SNAPPY, GZIP) are part of the on-disk format contract and must never be changed. The reserved name `CODEC` (the removed wrapper keyword) must never be reused as a codec name, so that old spec strings fail loudly rather than resolving to something else.
+
+**Rolling upgrade considerations**:
+
+1. **The `codecSpec` property requires upgraded config consumers.** Pre-1.6 controllers, servers,
+ minions, or external segment builders can reject the unknown nested JSON property before segment-format
+ compatibility is relevant. Upgrade every component that validates or consumes the table config before
+ enabling any `codecSpec`, including a legacy-format spec such as `LZ4` or `ZSTD(3)`.
+
+2. **V7 segments are unreadable by pre-V7 servers.** Servers built before this change use
+ `>= VERSION 4` to dispatch version 7 to the legacy fixed-byte reader. That reader encounters the
+ negative pipeline magic where it expects `numChunks` and rejects or fails the segment load.
+ **DO NOT enable a V7-requiring `codecSpec` on any column until every server in your fleet has
+ been upgraded** to a build that includes V7 support. This includes transforms, compression
+ chains, and compression options such as `ZSTD(5)`.
+
+3. **Adding a new codec to `CodecRegistry.DEFAULT` is also a rolling-upgrade-sensitive change.** A server that does not know a given codec name cannot read segments encoded with it (lookup throws `IllegalArgumentException`). Operators must ensure every server in the fleet runs a build that registers the codec **before** enabling that codec on any column in table config.
+
+4. **Rollback is one-way at the codec level.** Once a column has been written with a V7 segment, downgrading the server fleet requires (a) reverting the `codecSpec` config to a legacy `compressionCodec`, (b) reloading the segments via the new servers to convert them back to a legacy raw format (this is supported — see §7.3), and (c) only then downgrading servers. Skipping step (b) leaves V7 segments on disk that downgraded servers cannot read.
+
+Future work (§12) covers automating these constraints via a controller-side gate.
+
+---
+
+## 6. Configuration
+
+### Table config
+
+`codecSpec` is configured under the modern `indexes.forward` block (it is **not** a top-level
+`FieldConfig` field — that pattern is reserved for legacy settings like `compressionCodec`). A
+top-level legacy `compressionCodec` and an `indexes.forward.codecSpec` are mutually exclusive:
+
+```jsonc
+{
+ "fieldConfigList": [
+ {
+ "name": "ts",
+ "encodingType": "RAW",
+ "indexes": { "forward": { "codecSpec": "DELTADELTA,LZ4" } }
+ },
+ {
+ "name": "userId",
+ "encodingType": "RAW",
+ "indexes": { "forward": { "codecSpec": "ZSTD(3)" } }
+ },
+ {
+ "name": "eventName",
+ "encodingType": "DICTIONARY"
+ }
+ ]
+}
+```
+
+### Validation
+
+`ForwardIndexType.validateCodecSpec` runs at table-config validation time (after `FieldIndexConfigsUtil` has resolved `noDictionaryColumns` / `noDictionaryConfig` overrides into the effective `ForwardIndexConfig`) and rejects:
+
+- Non-RAW encoding type
+- Spec parse failures (unknown codec, syntax error, unsigned-only argument violation, the removed `CODEC(...)` wrapper)
+- Structural pipeline errors (a transform after a packing transform or compression stage, or a second packing
+ transform); multiple byte-compression stages are allowed after all transforms
+- Specs that require the V7 codec-pipeline writer (any transform, multiple compression stages, or compression-only with non-default arguments like `ZSTD(5)`) on multi-value columns
+- Specs that require the V7 codec-pipeline writer on non-INT/LONG stored types
+
+Compression-only specs whose arguments map to a legacy `ChunkCompressionType` (`LZ4`, `SNAPPY`, `GZIP`, `ZSTD`/`ZSTD(3)`) use the existing raw forward-index writers and support any SV/MV + fixed/var-byte column.
+
+### Builder API (programmatic)
+
+`codecSpec` is set on the forward-index config, which is carried in `FieldConfig.indexes.forward`:
+
+```java
+// Build the indexes.forward JSON block with the codecSpec.
+ObjectNode forward = JsonUtils.newObjectNode();
+forward.put("codecSpec", "DELTADELTA,LZ4");
+ObjectNode indexes = JsonUtils.newObjectNode();
+indexes.set("forward", forward);
+
+FieldConfig fc = new FieldConfig.Builder("ts")
+ .withEncodingType(EncodingType.RAW)
+ .withIndexes(indexes)
+ .build();
+
+// Equivalently, configure ForwardIndexConfig directly when building FieldIndexConfigs in code:
+ForwardIndexConfig fwd = new ForwardIndexConfig.Builder(EncodingType.RAW)
+ .withCodecSpec("DELTADELTA,LZ4")
+ .build();
+```
+
+A top-level legacy `compressionCodec` cannot coexist with an `indexes.forward.codecSpec`; the
+table-config validator rejects that combination.
+
+### Choosing a codec
+
+| Workload | Recommended spec |
+|-------------------------------------------------------|----------------------------------|
+| Monotonic timestamps (epoch ms / s) | `DELTADELTA,LZ4` |
+| Time-ordered counters | `DELTA,LZ4` |
+| Approximately uniform random IDs | `LZ4` or `ZSTD(3)` |
+| Cold fixed-byte INT/LONG data with trend | `DELTA,ZSTD(8)` |
+| Compatibility with old SNAPPY/GZIP segments | `SNAPPY` or `GZIP` |
+| Fastest decode, low compression ratio | `LZ4` |
+
+---
+
+## 7. User manual
+
+### 7.1 Enabling `codecSpec` on a new column
+
+1. Before enabling **any** `codecSpec`, upgrade every controller and tenant server, plus any minion or external segment builder that validates or consumes the table config, to a Pinot version that understands the field (≥ 1.6). Older components can reject the unknown JSON property even when the selected codec uses legacy segment bytes.
+2. For a V7-requiring spec (a transform, multiple compression stages, or non-default options such as `ZSTD(5)`), also verify every server that can load the table's segments has a V7 reader.
+3. Add `codecSpec` under the column's `indexes.forward` block. Do **not** also set a top-level `compressionCodec`; the table config validator will reject mutual presence.
+4. Rebuild affected segments (push offline data, wait for realtime → committed transitions, or trigger a segment reload — see §7.3).
+5. The single-stage legacy-compatible specs `LZ4`, `SNAPPY`, `GZIP`, and `ZSTD`/`ZSTD(3)` use existing raw forward-index formats. Every other valid spec uses V7 and embeds the canonical spec.
+
+### 7.2 Migrating from the legacy `compressionCodec`
+
+Edit the table config, replacing `compressionCodec` with the equivalent `codecSpec`:
+
+| Legacy `compressionCodec` | Equivalent `codecSpec` | Semantic note |
+|---------------------------|--------------------------------|---------------------------------------------------------------|
+| `LZ4` | `LZ4` | Drop-in equivalent |
+| `ZSTANDARD` | `ZSTD(3)` | Drop-in equivalent |
+| `SNAPPY` | `SNAPPY` | Drop-in equivalent |
+| `GZIP` | `GZIP` | Drop-in equivalent |
+| `DELTA` | `DELTA,LZ4` | **Adds LZ4 byte compression**; not a byte-for-byte equivalent |
+| `DELTADELTA` | `DELTADELTA,LZ4` | **Adds LZ4 byte compression**; not a byte-for-byte equivalent |
+| `PASS_THROUGH` | (no migration; keep legacy) | No codec to apply |
+| `MV_ENTRY_DICT`, CLP family | (no migration; not in scope) | |
+
+When migrating `DELTA`/`DELTADELTA`, note that the new spec adds LZ4 byte compression on top of the
+delta transform, so existing segments will be rewritten to a different byte layout on the next
+reload. Future migration tooling may automate this table-config rewrite (see §12); for now the edit
+is manual.
+
+### 7.3 Rolling back from `codecSpec`
+
+If a table needs to be downgraded to a Pinot version that does **not** understand `codecSpec`, first replace affected configs with legacy `compressionCodec` values. Tables with V7-requiring specs also need their V7 segments rewritten before downgrade:
+
+1. In table config, replace `codecSpec` with an equivalent legacy `compressionCodec` (e.g. `codecSpec="LZ4"` → `compressionCodec=LZ4`).
+2. For V7-requiring specs, trigger a segment reload. `ForwardIndexHandler` detects the legacy revert (its internal `shouldRewriteRawForwardIndex` check) and rewrites V7 segments to the legacy format.
+3. Once all segments are rewritten, the cluster can be downgraded.
+
+`ForwardIndexHandler` recognizes the legitimate legacy revert targets for fixed-byte SV columns:
+`PASS_THROUGH`, `SNAPPY`, `ZSTANDARD`, `LZ4`, `GZIP`, `DELTA`, `DELTADELTA` (CLP family is excluded — not applicable to fixed-byte SV).
+
+Segment reload also covers the forward direction and spec-to-spec changes: enabling a `codecSpec` on
+an existing column, or changing one `codecSpec` to another, rewrites the forward index on reload
+without requiring a data re-push.
+
+### 7.4 Mixed-version cluster safety
+
+There are two independent compatibility boundaries:
+
+- **Table config:** `codecSpec` is a new JSON property. Before enabling any spec, upgrade every controller and tenant server, plus segment-building minions or clients that validate or consume the config, to a version that understands it. This applies even to `LZ4` or `ZSTD(3)`, whose segment bytes remain legacy-compatible.
+- **Segment bytes:** V7 pipeline segments are **forward-only** and can only be read by Pinot ≥ 1.6. Before enabling a V7-requiring spec, verify with `kubectl get pods` (or equivalent) that every server able to load the table's segments is running a V7-capable binary.
+
+If a V7-requiring `codecSpec` is enabled before all servers are upgraded, older servers dispatch
+version 7 to the legacy fixed-byte reader, encounter the negative pipeline magic where they expect
+`numChunks`, and fail the segment load while parsing the incompatible header.
+
+### 7.5 Common errors and remediation
+
+| Error message | Remediation |
+|--------------------------------------------------------------------------------------------|--------------------------------------------------------------|
+| `Conflicting forward-index config for column: <col> — FieldConfig.compressionCodec=... but indexes.forward.codecSpec is also set` | Remove the legacy top-level `compressionCodec`; keep only `indexes.forward.codecSpec` |
+| `CODEC(...) wrapper is not supported; list codec invocations directly in: ...` | Drop the removed `CODEC(...)` wrapper; list the stages comma-separated (`DELTA,LZ4`) |
+| `codecSpec '...' requires the V7 codec-pipeline writer ... only supports single-value columns. Column 'X' is multi-value` | Use one legacy-compatible compression invocation or legacy `compressionCodec` |
+| `codecSpec '...' requires the V7 codec-pipeline writer ... only supports INT and LONG columns. Column 'X' has type: ...` | Use one legacy-compatible compression invocation or legacy `compressionCodec` |
+| `Unknown codec 'XYZ'. Known codecs: [DELTA, DELTADELTA, T64, GORILLA, ZSTD, ZSTANDARD, LZ4, SNAPPY, GZIP]` | Fix typo in DSL; `ZSTANDARD` is an alias for `ZSTD` |
+| `Transform stage '<name>' must operate on column values ...` | Put typed-layout-preserving transforms first, an optional T64/GORILLA packing transform next, and compression stages last |
+| `Leading sign is not allowed in codec argument at position N in: ...` | Use unsigned integer (e.g. `ZSTD(3)`, not `ZSTD(+3)`) |
+| `LZ4: decompressed length N in length prefix is out of range [0, 1073741824]. Segment may be corrupt.` | Segment file is corrupt; re-download from deep storage |
+
+---
+
+## 8. Threading and concurrency
+
+- **`CodecRegistry.DEFAULT`** is built in a `static {}` block and wrapped in `Collections.unmodifiableMap`. Safe for concurrent reads. The mutable `CodecRegistry()` constructor is `@VisibleForTesting`.
+- **`CodecPipelineExecutor`** is immutable after construction. `encode(src)` and `decode(src, dst)` are thread-safe.
+- **`FixedByteChunkSVForwardIndexReaderV7`** is immutable after construction and may be shared across threads. Each `ChunkReaderContext` is single-threaded — the returned chunk buffer is the context's reusable scratch and must not be retained across `getInt`/`getLong` calls.
+- **`FixedByteChunkForwardIndexWriterV7`** is `@NotThreadSafe`.
+- **`GzipCodecDefinition`** reuses `ThreadLocal<Deflater>` and `ThreadLocal<Inflater>` instances, resets them after every operation so they do not retain caller buffers, and reads from and writes to direct `ByteBuffer` instances without heap staging arrays.
+- **`Lz4CodecDefinition`** wraps `LZ4Factory.fastestInstance()` in a private inner holder class so a missing native library only fails when LZ4 is actually used (not at registry class init).
+
+---
+
+## 9. Performance characteristics
+
+- **Read hot path**: `getInt(int)`/`getLong(int)` use absolute `ByteBuffer.getXxx(int)` indexing — no per-row position mutation, no allocation. The reader returns the context's reusable scratch buffer directly (no per-chunk `duplicate()`).
+- **Chunk-cache**: `ChunkReaderContext` caches the last-decoded chunk; sequential reads stay in cache and pay the decode cost only on chunk transitions. `setChunkId(-1)` is set **before** decompress so a thrown decoder leaves the cache invalidated rather than appearing valid with partial data.
+- **Single-stage pipelines** decompress directly into the context buffer (no intermediate allocation).
+ Multi-stage decode sizes each scratch buffer from the validated outer decoded size and codec bounds;
+ every reverse stage uses bounded `decodeInto`. Intermediate direct buffers are deterministically
+ cleaned after encode/decode rather than retained until garbage collection.
+- **Resource bounds**: writer and reader both preflight the full-chunk composed size bound. A
+ pipeline/chunk-size combination is rejected before file creation or context allocation if any
+ stage can exceed the 128 MiB encoded/intermediate ceiling or the pipeline can exceed the
+ 256 MiB cumulative-work ceiling.
+- **GZIP** uses the JDK 11+ direct `ByteBuffer` APIs for both deflate and inflate, avoiding whole-chunk heap staging and copy-back. Thread-local `Deflater`/`Inflater` instances amortize native setup cost.
+- **Header parse cost**: O(numChunks) once per reader open (monotonicity scan) — bounded and amortized over the segment lifetime.
+
+---
+
+## 10. Backward compatibility
+
+| Concern | Status |
+|--------------------------------------------------------------------|---------------------------------------------------------------------------------------|
+| Existing tables with `compressionCodec` keep working | ✅ Legacy path unchanged |
+| `FieldConfig` constructor signature for plugins | ✅ Unchanged; `codecSpec` lives in the nested `indexes.forward` config |
+| Existing `ForwardIndexConfig` builder and JSON construction paths | ✅ Preserved; `codecSpec` is an additive builder/JSON property |
+| `ForwardIndexReader.getCodecSpec()` SPI addition | ✅ `default` method returning `null`; existing implementations don't break |
+| Arbitrary version tags emitted by the existing legacy fixed-byte writer (including 5, 6, 7, and 10) | ✅ Structurally distinguished from pipeline V7 and dispatched to the legacy reader |
+| Other old V1–V6 segments readable on new servers | ✅ Existing reader dispatch remains intact |
+| Any `codecSpec` consumed by pre-1.6 components | ❌ New JSON property — upgrade config consumers before enabling it |
+| V7 pipeline segments readable on old servers | ❌ Forward-only — upgrade the fleet before enabling any V7-requiring spec |
+| `CompressionCodec` enum unchanged | ✅ No values removed or renamed; legacy field and getters remain supported |
+| Mutual exclusion of `compressionCodec` and `codecSpec` | ✅ Old JSON remains valid; a column may set only one of the two paths |
+
+---
+
+## 11. Test coverage
+
+| Test class | What it locks in |
+|---------------------------------------------------------|---------------------------------------------------------------------|
+| `CodecSpecParserTest` | DSL grammar, ASCII identifiers and digits, signed-argument rejection, removed-wrapper rejection, and resource limits |
+| `CodecInvocationTest`, `CodecPipelineTest` | Immutable AST construction, normalization, and direct-construction limits |
+| `CodecPipelineValidatorTest` | Typed-layout-transform chaining, optional packing transform, compression chaining, invalid ordering and type checks |
+| `CodecPipelineExecutorTest`, `CodecRegistryTest`, `CodecSpecUtilsTest` | Pipeline binding, canonical spec emission, registry lookup/aliasing, and legacy-compatibility classification |
+| `DeltaCodecRoundTripTest` | DELTA/DELTADELTA round-trips including two's-complement wrap semantics |
+| `CodecPipelineForwardIndexTest` | Write/read round-trips for all 8 built-ins and representative chains across INT/LONG; boundaries; partial last chunks; V7 dispatch |
+| `T64CodecDefinitionTest`, `GorillaCodecDefinitionTest` | Codec-specific round-trips, boundary values, and corrupt-input handling |
+| `ZstdCodecDefinitionTest` | Empty-input round-trip edge case |
+| `CompressionCodecCorruptInputTest` | Bounded multi-stage decompression, complete-frame validation, and corrupt/truncated input rejection |
+| `FixedByteChunkSVForwardIndexReaderV7CorruptionTest` | Magic-only dispatch, truncated/oversized headers, overflow-safe offsets, per-chunk extents, exact sizes, and resource caps |
+| `ForwardIndexReaderFactoryBackwardCompatTest` | Legacy fixed-byte writer versions 4, 5, 6, 7, and 10 across fixed types and codecs remain readable |
+| `ForwardIndexConfigTest` | JSON round-trip, wrapper-less canonicalization, builder copy, and mutual exclusion |
+| `ForwardIndexCreatorFactoryTest`, `ForwardIndexTypeTest` | Writer selection (legacy vs V7) and table-config-time codecSpec validation |
+| `ForwardIndexHandlerTest` | Legacy↔V7 reload transitions, codecSpec-to-codecSpec changes, and the legacy revert path |
+| `TableConfigUtilsTest` | Table-config validation and effective forward-index config resolution |
+| `CodecPipelineIntegrationTest` | All configured codec specs in both query engines (SSE+MSE), including point, aggregate, cross-codec, and dictionary-coexistence assertions |
+
+---
+
+## 12. Future work
+
+- **Plugin codec registration**: expose `CodecRegistry.setDefault(...)` or service-loader so external modules can register codecs without forking.
+- **Variable-width / MV V7 support**: extend the V7 writer/reader to STRING, BYTES, and multi-value columns. Single-stage specs that map to a legacy compression codec already route to existing raw forward-index writers, so compression-only `codecSpec` works for those column shapes today.
+- **Transform chaining**: `DELTA`/`DELTADELTA` are header-less typed-layout-preserving transforms and chain freely; `T64`/`GORILLA` are packing transforms (bit-packed output) and so must be the last transform. A follow-up could give the packing transforms a typed-passthrough output form so chains like `T64,GORILLA` become expressible — though the practical value is limited.
+- **JMH benchmark**: add `pinot-perf` benchmarks comparing V7 (LZ4 / ZSTD / DELTA+LZ4) against legacy `FixedBytePower2ChunkSVForwardIndexReader` for read throughput and segment build cost.
+- **Controller-side capability gate**: refuse `codecSpec` table-config updates until every relevant config consumer and segment reader/builder advertises support (mixed-version safety).
+- **Migration tooling and sunset of legacy `compressionCodec`**: add tooling that rewrites legacy `compressionCodec` configs to the equivalent `codecSpec` (per the §7.2 table), then — once `codecSpec` covers all column types — deprecate `getCompressionCodec()` and the enum constants in 2.0.
+
+---
+
+## 13. References
+
+- Pull request: [#18229](https://github.com/apache/pinot/pull/18229) (split into a stacked PR series)
+- Source packages: `pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/codec/`, `pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/`
+- Reader: `pinot-segment-local/.../FixedByteChunkSVForwardIndexReaderV7.java`
+- Writer: `pinot-segment-local/.../FixedByteChunkForwardIndexWriterV7.java`
+- Handler: `pinot-segment-local/.../ForwardIndexHandler.java` (reload-time rewrite decisions)
+- Integration test: `pinot-integration-tests/.../custom/CodecPipelineIntegrationTest.java`
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CodecPipelineIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CodecPipelineIntegrationTest.java
new file mode 100644
index 00000000000..b952812b195
--- /dev/null
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CodecPipelineIntegrationTest.java
@@ -0,0 +1,426 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.integration.tests.custom;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+import org.apache.avro.Schema.Field;
+import org.apache.avro.Schema.Type;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// Integration test for forward-index `codecSpec` handling.
+///
+/// Builds an offline table where every supported transform codec spec is applied to its own INT
+/// and LONG raw column, and a compression-only `codecSpec` is applied to a STRING raw column. All
+/// INT/LONG columns are populated with identical values (intVal = i, longVal = i * 1_000_000_000L),
+/// so every codec must read back the same aggregates, filter counts, and point lookups. A STRING
+/// dictionary column verifies that codec-spec raw columns and dictionary-encoded columns coexist in
+/// the same segment.
+///
+/// Codec specs covered (each as its own column):
+///
+/// - `LZ4`, `ZSTD(3)`, `SNAPPY`, `GZIP` — single-stage compression
+/// - `DELTA,LZ4`, `DELTA,ZSTD(3)` — DELTA transform + compression
+/// - `DELTADELTA,LZ4` — second-order DELTA transform + compression
+/// - `T64`, `GORILLA` — packing transforms without compression
+/// - `T64,LZ4`, `GORILLA,ZSTD(3)` — packing transform + compression
+@Test(suiteName = "CustomClusterIntegrationTest")
+public class CodecPipelineIntegrationTest extends CustomDataQueryClusterIntegrationTest {
+
+ private static final String TABLE_NAME = "CodecPipelineIntegrationTest";
+ private static final int NUM_DOCS = 1000;
+ private static final int V7_TARGET_DOCS_PER_CHUNK = 256;
+
+ private static final String STR_COL = "strVal";
+ // STRING column with RAW encoding + compression-only codecSpec. This uses the existing raw
+ // forward-index writer path, not the V7 transform writer.
+ private static final String STR_RAW_COL = "strRawZstd";
+ private static final String TIME_COL = "ts";
+
+ // Expected aggregates: SUM(0..999) = 499_500
+ private static final long EXPECTED_INT_SUM = 499_500L;
+ private static final long EXPECTED_LONG_SUM = 499_500L * 1_000_000_000L;
+
+ /// Codec spec → column-name suffix. Each codec spec gets its own INT and LONG column
+ /// (`int<suffix>` / `long<suffix>`). Order matters only for diagnostic output.
+ /// LinkedHashMap preserves declaration order so the data provider is stable.
+ private static final Map<String, String> CODEC_SPECS;
+ static {
+ Map<String, String> m = new LinkedHashMap<>();
+ m.put("LZ4", "Lz4");
+ m.put("ZSTD(3)", "Zstd");
+ m.put("SNAPPY", "Snappy");
+ m.put("GZIP", "Gzip");
+ m.put("DELTA,LZ4", "DeltaLz4");
+ m.put("DELTA,ZSTD(3)", "DeltaZstd");
+ m.put("DELTADELTA,LZ4", "DeltadeltaLz4");
+ m.put("T64", "T64");
+ m.put("GORILLA", "Gorilla");
+ m.put("T64,LZ4", "T64Lz4");
+ m.put("GORILLA,ZSTD(3)", "GorillaZstd");
+ CODEC_SPECS = m;
+ }
+
+ private static String intColFor(String suffix) {
+ return "int" + suffix;
+ }
+
+ private static String longColFor(String suffix) {
+ return "long" + suffix;
+ }
+
+ @Override
+ public String getTableName() {
+ return TABLE_NAME;
+ }
+
+ @Override
+ public Schema createSchema() {
+ Schema.SchemaBuilder builder = new Schema.SchemaBuilder().setSchemaName(getTableName());
+ for (String suffix : CODEC_SPECS.values()) {
+ builder.addMetric(intColFor(suffix), FieldSpec.DataType.INT);
+ builder.addMetric(longColFor(suffix), FieldSpec.DataType.LONG);
+ }
+ builder.addSingleValueDimension(STR_COL, FieldSpec.DataType.STRING);
+ builder.addSingleValueDimension(STR_RAW_COL, FieldSpec.DataType.STRING);
+ builder.addDateTimeField(TIME_COL, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS");
+ return builder.build();
+ }
+
+ @Override
+ public List<File> createAvroFiles()
+ throws IOException {
+ org.apache.avro.Schema avroSchema = org.apache.avro.Schema.createRecord("codecRecord", null, null, false);
+ List<Field> fields = new ArrayList<>();
+ for (String suffix : CODEC_SPECS.values()) {
+ fields.add(new Field(intColFor(suffix), org.apache.avro.Schema.create(Type.INT), null, null));
+ fields.add(new Field(longColFor(suffix), org.apache.avro.Schema.create(Type.LONG), null, null));
+ }
+ fields.add(new Field(STR_COL, org.apache.avro.Schema.create(Type.STRING), null, null));
+ fields.add(new Field(STR_RAW_COL, org.apache.avro.Schema.create(Type.STRING), null, null));
+ fields.add(new Field(TIME_COL, org.apache.avro.Schema.create(Type.LONG), null, null));
+ avroSchema.setFields(fields);
+
+ try (AvroFilesAndWriters avroFilesAndWriters = createAvroFilesAndWriters(avroSchema)) {
+ List<DataFileWriter<GenericData.Record>> writers = avroFilesAndWriters.getWriters();
+ for (int i = 0; i < NUM_DOCS; i++) {
+ GenericData.Record record = new GenericData.Record(avroSchema);
+ for (String suffix : CODEC_SPECS.values()) {
+ record.put(intColFor(suffix), i);
+ record.put(longColFor(suffix), (long) i * 1_000_000_000L);
+ }
+ record.put(STR_COL, "str_" + i);
+ record.put(STR_RAW_COL, "rawstr_" + i);
+ record.put(TIME_COL, (long) i);
+ writers.get(i % getNumAvroFiles()).append(record);
+ }
+ return avroFilesAndWriters.getAvroFiles();
+ }
+ }
+
+ @Override
+ public String getTimeColumnName() {
+ return TIME_COL;
+ }
+
+ @Override
+ protected long getCountStarResult() {
+ return NUM_DOCS;
+ }
+
+ @Override
+ public TableConfig createOfflineTableConfig() {
+ return new TableConfigBuilder(TableType.OFFLINE).setTableName(getTableName())
+ .setNoDictionaryColumns(getNoDictionaryColumns())
+ .setFieldConfigList(getFieldConfigs())
+ .build();
+ }
+
+ @Override
+ protected List<String> getNoDictionaryColumns() {
+ // STR_COL uses a dictionary (default), so it is intentionally NOT in this list.
+ // STR_RAW_COL uses RAW encoding with a compression-only codecSpec.
+ List<String> noDict = new ArrayList<>();
+ for (String suffix : CODEC_SPECS.values()) {
+ noDict.add(intColFor(suffix));
+ noDict.add(longColFor(suffix));
+ }
+ noDict.add(STR_RAW_COL);
+ return noDict;
+ }
+
+ @Override
+ protected List<FieldConfig> getFieldConfigs() {
+ List<FieldConfig> fieldConfigs = new ArrayList<>();
+ for (Map.Entry<String, String> entry : CODEC_SPECS.entrySet()) {
+ String spec = entry.getKey();
+ String suffix = entry.getValue();
+ fieldConfigs.add(rawFieldConfigWithCodecSpec(intColFor(suffix), spec));
+ fieldConfigs.add(rawFieldConfigWithCodecSpec(longColFor(suffix), spec));
+ }
+ // STR_COL with dictionary encoding — verifies codec-pipeline and dict columns coexist.
+ fieldConfigs.add(new FieldConfig.Builder(STR_COL)
+ .withEncodingType(FieldConfig.EncodingType.DICTIONARY)
+ .build());
+ // STR_RAW_COL with RAW + compression-only codecSpec verifies the existing raw compression path
+ // is selected for STRING columns while transform codec specs still use V7 for INT/LONG.
+ fieldConfigs.add(rawFieldConfigWithCodecSpec(STR_RAW_COL, "ZSTD(3)"));
+ return fieldConfigs;
+ }
+
+ /// Builds a RAW FieldConfig whose codecSpec is configured via the modern `indexes.forward` block
+ /// (the only supported path; there is no top-level FieldConfig.codecSpec field).
+ private static FieldConfig rawFieldConfigWithCodecSpec(String column, String codecSpec) {
+ ObjectNode forward = JsonUtils.newObjectNode();
+ forward.put("codecSpec", codecSpec);
+ // Each input file contains 500 rows. Keep V7 chunks smaller than that so point lookups
+ // exercise both sides of a real chunk boundary within each generated segment.
+ forward.put("targetDocsPerChunk", V7_TARGET_DOCS_PER_CHUNK);
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set("forward", forward);
+ return new FieldConfig.Builder(column)
+ .withEncodingType(FieldConfig.EncodingType.RAW)
+ .withIndexes(indexes)
+ .build();
+ }
+
+ @Nullable
+ @Override
+ protected String getSortedColumn() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ protected List<String> getInvertedIndexColumns() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ protected List<String> getRangeIndexColumns() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ protected List<String> getBloomFilterColumns() {
+ return null;
+ }
+
+ /// Cartesian product of (codec spec, query engine) so every codec is exercised on both engines.
+ @DataProvider(name = "codecAndEngine")
+ public Object[][] codecAndEngine() {
+ List<Object[]> rows = new ArrayList<>(CODEC_SPECS.size() * 2);
+ for (Map.Entry<String, String> entry : CODEC_SPECS.entrySet()) {
+ String spec = entry.getKey();
+ String suffix = entry.getValue();
+ rows.add(new Object[]{spec, suffix, false});
+ rows.add(new Object[]{spec, suffix, true});
+ }
+ return rows.toArray(new Object[0][]);
+ }
+
+ @Test(dataProvider = "codecAndEngine")
+ public void testSumPerCodec(String codecSpec, String suffix, boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String intCol = intColFor(suffix);
+ String longCol = longColFor(suffix);
+
+ JsonNode intSum = postQuery("SELECT SUM(" + intCol + ") FROM " + getTableName());
+ assertEquals(intSum.get("resultTable").get("rows").get(0).get(0).asLong(), EXPECTED_INT_SUM,
+ "Unexpected SUM(" + intCol + ") for codec " + codecSpec);
+
+ JsonNode longSum = postQuery("SELECT SUM(" + longCol + ") FROM " + getTableName());
+ assertEquals(longSum.get("resultTable").get("rows").get(0).get(0).asLong(), EXPECTED_LONG_SUM,
+ "Unexpected SUM(" + longCol + ") for codec " + codecSpec);
+ }
+
+ @Test(dataProvider = "codecAndEngine")
+ public void testFilterPerCodec(String codecSpec, String suffix, boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String intCol = intColFor(suffix);
+ String longCol = longColFor(suffix);
+
+ // intVal < 100 → 100 rows (values 0..99)
+ JsonNode intFilter = postQuery("SELECT COUNT(*) FROM " + getTableName() + " WHERE " + intCol + " < 100");
+ assertEquals(intFilter.get("resultTable").get("rows").get(0).get(0).asLong(), 100L,
+ "Unexpected count for " + intCol + " < 100, codec " + codecSpec);
+
+ // longVal < 100_000_000_000L → 100 rows
+ JsonNode longFilter =
+ postQuery("SELECT COUNT(*) FROM " + getTableName() + " WHERE " + longCol + " < 100000000000");
+ assertEquals(longFilter.get("resultTable").get("rows").get(0).get(0).asLong(), 100L,
+ "Unexpected count for " + longCol + " < 100B, codec " + codecSpec);
+ }
+
+ /// Per-codec point lookups across multiple chunk boundaries. Aggregate queries can mask per-doc
+ /// decoding errors that average out — point lookups force the reader to materialize specific
+ /// values, including chunk-boundary docs.
+ @Test(dataProvider = "codecAndEngine")
+ public void testPointLookupsPerCodec(String codecSpec, String suffix, boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String intCol = intColFor(suffix);
+ String longCol = longColFor(suffix);
+
+ // Avro records are distributed round-robin across two files. IDs 510/512 are local rows
+ // 255/256 in one segment and 511/513 are rows 255/256 in the other segment.
+ int[] spotCheckIds = {0, 1, 510, 511, 512, 513, 999};
+ for (int id : spotCheckIds) {
+ JsonNode intResult =
+ postQuery("SELECT " + intCol + " FROM " + getTableName() + " WHERE " + TIME_COL + " = " + id);
+ assertEquals(intResult.get("resultTable").get("rows").get(0).get(0).asInt(), id,
+ "Wrong " + intCol + " for ts=" + id + ", codec " + codecSpec);
+
+ JsonNode longResult =
+ postQuery("SELECT " + longCol + " FROM " + getTableName() + " WHERE " + TIME_COL + " = " + id);
+ assertEquals(longResult.get("resultTable").get("rows").get(0).get(0).asLong(), (long) id * 1_000_000_000L,
+ "Wrong " + longCol + " for ts=" + id + ", codec " + codecSpec);
+ }
+ }
+
+ /// Verifies that a single SELECT touching multiple codec-encoded columns returns consistent values
+ /// across codecs in the same row — catches any chunk-state cross-talk between readers.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testCrossCodecConsistency(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ List<String> suffixes = new ArrayList<>(CODEC_SPECS.values());
+ String selectList = String.join(", ", Stream.concat(
+ suffixes.stream().map(CodecPipelineIntegrationTest::intColFor),
+ suffixes.stream().map(CodecPipelineIntegrationTest::longColFor))
+ .toArray(String[]::new));
+
+ int[] spotCheckIds = {0, 510, 511, 512, 513, 999};
+ for (int id : spotCheckIds) {
+ JsonNode result = postQuery("SELECT " + selectList + " FROM " + getTableName() + " WHERE ts = " + id);
+ JsonNode row = result.get("resultTable").get("rows").get(0);
+ // First N columns are int; next N are long.
+ for (int i = 0; i < suffixes.size(); i++) {
+ assertEquals(row.get(i).asInt(), id,
+ "Cross-codec int mismatch at suffix " + suffixes.get(i) + " for ts=" + id);
+ }
+ for (int i = 0; i < suffixes.size(); i++) {
+ assertEquals(row.get(suffixes.size() + i).asLong(), (long) id * 1_000_000_000L,
+ "Cross-codec long mismatch at suffix " + suffixes.get(i) + " for ts=" + id);
+ }
+ }
+ }
+
+ /// Verifies that a STRING column stored with dictionary encoding (not codec pipeline) reads back
+ /// correctly alongside codec-pipeline columns, confirming both can coexist in the same segment.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testStringColumnWithDictEncoding(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+
+ int[] spotCheckIds = {0, 42, 500, 999};
+ for (int id : spotCheckIds) {
+ JsonNode result = postQuery("SELECT strVal FROM " + getTableName() + " WHERE ts = " + id);
+ assertEquals(result.get("resultTable").get("rows").get(0).get(0).asText(), "str_" + id,
+ "Wrong strVal for ts=" + id);
+ }
+
+ JsonNode countDistinctResult = postQuery("SELECT COUNT(DISTINCT strVal) FROM " + getTableName());
+ assertEquals(countDistinctResult.get("resultTable").get("rows").get(0).get(0).asLong(), NUM_DOCS,
+ "Expected all " + NUM_DOCS + " distinct string values");
+ }
+
+ /// Verifies that a STRING column stored RAW with `codecSpec=ZSTD(3)` reads back correctly
+ /// alongside the transform codec-spec INT/LONG columns. This exercises the existing chunk
+ /// forward-index compression path within a segment that also contains V7 transform columns.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testStringColumnWithRawCodecSpecCompression(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+
+ int[] spotCheckIds = {0, 42, 500, 999};
+ for (int id : spotCheckIds) {
+ JsonNode result =
+ postQuery("SELECT " + STR_RAW_COL + " FROM " + getTableName() + " WHERE ts = " + id);
+ assertEquals(result.get("resultTable").get("rows").get(0).get(0).asText(), "rawstr_" + id,
+ "Wrong " + STR_RAW_COL + " for ts=" + id);
+ }
+
+ JsonNode countDistinctResult =
+ postQuery("SELECT COUNT(DISTINCT " + STR_RAW_COL + ") FROM " + getTableName());
+ assertEquals(countDistinctResult.get("resultTable").get("rows").get(0).get(0).asLong(), NUM_DOCS,
+ "Expected all " + NUM_DOCS + " distinct " + STR_RAW_COL + " values");
+ }
+
+ /// Verify COUNT(*) reads through every codec column path consistently via a join-style test.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testCountAcrossCodecs(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ // COUNT with no WHERE: should be NUM_DOCS regardless of column choice.
+ JsonNode count = postQuery("SELECT COUNT(*) FROM " + getTableName());
+ assertEquals(count.get("resultTable").get("rows").get(0).get(0).asLong(), NUM_DOCS,
+ "Unexpected total row count");
+
+ // COUNT WHERE intLz4 = intZstd (every row should match: same values across codecs).
+ JsonNode crossCount = postQuery(
+ "SELECT COUNT(*) FROM " + getTableName() + " WHERE intLz4 = intZstd AND longSnappy = longGzip");
+ assertEquals(crossCount.get("resultTable").get("rows").get(0).get(0).asLong(), NUM_DOCS,
+ "Cross-codec equality must hold for every row");
+ }
+
+ /// Sanity check: the codec spec list stays in sync with the rest of the test setup.
+ @Test
+ public void testAllCodecSpecsRegisteredInTableConfig() {
+ List<String> expectedColumns = new ArrayList<>();
+ for (String suffix : CODEC_SPECS.values()) {
+ expectedColumns.add(intColFor(suffix));
+ expectedColumns.add(longColFor(suffix));
+ }
+ expectedColumns.add(STR_RAW_COL);
+ List<String> noDict = getNoDictionaryColumns();
+ assertEquals(noDict.size(), expectedColumns.size(),
+ "noDictionaryColumns size must match the codec-spec matrix plus raw codecSpec columns");
+ for (String col : expectedColumns) {
+ if (!noDict.contains(col)) {
+ throw new AssertionError("Expected " + col + " in noDictionaryColumns; got " + noDict);
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org