refactor: unify BinarySection classes to single MemorySegment model and use string_view to avoid copies (alibaba#196)

Author: lxy-9602

include/paimon/data/blob.h

@@ -94,12 +94,10 @@ class PAIMON_EXPORT Blob {
     /// It automatically injects Paimon-specific metadata to identify the field as a BLOB.
     ///
     /// @param field_name The name of the Arrow field.
-    /// @param nullable Whether the field can contain null values (defaults to false).
     /// @param metadata A map of key-value metadata to be attached to the field.
     /// @return A result containing a unique pointer to the generated `ArrowSchema` or an error.
     static Result<std::unique_ptr<::ArrowSchema>> ArrowField(
-        const std::string& field_name, bool nullable = false,
-        std::unordered_map<std::string, std::string> metadata = {});
+        const std::string& field_name, std::unordered_map<std::string, std::string> metadata = {});
 
  private:
     class Impl;

src/paimon/common/data/abstract_binary_writer.cpp

@@ -16,7 +16,6 @@
 
 #include "paimon/common/data/abstract_binary_writer.h"
 
-#include <algorithm>
 #include <cassert>
 #include <memory>
 #include <optional>
@@ -48,11 +47,11 @@ void AbstractBinaryWriter::WriteString(int32_t pos, const BinaryString& input) {
     int32_t len = input.GetSizeInBytes();
     if (len <= BinarySection::MAX_FIX_PART_DATA_SIZE) {
         auto bytes = Bytes::AllocateBytes(len, pool_);
-        MemorySegmentUtils::CopyToBytes(input.GetSegments(), input.GetOffset(), bytes.get(), 0,
+        MemorySegmentUtils::CopyToBytes({input.GetSegment()}, input.GetOffset(), bytes.get(), 0,
                                         len);
         WriteBytesToFixLenPart(&segment_, GetFieldOffset(pos), *bytes, len);
     } else {
-        WriteSegmentsToVarLenPart(pos, input.GetSegments(), input.GetOffset(), len);
+        WriteSegmentToVarLenPart(pos, input.GetSegment(), input.GetOffset(), len);
     }
 }
 
@@ -75,20 +74,19 @@ void AbstractBinaryWriter::WriteStringView(int32_t pos, const std::string_view&
 }
 
 void AbstractBinaryWriter::WriteRow(int32_t pos, const BinaryRow& input) {
-    return WriteSegmentsToVarLenPart(pos, input.GetSegments(), input.GetOffset(),
-                                     input.GetSizeInBytes());
+    return WriteSegmentToVarLenPart(pos, input.GetSegment(), input.GetOffset(),
+                                    input.GetSizeInBytes());
 }
 
 void AbstractBinaryWriter::WriteArray(int32_t pos, const BinaryArray& input) {
-    return WriteSegmentsToVarLenPart(pos, input.GetSegments(), input.GetOffset(),
-                                     input.GetSizeInBytes());
+    return WriteSegmentToVarLenPart(pos, input.GetSegment(), input.GetOffset(),
+                                    input.GetSizeInBytes());
 }
 
 void AbstractBinaryWriter::WriteMap(int32_t pos, const BinaryMap& input) {
-    return WriteSegmentsToVarLenPart(pos, input.GetSegments(), input.GetOffset(),
-                                     input.GetSizeInBytes());
+    return WriteSegmentToVarLenPart(pos, input.GetSegment(), input.GetOffset(),
+                                    input.GetSizeInBytes());
 }
-
 void AbstractBinaryWriter::WriteDecimal(int32_t pos, const std::optional<Decimal>& value,
                                         int32_t precision) {
     assert(value == std::nullopt || precision == value.value().Precision());
@@ -152,44 +150,19 @@ void AbstractBinaryWriter::EnsureCapacity(int32_t needed_size) {
     }
 }
 
-void AbstractBinaryWriter::WriteSegmentsToVarLenPart(int32_t pos,
-                                                     const std::vector<MemorySegment>& segments,
-                                                     int32_t offset, int32_t size) {
+void AbstractBinaryWriter::WriteSegmentToVarLenPart(int32_t pos, const MemorySegment& segment,
+                                                    int32_t offset, int32_t size) {
     const int32_t rounded_size = RoundNumberOfBytesToNearestWord(size);
     // grow the global buffer before writing data.
     EnsureCapacity(rounded_size);
     ZeroOutPaddingBytes(size);
 
-    if (segments.size() == 1) {
-        segments[0].CopyTo(offset, &segment_, cursor_, size);
-    } else {
-        WriteMultiSegmentsToVarLenPart(segments, offset, size);
-    }
+    segment.CopyTo(offset, &segment_, cursor_, size);
     SetOffsetAndSize(pos, cursor_, size);
     // move the cursor forward.
     cursor_ += rounded_size;
 }
 
-void AbstractBinaryWriter::WriteMultiSegmentsToVarLenPart(
-    const std::vector<MemorySegment>& segments, int32_t offset, int32_t size) {
-    // Write the bytes to the variable length portion.
-    int32_t need_copy = size;
-    int32_t from_offset = offset;
-    int32_t to_offset = cursor_;
-    for (const auto& source_segment : segments) {
-        int32_t remain = source_segment.Size() - from_offset;
-        if (remain > 0) {
-            int32_t copy_size = std::min(remain, need_copy);
-            source_segment.CopyTo(from_offset, &segment_, to_offset, copy_size);
-            need_copy -= copy_size;
-            to_offset += copy_size;
-            from_offset = 0;
-        } else {
-            from_offset -= source_segment.Size();
-        }
-    }
-}
-
 template <typename T>
 void AbstractBinaryWriter::WriteBytesToVarLenPart(int32_t pos, const T& bytes, int32_t len) {
     const int32_t rounded_size = RoundNumberOfBytesToNearestWord(len);

src/paimon/common/data/abstract_binary_writer.h

@@ -19,7 +19,6 @@
 #include <cstdint>
 #include <optional>
 #include <string_view>
-#include <vector>
 
 #include "paimon/common/data/binary_writer.h"
 #include "paimon/common/memory/memory_segment.h"
@@ -56,7 +55,7 @@ class AbstractBinaryWriter : public BinaryWriter {
 
     void WriteStringView(int32_t pos, const std::string_view& view) override;
 
-    const MemorySegment& GetSegments() const {
+    const MemorySegment& GetSegment() const {
         return segment_;
     }
 
@@ -90,10 +89,8 @@ class AbstractBinaryWriter : public BinaryWriter {
     static void WriteBytesToFixLenPart(MemorySegment* segment, int32_t field_offset, const T& bytes,
                                        int32_t len);
 
-    void WriteSegmentsToVarLenPart(int32_t pos, const std::vector<MemorySegment>& segments,
-                                   int32_t offset, int32_t size);
-    void WriteMultiSegmentsToVarLenPart(const std::vector<MemorySegment>& segments, int32_t offset,
-                                        int32_t size);
+    void WriteSegmentToVarLenPart(int32_t pos, const MemorySegment& segment, int32_t offset,
+                                  int32_t size);
 
  protected:
     int32_t cursor_ = 0;

src/paimon/common/data/binary_array.cpp

@@ -40,36 +40,28 @@ int32_t BinaryArray::GetElementOffset(int32_t ordinal, int32_t element_size) con
 }
 
 void BinaryArray::PointTo(const MemorySegment& segment, int32_t offset, int32_t size_in_bytes) {
-    std::vector<MemorySegment> segments = {segment};
-    PointTo(segments, offset, size_in_bytes);
-}
-
-void BinaryArray::PointTo(const std::vector<MemorySegment>& segments, int32_t offset,
-                          int32_t size_in_bytes) {
-    // Read the number of elements from the first 4 bytes.
-    auto size = MemorySegmentUtils::GetValue<int32_t>(segments, offset);
+    auto size = MemorySegmentUtils::GetValue<int32_t>({segment}, offset);
     assert(size >= 0);
-
     size_ = size;
-    segments_ = segments;
+    segment_ = segment;
     offset_ = offset;
     size_in_bytes_ = size_in_bytes;
     element_offset_ = offset_ + CalculateHeaderInBytes(size_);
 }
 
 bool BinaryArray::IsNullAt(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::BitGet(segments_, offset_ + 4, pos);
+    return MemorySegmentUtils::BitGet({segment_}, offset_ + 4, pos);
 }
 
 int64_t BinaryArray::GetLong(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::GetValue<int64_t>(segments_, GetElementOffset(pos, 8));
+    return MemorySegmentUtils::GetValue<int64_t>({segment_}, GetElementOffset(pos, 8));
 }
 
 int32_t BinaryArray::GetInt(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::GetValue<int32_t>(segments_, GetElementOffset(pos, 4));
+    return MemorySegmentUtils::GetValue<int32_t>({segment_}, GetElementOffset(pos, 4));
 }
 int32_t BinaryArray::GetDate(int32_t pos) const {
     return GetInt(pos);
@@ -78,90 +70,96 @@ int32_t BinaryArray::GetDate(int32_t pos) const {
 BinaryString BinaryArray::GetString(int32_t pos) const {
     AssertIndexIsValid(pos);
     int32_t field_offset = GetElementOffset(pos, 8);
-    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>(segments_, field_offset);
-    return BinaryDataReadUtils::ReadBinaryString(segments_, offset_, field_offset, offset_and_size);
+    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>({segment_}, field_offset);
+    return BinaryDataReadUtils::ReadBinaryString(segment_, offset_, field_offset, offset_and_size);
+}
+
+std::string_view BinaryArray::GetStringView(int32_t pos) const {
+    BinaryString binary_string = GetString(pos);
+    return binary_string.GetStringView();
 }
 
 Decimal BinaryArray::GetDecimal(int32_t pos, int32_t precision, int32_t scale) const {
     AssertIndexIsValid(pos);
     if (Decimal::IsCompact(precision)) {
         return Decimal::FromUnscaledLong(
-            MemorySegmentUtils::GetValue<int64_t>(segments_, GetElementOffset(pos, 8)), precision,
+            MemorySegmentUtils::GetValue<int64_t>({segment_}, GetElementOffset(pos, 8)), precision,
             scale);
     }
 
     int32_t field_offset = GetElementOffset(pos, 8);
-    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>(segments_, field_offset);
-    return BinaryDataReadUtils::ReadDecimal(segments_, offset_, offset_and_size, precision, scale);
+    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>({segment_}, field_offset);
+    return BinaryDataReadUtils::ReadDecimal(segment_, offset_, offset_and_size, precision, scale);
 }
 
 Timestamp BinaryArray::GetTimestamp(int32_t pos, int32_t precision) const {
     AssertIndexIsValid(pos);
 
     if (Timestamp::IsCompact(precision)) {
         return Timestamp::FromEpochMillis(
-            MemorySegmentUtils::GetValue<int64_t>(segments_, GetElementOffset(pos, 8)));
+            MemorySegmentUtils::GetValue<int64_t>({segment_}, GetElementOffset(pos, 8)));
     }
 
     int32_t field_offset = GetElementOffset(pos, 8);
     const auto offset_and_nano_of_milli =
-        MemorySegmentUtils::GetValue<int64_t>(segments_, field_offset);
-    return BinaryDataReadUtils::ReadTimestampData(segments_, offset_, offset_and_nano_of_milli);
+        MemorySegmentUtils::GetValue<int64_t>({segment_}, field_offset);
+    return BinaryDataReadUtils::ReadTimestampData(segment_, offset_, offset_and_nano_of_milli);
 }
 
 std::shared_ptr<Bytes> BinaryArray::GetBinary(int32_t pos) const {
     AssertIndexIsValid(pos);
     int32_t field_offset = GetElementOffset(pos, 8);
-    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>(segments_, field_offset);
-    return BinarySection::ReadBinary(segments_, offset_, field_offset, offset_and_size,
+    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>({segment_}, field_offset);
+    return BinarySection::ReadBinary(segment_, offset_, field_offset, offset_and_size,
                                      GetDefaultPool().get());
 }
 
 std::shared_ptr<InternalArray> BinaryArray::GetArray(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return BinaryDataReadUtils::ReadArrayData(segments_, offset_, GetLong(pos));
+    return BinaryDataReadUtils::ReadArrayData(segment_, offset_, GetLong(pos));
 }
 
 std::shared_ptr<InternalMap> BinaryArray::GetMap(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return BinaryDataReadUtils::ReadMapData(segments_, offset_, GetLong(pos));
+    return BinaryDataReadUtils::ReadMapData(segment_, offset_, GetLong(pos));
 }
 
 std::shared_ptr<InternalRow> BinaryArray::GetRow(int32_t pos, int32_t num_fields) const {
     AssertIndexIsValid(pos);
     int32_t field_offset = GetElementOffset(pos, 8);
-    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>(segments_, field_offset);
-    return BinaryDataReadUtils::ReadRowData(segments_, num_fields, offset_, offset_and_size);
+    const auto offset_and_size = MemorySegmentUtils::GetValue<int64_t>({segment_}, field_offset);
+    return BinaryDataReadUtils::ReadRowData(segment_, num_fields, offset_, offset_and_size);
 }
 
 bool BinaryArray::GetBoolean(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::GetValue<bool>(segments_, GetElementOffset(pos, 1));
+    return MemorySegmentUtils::GetValue<bool>({segment_}, GetElementOffset(pos, 1));
 }
 
 char BinaryArray::GetByte(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::GetValue<char>(segments_, GetElementOffset(pos, 1));
+    return MemorySegmentUtils::GetValue<char>({segment_}, GetElementOffset(pos, 1));
 }
 
 int16_t BinaryArray::GetShort(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::GetValue<int16_t>(segments_, GetElementOffset(pos, sizeof(int16_t)));
+    return MemorySegmentUtils::GetValue<int16_t>({segment_},
+                                                 GetElementOffset(pos, sizeof(int16_t)));
 }
 
 float BinaryArray::GetFloat(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::GetValue<float>(segments_, GetElementOffset(pos, sizeof(float)));
+    return MemorySegmentUtils::GetValue<float>({segment_}, GetElementOffset(pos, sizeof(float)));
 }
 
 double BinaryArray::GetDouble(int32_t pos) const {
     AssertIndexIsValid(pos);
-    return MemorySegmentUtils::GetValue<double>(segments_, GetElementOffset(pos, sizeof(double)));
+    return MemorySegmentUtils::GetValue<double>({segment_}, GetElementOffset(pos, sizeof(double)));
 }
 
 bool BinaryArray::AnyNull() const {
     for (int32_t i = offset_ + 4; i < element_offset_; i += 4) {
-        if (MemorySegmentUtils::GetValue<int32_t>(segments_, i) != 0) {
+        if (MemorySegmentUtils::GetValue<int32_t>({segment_}, i) != 0) {
             return true;
         }
     }
@@ -172,7 +170,7 @@ Result<std::vector<char>> BinaryArray::ToBooleanArray() const {
     PAIMON_RETURN_NOT_OK(CheckNoNull());
     std::vector<char> values;
     values.resize(size_);
-    MemorySegmentUtils::CopyToUnsafe(segments_, element_offset_,
+    MemorySegmentUtils::CopyToUnsafe({segment_}, element_offset_,
                                      const_cast<void*>(static_cast<const void*>(values.data())),
                                      size_);
     return values;
@@ -182,7 +180,7 @@ Result<std::vector<char>> BinaryArray::ToByteArray() const {
     PAIMON_RETURN_NOT_OK(CheckNoNull());
     std::vector<char> values;
     values.resize(size_);
-    MemorySegmentUtils::CopyToUnsafe(segments_, element_offset_,
+    MemorySegmentUtils::CopyToUnsafe({segment_}, element_offset_,
                                      const_cast<void*>(static_cast<const void*>(values.data())),
                                      size_);
     return values;
@@ -192,7 +190,7 @@ Result<std::vector<int16_t>> BinaryArray::ToShortArray() const {
     PAIMON_RETURN_NOT_OK(CheckNoNull());
     std::vector<int16_t> values;
     values.resize(size_);
-    MemorySegmentUtils::CopyToUnsafe(segments_, element_offset_,
+    MemorySegmentUtils::CopyToUnsafe({segment_}, element_offset_,
                                      const_cast<void*>(static_cast<const void*>(values.data())),
                                      size_ * sizeof(int16_t));
     return values;
@@ -202,7 +200,7 @@ Result<std::vector<int32_t>> BinaryArray::ToIntArray() const {
     PAIMON_RETURN_NOT_OK(CheckNoNull());
     std::vector<int32_t> values;
     values.resize(size_);
-    MemorySegmentUtils::CopyToUnsafe(segments_, element_offset_,
+    MemorySegmentUtils::CopyToUnsafe({segment_}, element_offset_,
                                      const_cast<void*>(static_cast<const void*>(values.data())),
                                      size_ * sizeof(int32_t));
     return values;
@@ -212,7 +210,7 @@ Result<std::vector<int64_t>> BinaryArray::ToLongArray() const {
     PAIMON_RETURN_NOT_OK(CheckNoNull());
     std::vector<int64_t> values;
     values.resize(size_);
-    MemorySegmentUtils::CopyToUnsafe(segments_, element_offset_,
+    MemorySegmentUtils::CopyToUnsafe({segment_}, element_offset_,
                                      const_cast<void*>(static_cast<const void*>(values.data())),
                                      size_ * sizeof(int64_t));
     return values;
@@ -222,7 +220,7 @@ Result<std::vector<float>> BinaryArray::ToFloatArray() const {
     PAIMON_RETURN_NOT_OK(CheckNoNull());
     std::vector<float> values;
     values.resize(size_);
-    MemorySegmentUtils::CopyToUnsafe(segments_, element_offset_,
+    MemorySegmentUtils::CopyToUnsafe({segment_}, element_offset_,
                                      const_cast<void*>(static_cast<const void*>(values.data())),
                                      size_ * sizeof(float));
     return values;
@@ -232,7 +230,7 @@ Result<std::vector<double>> BinaryArray::ToDoubleArray() const {
     PAIMON_RETURN_NOT_OK(CheckNoNull());
     std::vector<double> values;
     values.resize(size_);
-    MemorySegmentUtils::CopyToUnsafe(segments_, element_offset_,
+    MemorySegmentUtils::CopyToUnsafe({segment_}, element_offset_,
                                      const_cast<void*>(static_cast<const void*>(values.data())),
                                      size_ * sizeof(double));
     return values;
@@ -246,7 +244,7 @@ BinaryArray BinaryArray::Copy(MemoryPool* pool) const {
 
 void BinaryArray::Copy(BinaryArray* reuse, MemoryPool* pool) const {
     std::shared_ptr<Bytes> bytes =
-        MemorySegmentUtils::CopyToBytes(segments_, offset_, size_in_bytes_, pool);
+        MemorySegmentUtils::CopyToBytes({segment_}, offset_, size_in_bytes_, pool);
     reuse->PointTo(MemorySegment::Wrap(bytes), 0, size_in_bytes_);
 }