This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new d602c2c5 perf: avoid shared pointer contention in manifest and Avro
decode (#239)
d602c2c5 is described below
commit d602c2c509f2495f6f5d721be1c5de995f4b5fcd
Author: gripleaf <[email protected]>
AuthorDate: Mon Aug 24 14:45:59 2026 +0800
perf: avoid shared pointer contention in manifest and Avro decode (#239)
---
src/paimon/core/manifest/manifest_file.cpp | 3 ++-
src/paimon/core/utils/objects_file.h | 3 ++-
src/paimon/format/avro/avro_direct_decoder.cpp | 28 +++++++++++++++++-----
src/paimon/format/avro/avro_direct_decoder.h | 20 ++++++++++++++++
.../avro/avro_direct_encoder_decoder_test.cpp | 23 ++++++++++++++++++
src/paimon/format/avro/avro_file_batch_reader.cpp | 1 +
6 files changed, 70 insertions(+), 8 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<arrow::StructArray>&
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<T>::Read(const std::string& file_name,
file_name,
[this, &filter, result](const std::shared_ptr<arrow::StructArray>&
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));
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<arrow::DataType> data_type = builder->type();
+ BuilderMetadata metadata{data_type->id(), std::nullopt};
+ if (data_type->id() == arrow::Type::TIMESTAMP) {
+ metadata.timestamp_unit =
+ checked_cast<const arrow::TimestampType*>(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<arrow::Int8Builder*>(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<arrow::Date32Builder*>(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<arrow::TimestampBuilder*>(array_builder);
- auto ts_type =
checked_cast<arrow::TimestampType*>(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 <optional>
#include <set>
+#include <string>
+#include <unordered_map>
+#include <vector>
#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<arrow::TimeUnit::type> 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<uint8_t> bytes_scratch;
+
+ private:
+ std::unordered_map<const arrow::ArrayBuilder*, BuilderMetadata>
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<arrow::Array> 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<uint64_t>::max();
previous_batch_row_count_ = 0;
next_row_to_read_ = std::numeric_limits<uint64_t>::max();
