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. parquet.read.enable-dictionary-passthrough makes ParquetFileBatchReader request set_read_dictionary for non-nested STRING 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.

BINARY is deliberately not forwarded, although Parquet stores it in the same BYTE_ARRAY leaf and dictionary-encodes it the same way: no value accessor here can read a dictionary(int32, binary). ColumnarUtils::GetView() asserts on a dictionary whose values are neither STRING nor LARGE_STRING and returns an empty string_view in a release build, and LiteralConverter rejects it outright. Every consumer understands a STRING dictionary because the ORC reader has always produced one under lazy decoding; none was ever handed a BINARY one. Widening this needs those consumers first.

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) is recoverable — ArrowUtils::IsDictionaryLayoutRecoverableValueType(). That is the writer's capability and it matches Velox's VARCHAR/VARBINARY; the STRING-only restriction above belongs to the reader, not to the format, and is applied at the gate so the two do not get conflated. CompactRewrite 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. The option is off by default and the rewrite never turns it on; enabling it is the table's decision, taken after measuring. GetDictionaryPassthroughVetoReason() only vetoes it — when the output is not Parquet, when parquet.enable-dictionary is false, or when a shredding plan is active — and logs the reason at DEBUG for a table that did ask for the passthrough. Setting the option therefore never makes a rewrite fail; at worst it has no effect. 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 that row group. Output row groups are cut by parquet.block.size and by the writer's memory limit, which are not aligned to input file boundaries. The rewritten data is unchanged, but the output file can be larger than one written from materialized values, and that column then loses eligibility for the next compaction round. This is why the option is opt-in rather than automatic, and it is documented in compaction.rst.

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 two places. 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. And Velox's reader can hand out a dictionary for VARBINARY because its consumers are type-erased over StringView, while the accessors here switch on the Arrow type id — hence the reader-side STRING restriction.

Tests

Unit:

• ArrowUtilsTest.TestResolveParquetDictionaryStructType — layout-derived resolution; rejects non-binary-like 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 forwarded columns across two batches with different dictionaries, which also covers the writer-held Arrow pool.
• ParquetFileBatchReaderTest.TestDictionaryPassthrough — on / explicitly off / option absent / file without dictionary pages.
• ParquetFileBatchReaderTest.TestDictionaryPassthroughSkipsBinaryColumn — f8 (STRING) is forwarded and f9 (BINARY) is not, from the same file with the same physical type and 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 / OfBinaryColumn / OfUnsupportedTypeIsRejected.
• ParquetFormatWriterTest.TestGetEstimateLengthWithDictionaryBatches — GetEstimateLength() and ReachTargetSize() still drive file rolling when batches arrive encoded.

Integration:

• AppendCompactionInteTest.TestAppendTableCompactionDictionaryPassthrough (Parquet + ORC) — the table opts in; asserts the input read types (s as dictionary(int32, utf8) on Parquet, id and b materialized, s as dictionary(int64, large_utf8) on ORC), a full read of those still-encoded files through the whole read path, the rewrite, that the output column has fallen back to plain, and a predicate read through the bitmap index.
• AppendCompactionInteTest.TestAppendTableCompactionDictionaryPassthroughDefaultOff — a table that says nothing about the option lands on the same rows and on a better encoded output, which is why it is not on by default.
• AppendCompactionInteTest.TestAppendTableCompactionDictionaryPassthroughVetoedByShredding — the option is set and vetoed; the input arrives encoded and the compacted column comes back dictionary-encoded end to end, which only a rewrite from materialized values produces.

Benchmarks (benchmark/parquet_format_benchmark.cpp): BM_ParquetWrite_String, BM_ParquetWrite_DictionaryStringIntoStringSchema and BM_ParquetWrite_ChangingDictionaryStringIntoStringSchema write the same logical column at the same cardinality — the third rotates the alphabet and shifts the indices back by the same amount, so every batch brings a different dictionary over identical data. The second against the first is what forwarding buys; the third against the second is what the fallback costs in time; the third against the first is the output size a rewrite that materialized and rebuilt would have produced. parquet_format_benchmark_test.cpp pins that premise. These are format-writer microbenchmarks and do not stand in for compaction time, CPU and peak memory on a real table — that measurement is what a user should make before enabling the option, which is why it ships off by default.

API and Format

• New ArrowUtils::IsDictionaryLayoutRecoverableValueType() and ArrowUtils::FlattenUnresolvableDictionaries().
• New table option parquet.read.enable-dictionary-passthrough, default false. It is a read option, so it applies to every read of the table and not only to the compaction rewrite; an eligible column then reaches the consumer as an Arrow DictionaryArray.
• AppendOnlyFileStoreWrite reads the two Parquet option names as local constants instead of including the format layer's headers, so core does not depend on symbols a replacement Parquet plugin need not export.

Documentation

• docs/source/user_guide/compaction.rst — a "Dictionary Passthrough" section: what it does, that it is opt-in and why, the read-option caveat, per-file eligibility and the vetoes, and a "Trade-off" subsection on the single dictionary per column chunk.
• docs/source/examples/benchmark.rst — the flat / shared-dictionary / changing-dictionary trio and what each pair measures.

[lucasfang]

LGTM

[lszskye]

Some format(AliORC), will lazy decoding exposes dictionary-encoded STRING columns as dictionary<int32, utf8>. This type is considered resolvable by IsResolvableDictionary(), so FlattenUnresolvableDictionaries() keeps it encoded during append-only compaction.
However, the non-Parquet veto only disables parquet.read.enable-dictionary-passthrough; it does not disable aliorc.read.enable-lazy-decoding. As a result, an AliORC dictionary batch can still reach OrcFormatWriter::AddBatch(), where it is imported using the logical utf8 schema. The actual ArrowArray child is dictionary-encoded, so the import will fail. Even if the dictionary type were recovered during import, OrcAdapter::WriteBatch() does not support arrow::Type::DICTIONARY.

[lszskye]

See one function about parquet in framework is a little bit weird