lxy-9602 commented on code in PR #257:
URL: https://github.com/apache/paimon-cpp/pull/257#discussion_r3885221387
##########
src/paimon/common/utils/arrow/arrow_utils.cpp:
##########
@@ -484,4 +536,95 @@ Result<arrow::Compression::type>
ArrowUtils::GetCompressionType(const std::strin
return compression_type;
}
+bool ArrowUtils::IsParquetDictionaryValueType(const arrow::DataType& type) {
+ return type.id() == arrow::Type::STRING || type.id() ==
arrow::Type::BINARY;
+}
Review Comment:
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.
##########
src/paimon/common/utils/arrow/arrow_utils.h:
##########
@@ -69,6 +71,84 @@ class PAIMON_EXPORT ArrowUtils {
/// Handles "none" and empty string by mapping them to "uncompressed".
static Result<arrow::Compression::type> GetCompressionType(const
std::string& compression);
+ /// Whether a column of `type` may be carried dictionary-encoded across
the Arrow C data
+ /// interface, which drops the type and leaves only the layout behind.
+ ///
+ /// A layout pins down neither the index width nor the offset width, so
the only encoding worth
+ /// carrying is the one a single known producer emits:
`dictionary(int32(), utf8()|binary())`,
+ /// which is what Arrow's Parquet reader produces for
+ /// `ArrowReaderProperties::set_read_dictionary`. `LARGE_STRING` is
deliberately excluded even
+ /// though it is binary-like: the ORC reader widens strings to
+ /// `dictionary(int64(), large_utf8())` under lazy decoding, and reading
that back as `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.
+ ///
+ /// This is the single definition shared by the reader that decides which
columns to request
+ /// encoded and by the writer that has to recognise them again on the
other side.
+ ///
+ /// @param type The column's value type, not its dictionary type.
+ /// @return True when `dictionary(int32(), type)` round-trips through an
`ArrowArray`.
+ static bool IsParquetDictionaryValueType(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.
+ ///
+ /// The `int32` index width is not inferred, it is assumed, and that
assumption is only valid
+ /// for Arrow's Parquet reader. **The value type check does not make it
safe for anything
+ /// else**: it rejects `dictionary(int64(), large_utf8())`, which is the
shape the ORC reader
+ /// produces, but nothing here can tell `dictionary(int32(), utf8())`
apart from
+ /// `dictionary(int64(), utf8())`, and the second would be read as the
first.
+ ///
+ /// So this is a contract, not a check, and it binds the code that
*produces* the batch rather
+ /// than the two places that call this. A producer must either be handing
on a batch that came
+ /// straight from Arrow's Parquet reader, or must run
FlattenUnresolvableDictionaries() while
+ /// the type is still known - that one does test the index width, and
decodes every column this
+ /// cannot resolve while leaving the rest encoded.
+ /// `AppendOnlyFileStoreWrite::CompactRewrite` is today's only production
path that can hand
+ /// over a batch whose dictionaries the schema does not declare, and it
takes the second route.
+ /// The callers themselves - `ParquetFormatWriter::ResolveBatchSchema` and
+ /// `DataFileWriterBase::AddFileIndexBatch` - are downstream of it and see
only the layout.
+ ///
+ /// The value-type rejection and the rejection of a dictionary below the
top level narrow the
+ /// blast radius; they do not close it. Closing it needs the real
`ArrowSchema` to reach the
+ /// writer, which the `FormatWriter::AddBatch(ArrowArray*)` signature
currently drops.
+ ///
+ /// A field that already carries a dictionary type is left alone:
`logical_type` then comes
+ /// from a caller that declared the encoding up front and already
describes the batch.
+ ///
+ /// @param logical_type The struct type the caller declares for the batch.
Returned unchanged
+ /// when it is not a struct or its field count does
not match `batch`,
+ /// leaving the mismatch to the import's own
diagnostics.
+ /// @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(
+ 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.
+ ///
+ /// This is the counterpart of the restriction above: exporting an array
through the C data
+ /// interface drops its type, so a column whose encoding does not survive
that round trip 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.
+ /// @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(
Review Comment:
Could we simplify the comments a bit? They’re quite dense right now and
getting hard to follow.
##########
src/paimon/format/parquet/parquet_format_writer.cpp:
##########
@@ -74,6 +81,51 @@ Status ParquetFormatWriter::AddBatch(ArrowArray* batch) {
return Status::OK();
}
+Result<std::shared_ptr<arrow::Schema>> ParquetFormatWriter::ResolveBatchSchema(
+ const ::ArrowArray* batch) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<arrow::DataType> batch_type,
+ ArrowUtils::ResolveParquetDictionaryStructType(logical_struct_type_,
batch));
+ if (batch_type == logical_struct_type_) {
+ return schema_;
+ }
+ if (dictionary_batch_type_ == nullptr ||
!dictionary_batch_type_->Equals(*batch_type)) {
+ dictionary_batch_type_ = batch_type;
+ dictionary_batch_schema_ = arrow::schema(batch_type->fields(),
schema_->metadata());
+ }
+ return dictionary_batch_schema_;
+}
+
+Result<std::shared_ptr<arrow::RecordBatch>>
ParquetFormatWriter::FlattenUnwritableDictionaries(
+ const std::shared_ptr<arrow::RecordBatch>& record_batch) const {
+ arrow::ArrayVector columns;
+ arrow::FieldVector fields;
+ arrow::compute::ExecContext exec_context(pool_.get());
+ for (int32_t i = 0; i < record_batch->num_columns(); ++i) {
+ const std::shared_ptr<arrow::Array>& column = record_batch->column(i);
+ if (column->type_id() != arrow::Type::DICTIONARY ||
+ checked_cast<const
arrow::DictionaryArray&>(*column).dictionary()->null_count() == 0) {
+ continue;
+ }
+ if (columns.empty()) {
+ columns = record_batch->columns();
+ fields = record_batch->schema()->fields();
+ }
+ const auto& dictionary_type = checked_cast<const
arrow::DictionaryType&>(*column->type());
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow::Datum flattened,
+ arrow::compute::Cast(column, dictionary_type.value_type(),
+ arrow::compute::CastOptions::Safe(),
&exec_context));
Review Comment:
`CastingUtils` maybe better
##########
src/paimon/core/io/data_file_index_writer.cpp:
##########
@@ -100,10 +104,40 @@ Status DataFileIndexWriter::AddBatch(const
std::shared_ptr<arrow::StructArray>&
if (finished_) {
return Status::Invalid("Data file index writer has already finished");
}
+ // Buffers allocated through the adaptor keep a raw pointer to it, so it
has to outlive every
+ // array decoded below. Built on first use, since most batches decode
nothing.
+ std::unique_ptr<arrow::MemoryPool> arrow_pool;
+ // One entry per indexed column, not per index: a column carrying both a
bitmap and a bloom
Review Comment:
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?
##########
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.
+
Review Comment:
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.
##########
src/paimon/common/utils/arrow/arrow_utils.cpp:
##########
@@ -484,4 +536,95 @@ Result<arrow::Compression::type>
ArrowUtils::GetCompressionType(const std::strin
return compression_type;
}
+bool ArrowUtils::IsParquetDictionaryValueType(const arrow::DataType& type) {
+ return type.id() == arrow::Type::STRING || type.id() ==
arrow::Type::BINARY;
+}
+
+Result<std::shared_ptr<arrow::DataType>>
ArrowUtils::ResolveParquetDictionaryStructType(
+ 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()) {
+ // Leave the mismatch to the import, which reports it with its own
diagnostics.
+ return logical_type;
+ }
+ arrow::FieldVector fields;
+ bool has_dictionary = false;
+ for (int32_t i = 0; i < logical_type->num_fields(); ++i) {
+ const std::shared_ptr<arrow::Field>& field = logical_type->field(i);
+ const ::ArrowArray* child = batch->children[i];
+ if (child == nullptr || child->dictionary == nullptr) {
+ if (HasUndeclaredDictionaryChild(field->type(), child)) {
+ return Status::NotImplemented(fmt::format(
+ "column '{}' is dictionary-encoded below its top level,
which the Arrow "
+ "import cannot describe without the producer's schema",
+ field->name()));
+ }
+ fields.push_back(field);
+ continue;
+ }
+ if (field->type()->id() == arrow::Type::DICTIONARY) {
+ // The caller already declares the column as a dictionary, so its
type describes the
+ // batch and nothing has to be recovered from the layout.
+ fields.push_back(field);
+ continue;
+ }
+ if (!IsParquetDictionaryValueType(*field->type())) {
+ return Status::NotImplemented(fmt::format(
+ "dictionary-encoded column '{}' of type {} cannot be resolved
from the layout of "
+ "an ArrowArray, which pins down neither the index nor the
offset width",
+ field->name(), field->type()->ToString()));
+ }
+ has_dictionary = true;
+ fields.push_back(field->WithType(arrow::dictionary(arrow::int32(),
field->type())));
+ }
+ if (!has_dictionary) {
+ return logical_type;
+ }
+ return arrow::struct_(fields);
+}
+
+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>& batch_type = batch->type();
+ if (logical_type->id() != arrow::Type::STRUCT ||
!HasDictionary(*batch_type)) {
+ return batch;
+ }
+ const auto& logical_struct_type = checked_cast<const
arrow::StructType&>(*logical_type);
+ arrow::compute::ExecContext exec_context(pool);
+ std::shared_ptr<arrow::ArrayData> data;
+ 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())) {
+ continue;
+ }
+ std::shared_ptr<arrow::Field> logical_field =
+ logical_struct_type.GetFieldByName(field->name());
+ if (logical_field == nullptr) {
+ // Nothing says what this column should decode to, so leave it for
the import to
+ // report against its own schema.
+ continue;
+ }
+ if (data == nullptr) {
+ // Copy once, on the first column that has to be decoded: the
parent keeps its offset,
+ // length and validity, and only the child data is swapped
underneath it.
+ data = batch->data()->Copy();
+ }
+ // Decode the whole child rather than the slice the parent exposes, so
the replacement
+ // lines up with the offset and length the parent still carries.
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow::Datum decoded,
+ arrow::compute::Cast(arrow::MakeArray(data->child_data[i]),
logical_field->type(),
+ arrow::compute::CastOptions::Safe(),
&exec_context));
Review Comment:
May prefer `CastingUtils::Cast`.
##########
src/paimon/format/parquet/parquet_format_writer.cpp:
##########
@@ -74,6 +81,51 @@ Status ParquetFormatWriter::AddBatch(ArrowArray* batch) {
return Status::OK();
}
+Result<std::shared_ptr<arrow::Schema>> ParquetFormatWriter::ResolveBatchSchema(
+ const ::ArrowArray* batch) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<arrow::DataType> batch_type,
+ ArrowUtils::ResolveParquetDictionaryStructType(logical_struct_type_,
batch));
+ if (batch_type == logical_struct_type_) {
+ return schema_;
+ }
+ if (dictionary_batch_type_ == nullptr ||
!dictionary_batch_type_->Equals(*batch_type)) {
+ dictionary_batch_type_ = batch_type;
+ dictionary_batch_schema_ = arrow::schema(batch_type->fields(),
schema_->metadata());
+ }
+ return dictionary_batch_schema_;
Review Comment:
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.
##########
src/paimon/core/operation/append_only_file_store_write.cpp:
##########
@@ -278,9 +299,25 @@ Result<AppendOnlyFileStoreWrite::WriterFactory>
AppendOnlyFileStoreWrite::GetDat
data_file_path_factory, pool_);
}
+Result<bool> AppendOnlyFileStoreWrite::CanUseDictionaryPassthrough(
+ const std::shared_ptr<ShreddingWritePlanFactory>& plan_factory) const {
+ std::shared_ptr<FileFormat> file_format = options_.GetFileFormat();
+ if (!file_format || file_format->Identifier() != "parquet") {
+ return false;
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ bool enable_dictionary,
+ OptionsUtils::GetValueFromMap<bool>(options_.ToMap(),
parquet::PARQUET_ENABLE_DICTIONARY,
+
::parquet::DEFAULT_IS_DICTIONARY_ENABLED));
+ if (!enable_dictionary) {
Review Comment:
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`.
##########
test/inte/append_compaction_inte_test.cpp:
##########
@@ -822,4 +824,239 @@ TEST_F(AppendCompactionInteTest,
TestAppendTableCompactionWithIOException) {
ASSERT_TRUE(compaction_run_complete);
}
+// Rewriting through the dictionary passthrough has to produce the same table
as rewriting through
+// materialized values, whatever encoding each input file happens to carry.
The interesting part is
+// the chain the unit tests cannot reach on their own: CompactRewrite hands
the batch to the file
+// index writer and to the format writer, and both of them recover each
column's encoding from the
+// batch layout after the type has been dropped by the C data interface.
+//
+// Parameterised over the two formats that have an encoding to forward or to
suppress: Parquet
+// turns the passthrough on, ORC forces it off because its writer cannot take
a dictionary-encoded
+// batch. ORC lazy decoding is on throughout, which makes the ORC reader hand
over
+// `dictionary(int64, large_utf8)` - a shape no layout can resolve, so it
exercises the
+// decode-at-the-source path rather than the passthrough.
+TEST_P(AppendCompactionInteTest,
TestAppendTableCompactionDictionaryPassthrough) {
+ auto file_format = GetParam();
+ if (file_format != "parquet" && file_format != "orc") {
+ GTEST_SKIP() << file_format << " has no dictionary encoding to forward
or to suppress";
+ }
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+
+ // `s` and `b` are low-cardinality and come back encoded; `id` is INT32,
which the gate excludes
+ // by physical type, so the rewrite carries both kinds of column at once.
`u` holds a distinct
+ // value per row, the shape passthrough saves least on. The
cardinality-driven half of the gate
+ // - a column that starts dictionary-encoded and falls back to plain
partway through a file -
+ // needs more rows than a readable fixture holds and is covered by
+ //
ParquetFileBatchReaderTest.TestDictionaryPassthroughSkipsFallbackToPlain
instead.
+ arrow::FieldVector fields = {
+ arrow::field("id", arrow::int32()), arrow::field("s", arrow::utf8()),
+ arrow::field("b", arrow::binary()), arrow::field("u", arrow::utf8())};
+ auto schema = arrow::schema(fields);
+
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, file_format},
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "id"},
+ {Options::FILE_SYSTEM, "local"},
+ {"orc.read.enable-lazy-decoding", "true"},
+ // Above the distinct/total ratio of every column here, so ORC
dictionary-encodes rather
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
