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

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. Forward the encoding
instead.

Reader: `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.

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; 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 - and leaves the rest encoded. A dictionary holding nulls in its values is
flattened at the writer, the one shape parquet::arrow rejects outright.

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.
A file index on a forwarded column materializes that column alone.

Note that 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.

Author: SteNicholas

benchmark/parquet_format_benchmark.cpp

@@ -51,6 +51,7 @@
 #include "arrow/c/helpers.h"
 #include "arrow/util/bit_util.h"
 #include "benchmark/benchmark.h"
+#include "fmt/format.h"
 #include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
 #include "paimon/common/utils/arrow/mem_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
@@ -104,6 +105,10 @@ constexpr int32_t kWriteBatchSize = 1024;
 constexpr int64_t kPageSizeBytes = 64 * 1024;
 // Four row groups per read fixture, so row-group pruning and page pruning are both in play.
 constexpr int64_t kRowGroupLength = 25'000;
+// Only StringFixture lowers this from arrow's 1MB default: at 1MB a 25K-row row group of distinct
+// `value_<n>` entries still fits, so no cardinality this file writes would ever make the writer
+// fall back to plain and the passthrough gate would have nothing to decline.
+constexpr int64_t kDictionaryPageSizeBytes = 64 * 1024;
 constexpr int64_t kStringCardinality = 1'000;
 // Few enough distinct values that arrow keeps the column dictionary-encoded for the whole file,
 // which is the shape the wide-schema case wants: per-column work small, per-batch cost visible.
@@ -160,6 +165,11 @@ std::shared_ptr<arrow::Schema> DecimalSchema(int32_t precision) {
     return arrow::schema({MakeField("amount", arrow::decimal128(precision, 4), 0)});
 }
 
+// One STRING column, so a dictionary case measures one encoder and nothing else.
+std::shared_ptr<arrow::Schema> StringSchema() {
+    return arrow::schema({MakeField("name", arrow::utf8(), 0)});
+}
+
 std::shared_ptr<arrow::Schema> DoubleSchema() {
     return arrow::schema({MakeField("value", arrow::float64(), 0)});
 }
@@ -422,6 +432,18 @@ BatchFactory SingleColumnBatch(const ColumnFactory& make_column) {
     };
 }
 
+// The same, but the batch is typed by the column rather than by the schema, so it can carry an
+// encoding the schema does not declare. That is the shape a compaction rewrite produces: the file
+// writer is built from the table's logical schema while the reader forwards whatever encoding the
+// input file already had, leaving the writer to recover it from the batch.
+BatchFactory SingleEncodedColumnBatch(const ColumnFactory& make_column) {
+    return [make_column](const std::shared_ptr<arrow::Schema>& schema, int64_t offset,
+                         int64_t rows) -> Result<std::shared_ptr<arrow::Array>> {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> column, make_column(rows, offset));
+        return MakeStructArray({schema->field(0)->WithType(column->type())}, {column});
+    };
+}
+
 Result<std::shared_ptr<arrow::Array>> MakeNullableFlatBatch(
     const std::shared_ptr<arrow::Schema>& schema, int64_t offset, int64_t rows, int64_t null_pct) {
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> ids, MakeInt64Column(rows, offset));
@@ -766,6 +788,27 @@ const ReadFixture& DoubleFixture() {
     return ColumnFixture("double", DoubleSchema(), &MakeDoubleColumn);
 }
 
+// A one-column STRING file at a chosen cardinality, written with a reduced dictionary page limit
+// so both regimes the passthrough gate distinguishes are reachable within a 100K-row file. Under
+// the limit every data page stays dictionary-encoded and the gate lets the column through; over
+// it the writer emits the dictionary page it has and encodes the rest as plain, which is the case
+// the gate has to decline. At kDictionaryPageSizeBytes a row group holds roughly 4K distinct
+// `value_<n>` entries before overflowing, so cardinality alone picks the regime.
+const ReadFixture& StringFixture(int64_t cardinality) {
+    const std::string key = fmt::format("string_{}", cardinality);
+    return GetFixture(key, [key, cardinality] {
+        std::map<std::string, std::string> options;
+        options[paimon::parquet::PARQUET_DICTIONARY_PAGE_SIZE] =
+            std::to_string(kDictionaryPageSizeBytes);
+        return std::make_unique<ReadFixture>(
+            key + ".parquet", StringSchema(),
+            SingleColumnBatch([cardinality](int64_t rows, int64_t offset) {
+                return MakeStringColumn(rows, offset, cardinality);
+            }),
+            options);
+    });
+}
+
 // The same data with dictionary encoding off, giving the read side a plain baseline.
 const ReadFixture& PlainFlatFixture() {
     return GetFixture("flat_plain", [] {
@@ -930,6 +973,21 @@ void BM_ParquetWrite_DictionaryString(::benchmark::State& state) {
                             kRowsPerBatch, /*options=*/{});
 }
 
+// arg: dictionary cardinality. The shape the append compaction rewrite actually produces, and the
+// one BM_ParquetWrite_DictionaryString does not cover: there the schema itself is a DictionaryType,
+// here the writer is built from a plain STRING schema - as a rewrite builds it, from the table's
+// logical schema - and the batch arrives dictionary-encoded anyway. The delta against
+// BM_ParquetWrite_String at the same cardinality is what the passthrough buys on the write side,
+// including the per-batch schema fixup that recovers the encoding from the batch layout.
+void BM_ParquetWrite_DictionaryStringIntoStringSchema(::benchmark::State& state) {
+    const int64_t cardinality = state.range(0);
+    RunWriteBenchmark(state, StringSchema(),
+                      SingleEncodedColumnBatch([cardinality](int64_t rows, int64_t offset) {
+                          return MakeDictionaryStringColumn(rows, offset, cardinality);
+                      }),
+                      kRowsPerBatch, /*options=*/{}, kDefaultCompression);
+}
+
 // The same axis on an INTEGER dictionary, which arrow cannot direct-write - is_base_binary_like
 // excludes int32, so it densifies first. Its baseline is BM_ParquetWrite_FlatInt32 at the same
 // cardinality, not the String case: only the flat INT32 control holds value, width and encoding
@@ -1197,6 +1255,22 @@ void BM_ParquetRead_Encoding(::benchmark::State& state, bool enable_dictionary)
                      /*selection_bitmap=*/std::nullopt, /*options=*/{}, kReadBatchSize);
 }
 
+// args: string cardinality, and whether the parquet dictionary passthrough is on. With it on, a
+// column the file stores dictionary-encoded end to end is handed back as a DictionaryArray instead
+// of one materialized value per row, so the pair at a fixed cardinality is what the read half of
+// the compaction rewrite saves. At a cardinality high enough that the writer fell back to plain,
+// the gate declines and the two runs measure the same work - a divergence there means the gate
+// stopped looking at the data page encodings and started trusting the dictionary page.
+void BM_ParquetRead_DictionaryPassthrough(::benchmark::State& state) {
+    const int64_t cardinality = state.range(0);
+    const bool enable_passthrough = state.range(1) != 0;
+    std::map<std::string, std::string> options;
+    options[paimon::parquet::PARQUET_READ_ENABLE_DICTIONARY_PASSTHROUGH] =
+        enable_passthrough ? "true" : "false";
+    RunReadBenchmark(state, StringFixture(cardinality), StringSchema(), /*predicate=*/nullptr,
+                     /*selection_bitmap=*/std::nullopt, options, kReadBatchSize);
+}
+
 // arg: decimal precision, the read side of BM_ParquetWrite_Decimal. Precision picks the physical
 // type - INT32, INT64 or FIXED_LEN_BYTE_ARRAY, since ParquetWriterBuilder enables
 // store_decimal_as_integer - and the three take different paths back to Decimal128Array.
@@ -1272,6 +1346,16 @@ BENCHMARK(BM_ParquetWrite_DictionaryString)
     ->Arg(10000)
     ->Unit(benchmark::kMillisecond)
     ->UseRealTime();
+// Same cardinality axis as BM_ParquetWrite_String and BM_ParquetWrite_StringNoDictionary, which
+// are its baselines: the three have to line up point for point or the low/medium/high comparison
+// cannot be made.
+BENCHMARK(BM_ParquetWrite_DictionaryStringIntoStringSchema)
+    ->ArgName("cardinality")
+    ->Arg(10)
+    ->Arg(1000)
+    ->Arg(kRowsPerFile)
+    ->Unit(benchmark::kMillisecond)
+    ->UseRealTime();
 BENCHMARK(BM_ParquetWrite_DictionaryInt32)
     ->ArgName("cardinality")
     ->Arg(10)
@@ -1402,6 +1486,19 @@ BENCHMARK_CAPTURE(BM_ParquetRead_Encoding, dictionary, true)
 BENCHMARK_CAPTURE(BM_ParquetRead_Encoding, plain, false)
     ->Unit(benchmark::kMillisecond)
     ->UseRealTime();
+// The same cardinality axis the write cases use, so the read and write halves of a rewrite can be
+// added up at each point. At kRowsPerFile every value is distinct, which overflows
+// kDictionaryPageSizeBytes and is the point where the gate has to decline.
+BENCHMARK(BM_ParquetRead_DictionaryPassthrough)
+    ->ArgNames({"cardinality", "passthrough"})
+    ->Args({10, 0})
+    ->Args({10, 1})
+    ->Args({1000, 0})
+    ->Args({1000, 1})
+    ->Args({kRowsPerFile, 0})
+    ->Args({kRowsPerFile, 1})
+    ->Unit(benchmark::kMillisecond)
+    ->UseRealTime();
 BENCHMARK(BM_ParquetRead_Decimal)
     ->ArgName("precision")
     ->Arg(9)

docs/source/user_guide/compaction.rst

@@ -88,6 +88,40 @@ After compaction, if the last output file is still smaller than
 ``compaction.file-size``, it is placed back into the compaction queue for future
 merging.
 
+Dictionary Passthrough
+~~~~~~~~~~~~~~~~~~~~~~
+An append-only compaction rewrite copies rows into the new file without
+inspecting any value, so a Parquet column that an input file already stores
+dictionary-encoded is forwarded to the writer still encoded instead of being
+expanded to one copy of the value per row and re-encoded. This saves the reader
+materializing the values and the writer hashing them again; how much that is
+worth depends on the column, and low-cardinality ``STRING``/``BINARY`` columns
+benefit most. Primary-key compaction merges rows and is not covered.
+
+This applies automatically. Eligibility is decided per input file: a non-nested
+``STRING``/``BINARY`` column is forwarded when its data pages are
+dictionary-encoded throughout every row group of *that* file, so one input file
+can be read encoded while the next one is read as ordinary values, and the
+writer takes both. A high-cardinality column that started dictionary-encoded and
+fell back to plain encoding therefore does not qualify, even though it still
+carries a dictionary page. Passthrough is also skipped when the table writes a
+format other than Parquet, when ``parquet.enable-dictionary`` is ``false``
+because the writer would only expand the values again, or when variant/map
+shredding is configured because those writers reshape each batch against a fixed
+physical schema.
+
+If a file index is configured on a forwarded column, that column alone is
+materialized so the index still sees its values; the other columns stay encoded.
+
+Passthrough changes what the rewrite costs, not what it produces, with one
+exception worth knowing: a Parquet column chunk can only carry one dictionary,
+so when the input files supply different dictionaries the output column keeps
+the first and falls back to plain encoding for the rest of the row group. The
+rewritten data is unchanged either way, but the output file may be larger than a
+rewrite that rebuilt a single dictionary from materialized values. Set
+``parquet.read.enable-dictionary-passthrough`` to ``false`` on the table to turn
+the optimization off and always rebuild.
+
 Append-Only Table Compaction Options
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

src/paimon/common/reader/reader_utils_test.cpp

@@ -28,8 +28,10 @@
 #include "arrow/api.h"
 #include "arrow/array/array_base.h"
 #include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
 #include "arrow/ipc/json_simple.h"
 #include "gtest/gtest.h"
+#include "paimon/common/utils/checked_cast.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/read_result_collector.h"
 #include "paimon/testing/utils/testharness.h"
@@ -104,4 +106,49 @@ TEST(ReaderUtilsTest, TestApplyBitmapToReadBatch) {
                  "except eof");
 }
 
+TEST(ReaderUtilsTest, TestApplyBitmapToReadBatchKeepsDictionaryEncoding) {
+    // A deletion vector on an append table routes the Parquet dictionary passthrough through here:
+    // ParquetFileBatchReader reports SupportPreciseBitmapSelection() == false, so RawFileSplitRead
+    // wraps it and the surviving rows are cut out by slicing and concatenating. The encoding
+    // survives that only because every slice shares one dictionary and arrow::Concatenate has a
+    // fast path for it; if it ever unified or densified instead, a compaction with deletion
+    // vectors would quietly stop forwarding the encoding the rewrite asked for.
+    auto dictionary_type = arrow::dictionary(arrow::int32(), arrow::utf8());
+    auto make_encoded = [&dictionary_type](const std::string& indices_json) {
+        return arrow::DictionaryArray::FromArrays(
+                   dictionary_type,
+                   arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), indices_json)
+                       .ValueOrDie(),
+                   arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(), R"(["a", "b"])")
+                       .ValueOrDie())
+            .ValueOrDie();
+    };
+    std::shared_ptr<arrow::Array> ids =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[0, 1, 2, 3, 4]").ValueOrDie();
+    auto src_array = arrow::StructArray::Make({make_encoded("[0, 1, 0, 1, 0]"), ids},
+                                              std::vector<std::string>{"s", "id"})
+                         .ValueOrDie();
+
+    ASSERT_OK_AND_ASSIGN(auto src_batch, ReadResultCollector::GetReadBatch(src_array));
+    // Two disjoint runs, so the filter has to concatenate rather than hand back a single slice.
+    auto batch_with_bitmap =
+        std::make_pair(std::move(src_batch), RoaringBitmap32::From(std::vector<int32_t>{0, 1, 4}));
+    ASSERT_OK_AND_ASSIGN(auto result_batch,
+                         ReaderUtils::ApplyBitmapToReadBatch(std::move(batch_with_bitmap),
+                                                             arrow::default_memory_pool()));
+    // Imported directly rather than through ReadResultCollector::GetArray, which decodes
+    // dictionaries on the way out and would hide the very thing being asserted.
+    auto& [c_array, c_schema] = result_batch;
+    std::shared_ptr<arrow::Array> result =
+        arrow::ImportArray(c_array.get(), c_schema.get()).ValueOrDie();
+    ASSERT_EQ(3, result->length());
+    auto result_struct = checked_pointer_cast<arrow::StructArray>(result);
+    ASSERT_EQ(arrow::Type::DICTIONARY, result_struct->field(0)->type()->id());
+    ASSERT_TRUE(result_struct->field(0)->Equals(*make_encoded("[0, 1, 0]")))
+        << "actual=" << result_struct->field(0)->ToString();
+    std::shared_ptr<arrow::Array> expected_ids =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[0, 1, 4]").ValueOrDie();
+    ASSERT_TRUE(result_struct->field(1)->Equals(*expected_ids));
+}
+
 }  // namespace paimon::test