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 7d992ca0 perf(read): reduce Arrow read-path overhead (#242)
7d992ca0 is described below
commit 7d992ca04af66adcef9db12b9367a951f412cc7e
Author: gripleaf <[email protected]>
AuthorDate: Tue Aug 25 19:17:06 2026 +0800
perf(read): reduce Arrow read-path overhead (#242)
---
src/paimon/common/data/columnar/columnar_row.h | 6 +++++
.../common/data/columnar/columnar_row_test.cpp | 10 ++++++++
src/paimon/core/manifest/manifest_file.cpp | 3 ++-
src/paimon/core/utils/objects_file.h | 3 ++-
src/paimon/format/avro/avro_direct_decoder.cpp | 24 +++++++++++++++++++
src/paimon/format/avro/avro_direct_decoder.h | 6 +++++
.../avro/avro_direct_encoder_decoder_test.cpp | 27 ++++++++++++++++++++++
src/paimon/format/avro/avro_file_batch_reader.cpp | 7 ++++++
src/paimon/format/orc/orc_adapter.cpp | 2 ++
9 files changed, 86 insertions(+), 2 deletions(-)
diff --git a/src/paimon/common/data/columnar/columnar_row.h
b/src/paimon/common/data/columnar/columnar_row.h
index 2156c814..f3d20dba 100644
--- a/src/paimon/common/data/columnar/columnar_row.h
+++ b/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();
}
diff --git a/src/paimon/common/data/columnar/columnar_row_test.cpp
b/src/paimon/common/data/columnar/columnar_row_test.cpp
index 6301cd29..d1a5c600 100644
--- a/src/paimon/common/data/columnar/columnar_row_test.cpp
+++ b/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) {
diff --git a/src/paimon/core/manifest/manifest_file.cpp
b/src/paimon/core/manifest/manifest_file.cpp
index 9f9c8aee..a556b82e 100644
--- a/src/paimon/core/manifest/manifest_file.cpp
+++ b/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;
diff --git a/src/paimon/core/utils/objects_file.h
b/src/paimon/core/utils/objects_file.h
index b3135b31..43d78201 100644
--- a/src/paimon/core/utils/objects_file.h
+++ b/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));
diff --git a/src/paimon/format/avro/avro_direct_decoder.cpp
b/src/paimon/format/avro/avro_direct_decoder.cpp
index f9c8a9a4..9f9e3642 100644
--- a/src/paimon/format/avro/avro_direct_decoder.cpp
+++ b/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
diff --git a/src/paimon/format/avro/avro_direct_decoder.h
b/src/paimon/format/avro/avro_direct_decoder.h
index 6422f915..3b316f54 100644
--- a/src/paimon/format/avro/avro_direct_decoder.h
+++ b/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
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 78f4ca48..8e65520d 100644
--- a/src/paimon/format/avro/avro_direct_encoder_decoder_test.cpp
+++ b/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()),
diff --git a/src/paimon/format/avro/avro_file_batch_reader.cpp
b/src/paimon/format/avro/avro_file_batch_reader.cpp
index 1e217f5c..f8b00c7c 100644
--- a/src/paimon/format/avro/avro_file_batch_reader.cpp
+++ b/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()));
diff --git a/src/paimon/format/orc/orc_adapter.cpp
b/src/paimon/format/orc/orc_adapter.cpp
index 4a03d7cd..dc99c5e6 100644
--- a/src/paimon/format/orc/orc_adapter.cpp
+++ b/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;
}
