perf(parquet): decode every dictionary when the passthrough is vetoed

Addresses the review on #257.

`FlattenUnresolvableDictionaries()` decided what to decode from whether a
dictionary survives the Arrow C data interface, but what decides it is whether
the writer receiving the batch recovers an encoding from the layout at all, and
only ParquetFormatWriter does. A reader can hand over a dictionary for reasons
the Parquet passthrough option does not govern - a format's own lazy-decoding
setting - and those stay on when the rewrite vetoes the passthrough. A
`dictionary(int32, utf8)` column could therefore reach a writer that imports it
against the logical type, where a dictionary child fails on the buffer count.
The built-in ORC reader widens to `dictionary(int64, large_utf8)`, which was
already decoded for its shape, so reaching this needs a reader outside this
repository.

CompactRewrite now derives `preserve_layout_recoverable_dictionaries` from the
veto: under a veto every dictionary is decoded before the export, whatever its
shape. This also stops a dictionary reaching a Parquet writer whose table sets
`parquet.enable-dictionary` to false.

Rename ArrowUtils::ResolveParquetDictionaryStructType() to
ResolveDictionaryStructTypeFromLayout(), and the file-local
IsResolvableDictionary() to IsDictionaryLayoutRecoverable(). Neither is
Parquet-specific: they read `ArrowArray::dictionary` and the caller's declared
type, and DataFileWriterBase::AddFileIndexBatch runs the former for every
format. Moving it into the Parquet layer instead would make core depend on
Parquet symbols.

Author: SteNicholas

docs/source/user_guide/compaction.rst

@@ -126,6 +126,11 @@ the values again, or when variant/map shredding is configured because those
 writers reshape each batch against a fixed physical schema. Setting the option
 therefore never makes a rewrite fail; at worst it has no effect.
 
+When the option is vetoed, the rewrite also enforces the veto on every input
+batch. A format reader can independently hand over dictionary-encoded columns
+because of its own lazy-decoding setting, so the rewrite decodes those columns
+before handing them to a writer that cannot accept dictionary arrays.
+
 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.
 

src/paimon/common/utils/arrow/arrow_utils.cpp

@@ -44,7 +44,7 @@ namespace {
 // Whether `type` is a dictionary this can carry across the C data interface unchanged. The index
 // width is part of the test because nothing in a layout reveals it; see
 // ArrowUtils::IsDictionaryLayoutRecoverableValueType().
-bool IsResolvableDictionary(const arrow::DataType& type) {
+bool IsDictionaryLayoutRecoverable(const arrow::DataType& type) {
     if (type.id() != arrow::Type::DICTIONARY) {
         return false;
     }
@@ -558,7 +558,7 @@ Result<arrow::Compression::type> ArrowUtils::GetCompressionType(const std::strin
 // `int32` indices over `int32` offsets would silently reinterpret both buffers instead of failing.
 //
 // This narrows what may be carried; it cannot verify what was. See
-// ResolveParquetDictionaryStructType() for where the index width becomes a caller contract.
+// ResolveDictionaryStructTypeFromLayout() for where the index width becomes a caller contract.
 bool ArrowUtils::IsDictionaryLayoutRecoverableValueType(const arrow::DataType& type) {
     return arrow::is_binary_like(type.id());
 }
@@ -574,7 +574,7 @@ bool ArrowUtils::IsDictionaryLayoutRecoverableValueType(const arrow::DataType& t
 // batch whose dictionaries the schema does not declare, and it honours the contract by running
 // FlattenUnresolvableDictionaries() first. Closing the hole instead of narrowing it needs the real
 // `ArrowSchema` to reach the writer, which `FormatWriter::AddBatch(ArrowArray*)` drops.
-Result<std::shared_ptr<arrow::DataType>> ArrowUtils::ResolveParquetDictionaryStructType(
+Result<std::shared_ptr<arrow::DataType>> ArrowUtils::ResolveDictionaryStructTypeFromLayout(
     const std::shared_ptr<arrow::DataType>& logical_type, const ::ArrowArray* batch) {
     if (batch == nullptr || logical_type->id() != arrow::Type::STRUCT ||
         batch->n_children != logical_type->num_fields()) {
@@ -619,7 +619,8 @@ Result<std::shared_ptr<arrow::DataType>> ArrowUtils::ResolveParquetDictionaryStr
 
 Result<std::shared_ptr<arrow::StructArray>> ArrowUtils::FlattenUnresolvableDictionaries(
     const std::shared_ptr<arrow::StructArray>& batch,
-    const std::shared_ptr<arrow::DataType>& logical_type, arrow::MemoryPool* pool) {
+    const std::shared_ptr<arrow::DataType>& logical_type, arrow::MemoryPool* pool,
+    bool preserve_layout_recoverable_dictionaries) {
     const std::shared_ptr<arrow::DataType>& batch_type = batch->type();
     if (logical_type->id() != arrow::Type::STRUCT || !HasDictionary(*batch_type)) {
         return batch;
@@ -629,7 +630,13 @@ Result<std::shared_ptr<arrow::StructArray>> ArrowUtils::FlattenUnresolvableDicti
     arrow::FieldVector fields = batch_type->fields();
     for (int32_t i = 0; i < batch_type->num_fields(); ++i) {
         std::shared_ptr<arrow::Field> field = fields[i];
-        if (IsResolvableDictionary(*field->type()) || !HasDictionary(*field->type())) {
+        if (!HasDictionary(*field->type())) {
+            continue;
+        }
+        // Surviving the export is not enough when the destination imports against the logical
+        // type: an undeclared dictionary child of any shape then fails on the buffer count.
+        if (preserve_layout_recoverable_dictionaries &&
+            IsDictionaryLayoutRecoverable(*field->type())) {
             continue;
         }
         std::shared_ptr<arrow::Field> logical_field =

src/paimon/common/utils/arrow/arrow_utils.h

@@ -86,14 +86,14 @@ class PAIMON_EXPORT ArrowUtils {
     /// @return True when `dictionary(int32(), type)` round-trips through an `ArrowArray`.
     static bool IsDictionaryLayoutRecoverableValueType(const arrow::DataType& type);
 
-    /// Recovers the struct type of a batch that Arrow's Parquet reader produced with
-    /// `set_read_dictionary` enabled: `logical_type` with every top-level field whose matching
-    /// child in `batch` carries a dictionary replaced by `dictionary(int32(), field type)`, or
-    /// `logical_type` itself when no child is dictionary-encoded.
+    /// Recovers dictionary fields omitted from a batch's declared logical type by inspecting its
+    /// layout: `logical_type` with every top-level field whose matching child in `batch` carries a
+    /// dictionary replaced by `dictionary(int32(), field type)`, or `logical_type` itself when no
+    /// child is dictionary-encoded.
     ///
     /// The `int32` index width is assumed rather than inferred, so this is a contract on whoever
-    /// produces the batch, not a check the callers can rely on: the producer must either hand on a
-    /// batch that came straight from Arrow's Parquet reader, or run
+    /// produces the batch, not a check the callers can rely on: the producer must either provide a
+    /// batch whose dictionaries all have `int32` indices, or run
     /// FlattenUnresolvableDictionaries() while the type is still known. The definition spells out
     /// what that buys and what it does not.
     ///
@@ -106,27 +106,33 @@ class PAIMON_EXPORT ArrowUtils {
     /// @param batch Only its structure is inspected, never its data, and it is not consumed.
     /// @return `logical_type` or a copy of it carrying the recovered dictionary fields, or
     ///         NotImplemented for a dictionary this cannot describe.
-    static Result<std::shared_ptr<arrow::DataType>> ResolveParquetDictionaryStructType(
+    static Result<std::shared_ptr<arrow::DataType>> ResolveDictionaryStructTypeFromLayout(
         const std::shared_ptr<arrow::DataType>& logical_type, const ::ArrowArray* batch);
 
-    /// Returns `batch` with every top-level column that ResolveParquetDictionaryStructType() could
-    /// not resolve decoded to the type its field carries in `logical_type`. A column it can
-    /// resolve stays dictionary-encoded, so one column that has to be decoded does not cost the
-    /// others their encoding, and a batch that needs no decoding is returned unchanged.
+    /// Returns a copy of `batch` in which every top-level column that cannot be preserved for the
+    /// destination has been decoded to the type its field carries in `logical_type`. A layout-
+    /// recoverable column may stay dictionary-encoded, so one column that has to be decoded does
+    /// not cost the others their encoding, and a batch that needs no decoding is returned
+    /// unchanged.
     ///
-    /// The counterpart of the restriction above: an encoding that does not survive the export has
-    /// to be decoded while the type is still known.
+    /// The counterpart of the restriction above: an encoding the destination cannot take has to be
+    /// decoded while the type is still known.
     ///
     /// @param batch The batch to decode, matched to `logical_type` by field name; a column with no
     ///              matching field is left alone.
     /// @param logical_type The struct type the decoded columns are cast to. `batch` is returned
     ///                     unchanged when it is not a struct.
     /// @param pool Allocates the decoded columns. Only used when a column is actually decoded.
+    /// @param preserve_layout_recoverable_dictionaries Whether dictionaries recoverable through
+    ///                     ResolveDictionaryStructTypeFromLayout() may remain encoded. Pass false
+    ///                     to decode every dictionary not already declared by `logical_type`,
+    ///                     whatever its shape.
     /// @return `batch` itself when nothing had to be decoded, otherwise a copy of it with the
     ///         offset, length and validity of the original and the decoded columns swapped in.
     static Result<std::shared_ptr<arrow::StructArray>> FlattenUnresolvableDictionaries(
         const std::shared_ptr<arrow::StructArray>& batch,
-        const std::shared_ptr<arrow::DataType>& logical_type, arrow::MemoryPool* pool);
+        const std::shared_ptr<arrow::DataType>& logical_type, arrow::MemoryPool* pool,
+        bool preserve_layout_recoverable_dictionaries);
 
  private:
     static Status InnerCheckNullabilityMatch(const std::shared_ptr<arrow::Field>& field,