SteNicholas commented on code in PR #257:
URL: https://github.com/apache/paimon-cpp/pull/257#discussion_r3892511954
##########
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:
Fair point, and it was a real layering break:
`append_only_file_store_write.cpp` was the only file outside
`src/paimon/format/parquet/` including `parquet_format_defs.h`, plus it pulled
in `parquet/properties.h` for the default.
Both includes are gone. The option names and their defaults are now local
`constexpr` in an anonymous namespace in that `.cpp`, so core links no Parquet
symbols at all and a replacement Parquet plugin cannot break the build here.
The file already compared the format identifier against a plain `"parquet"`
literal, so this is consistent with what was there.
One small deviation from the literal suggestion: I used named constants
rather than inlining the strings, because
`parquet.read.enable-dictionary-passthrough` is needed in two functions and
writing it twice invites a typo. They are string literals in the same
translation unit, so the dependency is gone either way. Happy to inline them if
you would rather.
##########
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:
No, it was not a measured hotspot — I had reasoned about it rather than
profiled it, which is not a good enough reason for mutable state on the writer.
Both members are gone and `ResolveBatchSchema()` builds and returns the schema
directly; it is `const` now as a result.
Your second point is the stronger one: the cache was keyed on the batch
type, so an alternating encoding would have rebuilt it on every batch anyway,
and it added a way for the cached schema to drift from the batch it describes.
Velox has to reconcile exactly that (it caches its Arrow schema for field-id
fixups and then rewrites cached fields when a column's encoding changes across
batches); rebuilding per batch means there is no such window here.
##########
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:
Done — `FlattenUnwritableDictionaries()` uses `CastingUtils::Cast()` with
the writer's pool, so the local `arrow::compute::ExecContext` is gone here too.
##########
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 — and please do. The ORC path here only exercises it from the append
side: with `orc.read.enable-lazy-decoding` on, the reader hands over
`dictionary(int64, large_utf8)`, whose index and offset widths no `ArrowArray`
layout can report, so `CompactRewrite` decodes it at the source through
`FlattenUnresolvableDictionaries()` while the type is still known.
`AppendCompactionInteTest.TestAppendTableCompactionDictionaryPassthrough` is
parameterised over ORC precisely to keep that path covered. If the PK fix needs
the same decode-before-export step, that helper is reusable as is.
--
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]
