perf(read): reduce Arrow read-path overhead (#242)

Author: gripleaf

src/paimon/common/data/columnar/columnar_row.h

@@ -78,6 +78,12 @@ class ColumnarRow : public InternalRow {
         row_kind_ = kind;
     }
 
+    /// Update the row represented by this view without rebuilding its column pointers.
+    /// @param row_id Zero-based row index in the underlying arrays.
+    void SetRowId(int64_t row_id) {
+        row_id_ = row_id;
+    }
+
     int32_t GetFieldCount() const override {
         return array_vec_.size();
     }

src/paimon/common/data/columnar/columnar_row_test.cpp

@@ -70,6 +70,16 @@ TEST(ColumnarRowTest, TestSimple) {
     ASSERT_EQ(row.GetDouble(6), 5.5);
     ASSERT_EQ(row.GetString(7).ToString(), "Hello");
     ASSERT_EQ(std::string(row.GetStringView(7)), "Hello");
+
+    row.SetRowId(3);
+    ASSERT_TRUE(row.GetBoolean(0));
+    ASSERT_EQ(row.GetByte(1), 3);
+    ASSERT_EQ(row.GetShort(2), 7);
+    ASSERT_EQ(row.GetInt(3), 13);
+    ASSERT_EQ(row.GetLong(4), 18);
+    ASSERT_EQ(row.GetFloat(5), 3.3f);
+    ASSERT_EQ(row.GetDouble(6), 8.8);
+    ASSERT_EQ(std::string(row.GetStringView(7)), "WORLD");
 }
 
 TEST(ColumnarRowRefTest, TestSimple) {

src/paimon/core/manifest/manifest_file.cpp

@@ -93,8 +93,9 @@ Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t buc
         file_name,
         [this, bucket, entries](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
             const arrow::ArrayVector& fields = batch->fields();
+            ColumnarRow row(fields, pool_, /*row_id=*/0);
             for (int64_t i = 0; i < batch->length(); i++) {
-                ColumnarRow row(fields, pool_, i);
+                row.SetRowId(i);
                 PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(row.GetInt(0)));
                 if (ManifestEntrySerializer::GetBucket(row) != bucket) {
                     continue;

src/paimon/core/utils/objects_file.h

@@ -135,8 +135,9 @@ Status ObjectsFile<T>::Read(const std::string& 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();
+            ColumnarRow row(fields, pool_, /*row_id=*/0);
             for (int64_t i = 0; i < struct_array->length(); i++) {
-                ColumnarRow row(fields, pool_, i);
+                row.SetRowId(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

@@ -57,6 +57,20 @@ Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node,
                             ::avro::Decoder* decoder, arrow::ArrayBuilder* array_builder,
                             AvroDirectDecoder::DecodeContext* ctx);
 
+Status ReserveBuilderCapacityImpl(int64_t capacity, arrow::ArrayBuilder* array_builder) {
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(array_builder->Reserve(capacity));
+    if (array_builder->type()->id() != arrow::Type::STRUCT) {
+        return Status::OK();
+    }
+
+    auto* struct_builder = checked_cast<arrow::StructBuilder*>(array_builder);
+    for (int32_t i = 0; i < struct_builder->num_fields(); ++i) {
+        PAIMON_RETURN_NOT_OK(
+            ReserveBuilderCapacityImpl(capacity, struct_builder->field_builder(i)));
+    }
+    return Status::OK();
+}
+
 /// \brief Skip an Avro value based on its schema without decoding
 Status SkipAvroValue(const ::avro::NodePtr& avro_node, ::avro::Decoder* decoder) {
     switch (avro_node->type()) {
@@ -193,6 +207,7 @@ Status DecodeListToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder* de
     // Read array block count
     int64_t block_count = decoder->arrayStart();
     while (block_count != 0) {
+        PAIMON_RETURN_NOT_OK(ReserveBuilderCapacityImpl(block_count, value_builder));
         for (int64_t i = 0; i < block_count; ++i) {
             PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(element_node, /*projection=*/std::nullopt,
                                                       decoder, value_builder, ctx));
@@ -221,6 +236,8 @@ Status DecodeMapToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder* dec
         // Read map block count
         int64_t block_count = decoder->mapStart();
         while (block_count != 0) {
+            PAIMON_RETURN_NOT_OK(ReserveBuilderCapacityImpl(block_count, key_builder));
+            PAIMON_RETURN_NOT_OK(ReserveBuilderCapacityImpl(block_count, item_builder));
             for (int64_t i = 0; i < block_count; ++i) {
                 PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(key_node, /*projection=*/std::nullopt,
                                                           decoder, key_builder, ctx));
@@ -248,6 +265,8 @@ Status DecodeMapToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder* dec
         // Read array block count
         int64_t block_count = decoder->arrayStart();
         while (block_count != 0) {
+            PAIMON_RETURN_NOT_OK(ReserveBuilderCapacityImpl(block_count, key_builder));
+            PAIMON_RETURN_NOT_OK(ReserveBuilderCapacityImpl(block_count, item_builder));
             for (int64_t i = 0; i < block_count; ++i) {
                 PAIMON_RETURN_NOT_OK(DecodeFieldToBuilder(key_node, /*projection=*/std::nullopt,
                                                           decoder, key_builder, ctx));
@@ -447,4 +466,9 @@ Status AvroDirectDecoder::DecodeAvroToBuilder(const ::avro::NodePtr& avro_node,
     return DecodeFieldToBuilder(avro_node, projection, decoder, array_builder, ctx);
 }
 
+Status AvroDirectDecoder::ReserveBuilderCapacity(int64_t capacity,
+                                                 arrow::ArrayBuilder* array_builder) {
+    return ReserveBuilderCapacityImpl(capacity, array_builder);
+}
+
 }  // namespace paimon::avro

src/paimon/format/avro/avro_direct_decoder.h

@@ -81,6 +81,12 @@ class AvroDirectDecoder {
                                       const std::optional<std::set<size_t>>& projection,
                                       ::avro::Decoder* decoder, arrow::ArrayBuilder* array_builder,
                                       DecodeContext* ctx);
+
+    /// Reserve slots for a builder and any struct children with the same cardinality.
+    /// @param capacity Number of additional values to append.
+    /// @param array_builder Builder to reserve.
+    /// @return Status::OK if all reservations succeed.
+    static Status ReserveBuilderCapacity(int64_t capacity, arrow::ArrayBuilder* array_builder);
 };
 
 }  // namespace paimon::avro

src/paimon/format/avro/avro_direct_encoder_decoder_test.cpp

@@ -361,6 +361,33 @@ TEST_F(AvroDirectEncoderDecoderTest, TestRecordType) {
     CheckResult(schema_json, input_array, &struct_builder);
 }
 
+TEST_F(AvroDirectEncoderDecoderTest, TestReserveBuilderCapacity) {
+    std::shared_ptr<arrow::DataType> nested_type = arrow::struct_({
+        arrow::field("id", arrow::int32()),
+        arrow::field("nested", arrow::struct_({arrow::field("value", arrow::int64())})),
+        arrow::field("items", arrow::list(arrow::int32())),
+    });
+    arrow::Result<std::unique_ptr<arrow::ArrayBuilder>> builder_result =
+        arrow::MakeBuilder(nested_type);
+    ASSERT_TRUE(builder_result.ok()) << builder_result.status().ToString();
+    std::unique_ptr<arrow::ArrayBuilder> builder = std::move(builder_result).ValueOrDie();
+
+    constexpr int64_t capacity = 1024;
+    ASSERT_OK(AvroDirectDecoder::ReserveBuilderCapacity(capacity, builder.get()));
+
+    auto* root_builder = checked_cast<arrow::StructBuilder*>(builder.get());
+    ASSERT_GE(root_builder->capacity(), capacity);
+    ASSERT_GE(root_builder->field_builder(0)->capacity(), capacity);
+
+    auto* nested_builder = checked_cast<arrow::StructBuilder*>(root_builder->field_builder(1));
+    ASSERT_GE(nested_builder->capacity(), capacity);
+    ASSERT_GE(nested_builder->field_builder(0)->capacity(), capacity);
+
+    auto* list_builder = checked_cast<arrow::ListBuilder*>(root_builder->field_builder(2));
+    ASSERT_GE(list_builder->capacity(), capacity);
+    ASSERT_EQ(list_builder->value_builder()->capacity(), 0);
+}
+
 TEST_F(AvroDirectEncoderDecoderTest, TestDecodeWithProjection) {
     arrow::FieldVector fields = {
         arrow::field("f0", arrow::boolean()),

src/paimon/format/avro/avro_file_batch_reader.cpp

@@ -110,6 +110,10 @@ Result<BatchReader::ReadBatch> AvroFileBatchReader::NextBatch() {
             if (!reader_->hasMore()) {
                 break;
             }
+            if (array_builder_->length() == 0) {
+                PAIMON_RETURN_NOT_OK(
+                    AvroDirectDecoder::ReserveBuilderCapacity(batch_size_, array_builder_.get()));
+            }
             reader_->decr();
             PAIMON_RETURN_NOT_OK(AvroDirectDecoder::DecodeAvroToBuilder(
                 reader_->dataSchema().root(), read_fields_projection_, &reader_->decoder(),
@@ -123,7 +127,10 @@ Result<BatchReader::ReadBatch> AvroFileBatchReader::NextBatch() {
         }
         PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
                                           array_builder_->Finish());
+#ifndef NDEBUG
+        // Keep structural validation in debug builds without adding its recursive cost to reads.
         PAIMON_RETURN_NOT_OK_FROM_ARROW(array->Validate());
+#endif
         std::unique_ptr<ArrowArray> c_array = std::make_unique<ArrowArray>();
         std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
         PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, c_array.get(), c_schema.get()));

src/paimon/format/orc/orc_adapter.cpp

@@ -940,6 +940,8 @@ Result<std::shared_ptr<arrow::Array>> OrcAdapter::AppendBatch(
                            MakeArrowBuilder(type, batch, pool));
     std::shared_ptr<arrow::Array> array;
     PAIMON_RETURN_NOT_OK_FROM_ARROW(builder->Finish(&array));
+    // Keep this check in release builds so malformed nested arrays return a Status before they
+    // reach Arrow constructors that enforce their invariants with a process-terminating check.
     PAIMON_RETURN_NOT_OK_FROM_ARROW(array->Validate());
     return array;
 }