perf(parquet): forward dictionary encoding through append compaction rewrite

[SteNicholas]

Purpose

Linked issue: close #230

An append-only compaction rewrite copies rows into the new file without inspecting any value, so expanding a dictionary-encoded Parquet column on read and hashing it again on write is work neither side needs. This forwards the encoding instead.

Reader. New option parquet.read.enable-dictionary-passthrough, off by default, makes ParquetFileBatchReader request set_read_dictionary for non-nested STRING/BINARY columns whose every data page, in every row group of the file, is dictionary-encoded. A dictionary page alone cannot be the signal — a column that outgrew its page limit still carries the page it had already emitted — so the gate reads the encoding statistics. The file schema keeps reporting the logical type; only the emitted batches carry the dictionary.

Writer. ParquetFormatWriter recovers each batch's encoding from its layout, because exporting through the Arrow C data interface drops the type. A layout pins down neither the index nor the offset width, so only dictionary(int32, utf8|binary) — exactly what Arrow's Parquet reader emits — is recoverable. CompactRewrite therefore decodes anything else per column, while the type is still known: the ORC reader's dictionary(int64, large_utf8) under lazy decoding, dictionaries below the top level, non-binary value types. The rest stay encoded, so one awkward column does not cost the others their encoding. A dictionary holding nulls in its values is flattened at the writer — the one shape parquet::arrow rejects outright.

Gating. Compaction opts in only when the output is Parquet, parquet.enable-dictionary is on, and no shredding plan is active; anything else forces the read option off regardless of the table setting. A file index configured on a forwarded column materializes that column alone.

Known trade-off. A Parquet column chunk carries one dictionary, so when the input files supply different ones the output keeps the first and falls back to plain for the rest of the row group. The rewritten data is unchanged, but the output file may be larger than one written from materialized values. This is documented in compaction.rst and pinned by TestWriteDictionaryChangingAcrossBatches. Whether it is worth mitigating (for example by starting a new row group at each dictionary boundary) should be decided from the benchmark numbers below.

Design follows Velox's perf(parquet): Dictionary passthrough and selective flattening in Parquet writer (facebookincubator/velox#17986): per-column selective flatten, only VARCHAR/VARBINARY passed through, dictionaries with null values flattened, import schema reconciled per batch. It differs in where the flatten happens — Velox still holds a typed Vector inside its writer, whereas FormatWriter::AddBatch(ArrowArray*) here receives an untyped array, so the flatten has to run one layer up, before the type is dropped.

Tests

Unit:

• ArrowUtilsTest.TestResolveParquetDictionaryStructType — layout-derived resolution; rejects non-STRING/BINARY value types, large_utf8, and dictionaries below the top level; preserves a caller-declared dictionary type.
• ArrowUtilsTest.TestFlattenUnresolvableDictionaries — selective flatten: dictionary(int64, large_utf8) and dictionary(int64, utf8) decoded while the int32 neighbour stays encoded; nested dictionary decoded; unchanged batch returned by identity; sliced batch keeps its offset.
• ReaderUtilsTest.TestApplyBitmapToReadBatchKeepsDictionaryEncoding — a deletion vector filters batches by slice + arrow::Concatenate; pins that the encoding survives it.
• DataFileIndexWriterTest.TestDictionaryEncodedIndexedColumnRoundTrip — bitmap index built from a forwarded column.
• ParquetFileBatchReaderTest.TestDictionaryPassthrough — on / explicitly off / option absent / file without dictionary pages.
• ParquetFileBatchReaderTest.TestDictionaryPassthroughSkipsFallbackToPlain — a column that falls back to plain inside one chunk is declined.
• ParquetFileBatchReaderTest.TestDictionaryPassthroughRequiresEveryRowGroup — first row group fully dictionary-encoded, second falls back; the whole column is declined.
• ParquetFormatWriterTest.TestWriteDictionaryEncodedColumn / ChangingAcrossBatches / WithNullsInDictionary / WithNullRows / WithDuplicateValues / EmptyBatch / OfUnsupportedTypeIsRejected.
• ParquetFormatWriterTest.TestGetEstimateLengthWithDictionaryBatches — GetEstimateLength() and ReachTargetSize() still drive file rolling when batches arrive encoded.

Integration:

• AppendCompactionInteTest.TestAppendTableCompactionDictionaryPassthrough (Parquet + ORC) — asserts the input read types (id INT32, s/b dictionary(int32, utf8) on Parquet, s dictionary(int64, large_utf8) on ORC), then the full rewrite, then a predicate read through the bitmap index.
• AppendCompactionInteTest.TestAppendTableCompactionDictionaryPassthroughDisabled — the kill switch produces the same table.

Benchmarks (benchmark/parquet_format_benchmark.cpp):

• BM_ParquetWrite_DictionaryStringIntoStringSchema — the shape a rewrite produces (plain STRING write schema, dictionary-encoded batch), on the same 10 / 1000 / kRowsPerFile axis as its BM_ParquetWrite_String baseline.
• BM_ParquetRead_DictionaryPassthrough — the same axis with the option on and off at each cardinality; the high-cardinality point is where the gate declines and the two runs should measure the same work.

API and Format

No public API under include/ changed. ArrowUtils (internal, PAIMON_EXPORT) gains IsParquetDictionaryValueType, ResolveParquetDictionaryStructType and FlattenUnresolvableDictionaries; two private AppendOnlyFileStoreWrite helpers changed signature.

No storage format or protocol change. Output files remain standard Parquet — a forwarded dictionary is written through Arrow's ordinary dictionary path, and a file written with the option on is readable by any reader.

One new table option, parquet.read.enable-dictionary-passthrough, default false. The append compaction rewrite turns it on for itself when eligible and forces it off when not; setting it to false on the table disables the optimization.

Documentation

Yes — docs/source/user_guide/compaction.rst gains a "Dictionary Passthrough" section covering the scope (append-only only), per-file eligibility, the three gating conditions, file-index behaviour, the plain-fallback trade-off and the kill switch.

Generative AI tooling

Generated-by: Claude Code (Claude Opus 5)

🤖 Generated with Claude Code

[lucasfang]

LGTM

[lxy-9602]

I suggest keeping automatic enablement of parquet.read.enable-dictionary-passthrough disabled for compaction initially. Users could enable it after confirming a benefit with benchmarks representative of their workloads.

Velox has its own Parquet implementation, whose behavior is not necessarily consistent with Arrow Parquet. Therefore, before enabling this optimization by default, I would like to see before-and-after results from Paimon C++ using representative production data, including compaction time, CPU usage, peak memory, and output file size.

I am also concerned about a possible side effect related to output row-group boundaries. In the current Parquet write path, row groups are split not only according to the configured row-group size and row count, but also according to the writer’s memory limit.

For example, suppose file0 and file1 each contain three row groups. If compaction produces an output file with six aligned and independent row groups, each output column chunk could reuse this PR’s dictionary passthrough optimization as expected. However, the output row-group boundaries may not align with the input boundaries—for example, compaction might produce four row groups instead. In that case, one output row group may contain batches carrying different dictionaries from multiple input row groups or files. Arrow may retain the first dictionary and fall back to PLAIN encoding for the remaining data in that column chunk. Potentially, multiple output row groups could therefore become only partially dictionary-encoded, making the column ineligible for subsequent dictionary passthrough and possibly increasing the output file size.

[lxy-9602]

Could we simplify the comments a bit? They’re quite dense right now and getting hard to follow.

[lxy-9602]

Could we use a more general function name here and avoid mentioning Parquet — something like IsBinaryType? Judging from the implementation alone, it doesn’t really seem Parquet-specific.

[lxy-9602]

May prefer CastingUtils::Cast.

[lxy-9602]

I’m a bit concerned that using a temporary arrow_pool here is somewhat risky — future changes could easily break its lifetime assumptions. Would it make sense to turn it into a member variable instead?

[lxy-9602]

Could we use string literals directly here? Some engines have reimplemented the Parquet format layer, and I’m not sure their plugins still expose symbols like PARQUET_ENABLE_DICTIONARY and DEFAULT_IS_DICTIONARY_ENABLED.

[lxy-9602]

It looks like dictionary_batch_schema_ and dictionary_batch_type_ were made member variables mainly to avoid repeated arrow::schema conversion overhead. Have we already identified that as a clear hotspot? I feel ResolveBatchSchema could just return directly. Also, if the dictionary mode can switch, the benefit of caching these as member variables seems fairly limited.

[lxy-9602]

CastingUtils maybe better

[lxy-9602]

Thanks a lot for also fixing the orthogonal compaction bug in ORC when orc.read.enable-lazy-decoding is enabled. We’ll follow up with a corresponding fix for PK tables as well.