fix: avoid null pointers in zero-length memory copies

build_support/run-test.sh

@@ -84,6 +84,7 @@ function setup_sanitizers() {
 
   UBSAN_OPTIONS="$UBSAN_OPTIONS print_stacktrace=1"
   UBSAN_OPTIONS="$UBSAN_OPTIONS suppressions=$ROOT/build_support/ubsan-suppressions.txt"
+  UBSAN_OPTIONS="$UBSAN_OPTIONS halt_on_error=1"
   export UBSAN_OPTIONS
 
   ASAN_OPTIONS="$ASAN_OPTIONS disable_coredump=0"

include/paimon/memory/memory_segment.h

@@ -166,6 +166,12 @@ class PAIMON_EXPORT MemorySegment {
 
     bool EqualTo(const MemorySegment& seg2, int32_t offset1, int32_t offset2, int32_t length) const;
 
+    /// Returns existing heap memory or copies a non-owning view into heap memory.
+    /// @return The existing Bytes object for an owning segment, including an empty one.
+    /// For a non-owning view with non-null data, returns a new Bytes object;
+    /// WrapView("", 0) returns a non-null, empty Bytes object.
+    /// Returns nullptr when there is no backing data, as with WrapView(nullptr, 0)
+    /// or a default-constructed segment.
     std::shared_ptr<Bytes> GetOrCreateHeapMemory(MemoryPool* pool) const {
         if (heap_memory_) {
             return heap_memory_;
@@ -174,7 +180,9 @@ class PAIMON_EXPORT MemorySegment {
             return nullptr;
         }
         auto copy = std::make_shared<Bytes>(size_, pool);
-        std::memcpy(const_cast<char*>(copy->data()), data_, size_);
+        if (size_ > 0) {
+            std::memcpy(copy->data(), data_, size_);
+        }
         return copy;
     }
 

src/paimon/common/io/byte_array_input_stream.cpp

@@ -70,7 +70,9 @@ Result<int64_t> ByteArrayInputStream::Read(char* buffer, int64_t size) {
                         "position {}, exceed length {}",
                         size, position_, length_));
     }
-    memcpy(buffer, buffer_ + position_, static_cast<size_t>(size));
+    if (size > 0) {
+        memcpy(buffer, buffer_ + position_, static_cast<size_t>(size));
+    }
     position_ += size;
     return size;
 }
@@ -82,7 +84,9 @@ Result<int64_t> ByteArrayInputStream::Read(char* buffer, int64_t size, int64_t o
                         "stream length {}",
                         size, offset, length_));
     }
-    memcpy(buffer, buffer_ + offset, static_cast<size_t>(size));
+    if (size > 0) {
+        memcpy(buffer, buffer_ + offset, static_cast<size_t>(size));
+    }
     return size;
 }
 

src/paimon/common/io/byte_array_input_stream_test.cpp

@@ -19,6 +19,8 @@
 
 #include "paimon/io/byte_array_input_stream.h"
 
+#include <chrono>
+#include <future>
 #include <memory>
 #include <utility>
 
@@ -81,4 +83,54 @@ TEST(ByteArrayInputStreamTest, TestSimple) {
     ASSERT_OK(input_stream->Close());
 }
 
+TEST(ByteArrayInputStreamTest, TestZeroLengthRead) {
+    auto pool = GetDefaultPool();
+    Bytes empty_bytes(0, pool.get());
+    ByteArrayInputStream input_stream("abc", 3);
+    ASSERT_OK(input_stream.Seek(1, SeekOrigin::FS_SEEK_SET));
+
+    ASSERT_OK_AND_ASSIGN(int64_t read_size, input_stream.Read(empty_bytes.data(), 0));
+    ASSERT_EQ(0, read_size);
+    ASSERT_OK_AND_ASSIGN(int64_t position, input_stream.GetPos());
+    ASSERT_EQ(1, position);
+
+    ASSERT_OK_AND_ASSIGN(read_size, input_stream.Read(empty_bytes.data(), 0, /*offset=*/0));
+    ASSERT_EQ(0, read_size);
+    ASSERT_OK_AND_ASSIGN(read_size, input_stream.Read(empty_bytes.data(), 0, /*offset=*/3));
+    ASSERT_EQ(0, read_size);
+    ASSERT_OK_AND_ASSIGN(position, input_stream.GetPos());
+    ASSERT_EQ(1, position);
+
+    auto read_promise = std::make_shared<std::promise<Status>>();
+    std::future<Status> read_result = read_promise->get_future();
+    input_stream.ReadAsync(empty_bytes.data(), 0, /*offset=*/3, [read_promise](Status status) {
+        read_promise->set_value(std::move(status));
+    });
+    ASSERT_EQ(std::future_status::ready, read_result.wait_for(std::chrono::seconds(5)));
+    ASSERT_OK(read_result.get());
+    ASSERT_OK_AND_ASSIGN(position, input_stream.GetPos());
+    ASSERT_EQ(1, position);
+
+    ASSERT_NOK_WITH_MSG(input_stream.Read(empty_bytes.data(), 0, /*offset=*/-1),
+                        "boundary check failed");
+    ASSERT_NOK_WITH_MSG(input_stream.Read(empty_bytes.data(), 0, /*offset=*/4),
+                        "boundary check failed");
+
+    ASSERT_OK(input_stream.Seek(3, SeekOrigin::FS_SEEK_SET));
+    ASSERT_OK_AND_ASSIGN(read_size, input_stream.Read(empty_bytes.data(), 0));
+    ASSERT_EQ(0, read_size);
+    ASSERT_OK_AND_ASSIGN(position, input_stream.GetPos());
+    ASSERT_EQ(3, position);
+}
+
+TEST(ByteArrayInputStreamTest, TestZeroLengthReadFromEmptyStream) {
+    ByteArrayInputStream input_stream("", 0);
+    ASSERT_OK_AND_ASSIGN(int64_t read_size, input_stream.Read(nullptr, 0));
+    ASSERT_EQ(0, read_size);
+    ASSERT_OK_AND_ASSIGN(read_size, input_stream.Read(nullptr, 0, /*offset=*/0));
+    ASSERT_EQ(0, read_size);
+    ASSERT_OK_AND_ASSIGN(int64_t position, input_stream.GetPos());
+    ASSERT_EQ(0, position);
+}
+
 }  // namespace paimon::test

src/paimon/common/memory/memory_segment_test.cpp

@@ -675,6 +675,20 @@ TEST(MemorySegmentTest, TestEqual) {
     ASSERT_FALSE(seg2 == seg1);
 }
 
+TEST(MemorySegmentTest, TestGetOrCreateHeapMemoryForEmptySegment) {
+    auto pool = paimon::GetDefaultPool();
+    auto view = MemorySegment::WrapView("", 0);
+    auto heap = view.GetOrCreateHeapMemory(pool.get());
+    ASSERT_TRUE(heap);
+    ASSERT_EQ(0u, heap->size());
+
+    auto owning = MemorySegment::Wrap(heap);
+    ASSERT_EQ(heap, owning.GetOrCreateHeapMemory(pool.get()));
+
+    auto null_view = MemorySegment::WrapView(nullptr, 0);
+    ASSERT_EQ(nullptr, null_view.GetOrCreateHeapMemory(pool.get()));
+}
+
 TEST(MemorySegmentTest, TestNonOwningWrapView) {
     // Prepare owning data as source
     auto pool = paimon::GetDefaultPool();

src/paimon/common/memory/memory_slice.cpp

@@ -78,7 +78,9 @@ std::string_view MemorySlice::ReadStringView() const {
 
 std::shared_ptr<Bytes> MemorySlice::CopyBytes(MemoryPool* pool) const {
     auto bytes = std::make_shared<Bytes>(length_, pool);
-    std::memcpy(const_cast<char*>(bytes->data()), Data(), length_);
+    if (length_ > 0) {
+        std::memcpy(bytes->data(), Data(), length_);
+    }
     return bytes;
 }
 

src/paimon/common/memory/memory_slice_test.cpp

@@ -275,6 +275,20 @@ TEST_F(MemorySliceTest, TestSliceCopyBytes) {
     ASSERT_NE(bytes->data(), copied->data());
 }
 
+TEST_F(MemorySliceTest, TestSliceCopyEmptyBytes) {
+    auto bytes = std::make_shared<Bytes>(0, pool_.get());
+    MemorySlice slice = MemorySlice::Wrap(bytes);
+    auto copied = slice.CopyBytes(pool_.get());
+    ASSERT_TRUE(copied);
+    ASSERT_EQ(0u, copied->size());
+
+    auto non_empty_bytes = std::make_shared<Bytes>("test", pool_.get());
+    MemorySlice empty_slice(MemorySegment::Wrap(non_empty_bytes), /*offset=*/2, /*length=*/0);
+    copied = empty_slice.CopyBytes(pool_.get());
+    ASSERT_TRUE(copied);
+    ASSERT_EQ(0u, copied->size());
+}
+
 TEST_F(MemorySliceTest, TestSliceGetOrCreateHeapMemory) {
     auto bytes = std::make_shared<Bytes>("test", pool_.get());
     MemorySlice slice = MemorySlice::Wrap(bytes);