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

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