From 30f8f7ef0293289ef0e0fe88f44d76f869122970 Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Mon, 24 Aug 2026 10:53:37 +0800 Subject: [PATCH 1/2] perf(manifest): reuse struct fields while reading rows --- src/paimon/core/manifest/manifest_file.cpp | 3 ++- src/paimon/core/utils/objects_file.h | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/paimon/core/manifest/manifest_file.cpp b/src/paimon/core/manifest/manifest_file.cpp index 1be49d0b..9f9c8aee 100644 --- a/src/paimon/core/manifest/manifest_file.cpp +++ b/src/paimon/core/manifest/manifest_file.cpp @@ -92,8 +92,9 @@ Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t buc return ReadArrowBatches( file_name, [this, bucket, entries](const std::shared_ptr& batch) -> Status { + const arrow::ArrayVector& fields = batch->fields(); for (int64_t i = 0; i < batch->length(); i++) { - ColumnarRow row(batch->fields(), pool_, i); + ColumnarRow row(fields, pool_, i); PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(row.GetInt(0))); if (ManifestEntrySerializer::GetBucket(row) != bucket) { continue; diff --git a/src/paimon/core/utils/objects_file.h b/src/paimon/core/utils/objects_file.h index a56952ae..b3135b31 100644 --- a/src/paimon/core/utils/objects_file.h +++ b/src/paimon/core/utils/objects_file.h @@ -134,8 +134,9 @@ Status ObjectsFile::Read(const std::string& file_name, file_name, [this, &filter, result](const std::shared_ptr& struct_array) -> Status { result->reserve(result->size() + struct_array->length()); + const arrow::ArrayVector& fields = struct_array->fields(); for (int64_t i = 0; i < struct_array->length(); i++) { - ColumnarRow row(struct_array->fields(), pool_, i); + ColumnarRow row(fields, pool_, i); PAIMON_ASSIGN_OR_RAISE(T obj, serializer_->FromRow(row)); if (filter) { PAIMON_ASSIGN_OR_RAISE(bool filter_res, filter(obj)); From ce7269142377773e813b7ea12ce6edae71af902d Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Mon, 24 Aug 2026 10:53:45 +0800 Subject: [PATCH 2/2] perf(avro): cache builder type metadata during decode --- .../format/avro/avro_direct_decoder.cpp | 28 +++++++++++++++---- src/paimon/format/avro/avro_direct_decoder.h | 20 +++++++++++++ .../avro/avro_direct_encoder_decoder_test.cpp | 23 +++++++++++++++ .../format/avro/avro_file_batch_reader.cpp | 1 + 4 files changed, 66 insertions(+), 6 deletions(-) diff --git a/src/paimon/format/avro/avro_direct_decoder.cpp b/src/paimon/format/avro/avro_direct_decoder.cpp index f837eed0..f9c8a9a4 100644 --- a/src/paimon/format/avro/avro_direct_decoder.cpp +++ b/src/paimon/format/avro/avro_direct_decoder.cpp @@ -33,6 +33,22 @@ namespace paimon::avro { +const AvroDirectDecoder::DecodeContext::BuilderMetadata& +AvroDirectDecoder::DecodeContext::GetBuilderMetadata(const arrow::ArrayBuilder* builder) { + auto iter = builder_metadata_.find(builder); + if (iter != builder_metadata_.end()) { + return iter->second; + } + + std::shared_ptr data_type = builder->type(); + BuilderMetadata metadata{data_type->id(), std::nullopt}; + if (data_type->id() == arrow::Type::TIMESTAMP) { + metadata.timestamp_unit = + checked_cast(data_type.get())->unit(); + } + return builder_metadata_.emplace(builder, metadata).first->second; +} + namespace { /// Forward declaration for mutual recursion. @@ -266,8 +282,8 @@ Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node, case ::avro::AVRO_INT: { int32_t value = decoder->decodeInt(); - auto arrow_type = array_builder->type(); - switch (arrow_type->id()) { + const auto& builder_metadata = ctx->GetBuilderMetadata(array_builder); + switch (builder_metadata.type) { case arrow::Type::INT8: { auto* builder = checked_cast(array_builder); PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(value)); @@ -287,7 +303,7 @@ Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node, if (logical_type.type() != ::avro::LogicalType::Type::DATE) { return Status::TypeError( fmt::format("Unexpected avro type [{}] with arrow type [{}].", - ::avro::toString(type), arrow_type->ToString())); + ::avro::toString(type), array_builder->type()->ToString())); } auto* builder = checked_cast(array_builder); PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(value)); @@ -296,7 +312,7 @@ Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node, default: return Status::TypeError( fmt::format("Unexpected avro type [{}] with arrow type [{}].", - ::avro::toString(type), arrow_type->ToString())); + ::avro::toString(type), array_builder->type()->ToString())); } } @@ -315,9 +331,9 @@ Status DecodeAvroValueToBuilder(const ::avro::NodePtr& avro_node, case ::avro::LogicalType::Type::LOCAL_TIMESTAMP_MICROS: case ::avro::LogicalType::Type::LOCAL_TIMESTAMP_NANOS: { auto* builder = checked_cast(array_builder); - auto ts_type = checked_cast(builder->type().get()); // for arrow second, we need to convert it from avro millisecond - if (ts_type->unit() == arrow::TimeUnit::type::SECOND) { + const auto& builder_metadata = ctx->GetBuilderMetadata(builder); + if (builder_metadata.timestamp_unit == arrow::TimeUnit::type::SECOND) { value /= DateTimeUtils::CONVERSION_FACTORS[DateTimeUtils::MILLISECOND]; } PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Append(value)); diff --git a/src/paimon/format/avro/avro_direct_decoder.h b/src/paimon/format/avro/avro_direct_decoder.h index c507091a..6422f915 100644 --- a/src/paimon/format/avro/avro_direct_decoder.h +++ b/src/paimon/format/avro/avro_direct_decoder.h @@ -22,7 +22,11 @@ #pragma once +#include #include +#include +#include +#include #include "arrow/array/builder_base.h" #include "avro/Decoder.hh" @@ -41,10 +45,26 @@ class AvroDirectDecoder { /// Avoids frequent small allocations by reusing temporary buffers across multiple decode /// operations. This is particularly important for string, binary, and decimal data types. struct DecodeContext { + struct BuilderMetadata { + arrow::Type::type type; + std::optional timestamp_unit; + }; + + /// Returns immutable type metadata without repeatedly copying the builder's DataType. + const BuilderMetadata& GetBuilderMetadata(const arrow::ArrayBuilder* builder); + + /// Clears metadata before the builder tree is replaced or destroyed. + void ClearBuilderMetadata() { + builder_metadata_.clear(); + } + // Scratch buffer for string decoding (reused across rows) std::string string_scratch; // Scratch buffer for binary/decimal data (reused across rows) std::vector bytes_scratch; + + private: + std::unordered_map builder_metadata_; }; /// Directly decode Avro data to Arrow array builders without GenericDatum diff --git a/src/paimon/format/avro/avro_direct_encoder_decoder_test.cpp b/src/paimon/format/avro/avro_direct_encoder_decoder_test.cpp index f276d946..78f4ca48 100644 --- a/src/paimon/format/avro/avro_direct_encoder_decoder_test.cpp +++ b/src/paimon/format/avro/avro_direct_encoder_decoder_test.cpp @@ -62,6 +62,7 @@ class AvroDirectEncoderDecoderTest : public ::testing::Test { auto decoder = ::avro::binaryDecoder(); decoder->init(*input_stream); + decode_ctx_.ClearBuilderMetadata(); for (int32_t i = 0; i < expected_count; ++i) { PAIMON_RETURN_NOT_OK(AvroDirectDecoder::DecodeAvroToBuilder( avro_node, projection, decoder.get(), builder, &decode_ctx_)); @@ -157,6 +158,18 @@ TEST_F(AvroDirectEncoderDecoderTest, TestIntegerTypes) { CheckResult(schema_json, input_array, &builder); } + // Test INT16 + { + std::string schema_json = R"({"type": "int"})"; + arrow::Int16Builder builder; + ASSERT_TRUE(builder.Append(1).ok()); + ASSERT_TRUE(builder.Append(-32768).ok()); + ASSERT_TRUE(builder.Append(32767).ok()); + std::shared_ptr input_array; + ASSERT_TRUE(builder.Finish(&input_array).ok()); + CheckResult(schema_json, input_array, &builder); + } + // Test INT32 { std::string schema_json = R"({"type": "int"})"; @@ -182,6 +195,16 @@ TEST_F(AvroDirectEncoderDecoderTest, TestIntegerTypes) { } } +TEST_F(AvroDirectEncoderDecoderTest, TestDecodeContextBuilderMetadataLifecycle) { + arrow::Int8Builder int8_builder; + ASSERT_EQ(decode_ctx_.GetBuilderMetadata(&int8_builder).type, arrow::Type::INT8); + + decode_ctx_.ClearBuilderMetadata(); + + arrow::Int16Builder int16_builder; + ASSERT_EQ(decode_ctx_.GetBuilderMetadata(&int16_builder).type, arrow::Type::INT16); +} + TEST_F(AvroDirectEncoderDecoderTest, TestFloatingPointTypes) { // Test FLOAT { diff --git a/src/paimon/format/avro/avro_file_batch_reader.cpp b/src/paimon/format/avro/avro_file_batch_reader.cpp index f48ec4cc..1e217f5c 100644 --- a/src/paimon/format/avro/avro_file_batch_reader.cpp +++ b/src/paimon/format/avro/avro_file_batch_reader.cpp @@ -172,6 +172,7 @@ Status AvroFileBatchReader::SetReadSchema(::ArrowSchema* read_schema, } reader_ = std::move(reader); array_builder_ = std::move(array_builder); + decode_context_.ClearBuilderMetadata(); previous_first_row_ = std::numeric_limits::max(); previous_batch_row_count_ = 0; next_row_to_read_ = std::numeric_limits::max();