fix: Fix MemorySegment::Compare to use big-endian byte-order comparison semantics (alibaba#298)

Author: lxy-9602

src/paimon/common/memory/memory_segment.cpp

@@ -23,11 +23,8 @@ namespace paimon {
 int32_t MemorySegment::Compare(const MemorySegment& seg2, int32_t offset1, int32_t offset2,
                                int32_t len) const {
     while (len >= 8) {
-        // TODO(yonghao.fyh): support big endian, decide SmallEndian or BigEndian
-        // long l1 = GetLongBigEndian(offset1);
-        // long l2 = seg2.getLongBigEndian(offset2);
-        uint64_t l1 = GetValue<int64_t>(offset1);
-        uint64_t l2 = seg2.GetValue<int64_t>(offset2);
+        uint64_t l1 = GetLongBigEndian(offset1);
+        uint64_t l2 = seg2.GetLongBigEndian(offset2);
 
         if (l1 != l2) {
             return (l1 < l2) ? -1 : 1;

src/paimon/common/memory/memory_segment.h

@@ -22,6 +22,8 @@
 #include <memory>
 #include <type_traits>
 
+#include "paimon/common/utils/math.h"
+#include "paimon/io/byte_order.h"
 #include "paimon/memory/bytes.h"
 #include "paimon/visibility.h"
 
@@ -132,6 +134,14 @@ class PAIMON_EXPORT MemorySegment {
         std::memcpy(MutableData() + index, &value, sizeof(T));
     }
 
+    inline uint64_t GetLongBigEndian(int32_t index) const {
+        auto value = GetValue<uint64_t>(index);
+        if constexpr (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
+            return EndianSwapValue(value);
+        }
+        return value;
+    }
+
     void CopyTo(int32_t offset, MemorySegment* target, int32_t target_offset,
                 int32_t num_bytes) const {
         assert(offset >= 0);

src/paimon/common/memory/memory_segment_test.cpp

@@ -158,6 +158,32 @@ TEST(MemorySegmentTest, TestCompare) {
     seg2.Put(i + 8, static_cast<char>(10));
     ASSERT_EQ(seg1.Compare(seg2, i, i, 7, 7), 0);
     ASSERT_LT(seg1.Compare(seg2, i, i, 9, 9), 0);
+
+    // Verify big-endian byte-order comparison semantics within a single 8-byte block.
+    // On little-endian machines, naive native-endian uint64 comparison would give wrong results.
+    MemorySegment seg3 = MemorySegment::AllocateHeapMemory(16, pool.get());
+    MemorySegment seg4 = MemorySegment::AllocateHeapMemory(16, pool.get());
+    Bytes zeros(16, pool.get());
+    seg3.Put(0, zeros);
+    seg4.Put(0, zeros);
+
+    // seg3: [0x00, 0x01, 0, 0, 0, 0, 0, 0] at offset 0
+    // seg4: [0x01, 0x00, 0, 0, 0, 0, 0, 0] at offset 0
+    // Lexicographic (byte-order) comparison: first byte 0x00 < 0x01, so seg3 < seg4.
+    seg3.Put(1, static_cast<char>(0x01));
+    seg4.Put(0, static_cast<char>(0x01));
+    ASSERT_LT(seg3.Compare(seg4, 0, 0, 8), 0);
+    ASSERT_GT(seg4.Compare(seg3, 0, 0, 8), 0);
+
+    // seg3: [0x01, 0x02, 0, 0, 0, 0, 0, 0]
+    // seg4: [0x01, 0x01, 0, 0, 0, 0, 0, 0]
+    // First bytes equal (0x01 == 0x01), second byte 0x02 > 0x01, so seg3 > seg4.
+    seg3.Put(0, static_cast<char>(0x01));
+    seg3.Put(1, static_cast<char>(0x02));
+    seg4.Put(0, static_cast<char>(0x01));
+    seg4.Put(1, static_cast<char>(0x01));
+    ASSERT_GT(seg3.Compare(seg4, 0, 0, 8), 0);
+    ASSERT_LT(seg4.Compare(seg3, 0, 0, 8), 0);
 }
 
 TEST(MemorySegmentTest, TestCharAccess) {