perf: avoid shared pointer contention in manifest and Avro decode

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;

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));

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));

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

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
     {

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();