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:49 UTC
(pinot) branch xiangfu0/codex/codec-stack/07-integration-docs created (now 88e1d409026)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a change to branch xiangfu0/codex/codec-stack/07-integration-docs
in repository https://gitbox.apache.org/repos/asf/pinot.git
at 88e1d409026 Add codec pipeline integration tests and design doc
This branch includes the following new commits:
new 88e1d409026 Add codec pipeline integration tests and design doc
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/01: Add codec pipeline integration tests and design doc
Posted by xi...@apache.org.
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