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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
