fix: fix undefined behavior in bitwise left-shift operations across codebase

src/paimon/common/compression/block_decompressor.cpp

@@ -20,8 +20,10 @@
 namespace paimon {
 
 int32_t BlockDecompressor::ReadIntLE(const char* buf) {
-    return (buf[0] & 0xFF) | ((buf[1] & 0xFF) << 8) | ((buf[2] & 0xFF) << 16) |
-           ((buf[3] & 0xFF) << 24);
+    return static_cast<int32_t>(static_cast<uint32_t>(static_cast<uint8_t>(buf[0])) |
+                                (static_cast<uint32_t>(static_cast<uint8_t>(buf[1])) << 8) |
+                                (static_cast<uint32_t>(static_cast<uint8_t>(buf[2])) << 16) |
+                                (static_cast<uint32_t>(static_cast<uint8_t>(buf[3])) << 24));
 }
 
 Status BlockDecompressor::ValidateLength(int32_t compressed_len, int32_t original_len) {

src/paimon/common/data/abstract_binary_writer.cpp

@@ -199,20 +199,18 @@ int32_t AbstractBinaryWriter::RoundNumberOfBytesToNearestWord(int32_t num_bytes)
 template <typename T>
 void AbstractBinaryWriter::WriteBytesToFixLenPart(MemorySegment* segment, int32_t field_offset,
                                                   const T& bytes, int32_t len) {
-    int64_t first_byte = len | 0x80;  // first bit is 1, other bits is len
-    int64_t seven_bytes = 0L;         // real data
+    uint64_t first_byte = static_cast<uint64_t>(len) | 0x80u;  // first bit is 1, other bits is len
+    uint64_t seven_bytes = 0;                                  // real data
     if ((SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN)) {
         for (int32_t i = 0; i < len; i++) {
-            seven_bytes |= ((0x00000000000000FFL & bytes[i]) << (i * 8L));
+            seven_bytes |= (static_cast<uint64_t>(static_cast<uint8_t>(bytes[i])) << (i * 8));
         }
     } else {
         for (int32_t i = 0; i < len; i++) {
-            seven_bytes |= ((0x00000000000000FFL & bytes[i]) << ((6 - i) * 8L));
+            seven_bytes |= (static_cast<uint64_t>(static_cast<uint8_t>(bytes[i])) << ((6 - i) * 8));
         }
     }
-    const int64_t offset_and_size =
-        (first_byte << 56) |  // NOLINT(clang-analyzer-core.UndefinedBinaryOperatorResult)
-        seven_bytes;
+    const auto offset_and_size = static_cast<int64_t>(first_byte << 56 | seven_bytes);
     segment->PutValue<int64_t>(field_offset, offset_and_size);
 }
 

src/paimon/common/data/binary_array_writer.cpp

@@ -95,7 +95,8 @@ int32_t BinaryArrayWriter::GetFieldOffset(int32_t pos) const {
 }
 
 void BinaryArrayWriter::SetOffsetAndSize(int32_t pos, int32_t offset, int64_t size) {
-    const int64_t offset_and_size = (static_cast<int64_t>(offset) << 32) | size;
+    const auto offset_and_size =
+        static_cast<int64_t>((static_cast<uint64_t>(offset) << 32) | static_cast<uint64_t>(size));
     segment_.PutValue<int64_t>(GetElementOffset(pos, 8), offset_and_size);
 }
 

src/paimon/common/data/binary_row.cpp

@@ -28,7 +28,9 @@
 
 namespace paimon {
 const int64_t BinaryRow::FIRST_BYTE_ZERO =
-    (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) ? (~0xFFL) : (~(0xFFL << 56L));
+    (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN)
+        ? static_cast<int64_t>(~static_cast<uint64_t>(0xFF))
+        : static_cast<int64_t>(~(static_cast<uint64_t>(0xFF) << 56));
 
 const BinaryRow& BinaryRow::EmptyRow() {
     static const BinaryRow empty_row = GetEmptyRow();

src/paimon/common/data/binary_row_writer.h

@@ -100,7 +100,8 @@ class BinaryRowWriter : public AbstractBinaryWriter {
     }
 
     void SetOffsetAndSize(int32_t pos, int32_t offset, int64_t size) override {
-        const int64_t offset_and_size = (static_cast<int64_t>(offset) << 32) | size;
+        const auto offset_and_size = static_cast<int64_t>((static_cast<uint64_t>(offset) << 32) |
+                                                          static_cast<uint64_t>(size));
         segment_.PutValue<int64_t>(GetFieldOffset(pos), offset_and_size);
     }
 

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

@@ -46,8 +46,11 @@ Decimal ColumnarArray::GetDecimal(int32_t pos, int32_t precision, int32_t scale)
     auto array = arrow::internal::checked_cast<const ArrayType*>(array_);
     assert(array);
     arrow::Decimal128 decimal(array->GetValue(offset_ + pos));
-    return Decimal(precision, scale,
-                   static_cast<Decimal::int128_t>(decimal.high_bits()) << 64 | decimal.low_bits());
+    return Decimal(
+        precision, scale,
+        static_cast<Decimal::int128_t>(
+            static_cast<Decimal::uint128_t>(static_cast<uint64_t>(decimal.high_bits())) << 64 |
+            decimal.low_bits()));
 }
 
 Timestamp ColumnarArray::GetTimestamp(int32_t pos, int32_t precision) const {

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

@@ -26,7 +26,9 @@
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/decimal.h"
 #include "paimon/common/data/columnar/columnar_array.h"
+#include "paimon/common/data/columnar/columnar_batch_context.h"
 #include "paimon/common/data/columnar/columnar_map.h"
+#include "paimon/common/data/columnar/columnar_row_ref.h"
 #include "paimon/common/utils/date_time_utils.h"
 
 namespace paimon {
@@ -35,8 +37,11 @@ Decimal ColumnarRow::GetDecimal(int32_t pos, int32_t precision, int32_t scale) c
     auto array = arrow::internal::checked_cast<const ArrayType*>(array_vec_[pos]);
     assert(array);
     arrow::Decimal128 decimal(array->GetValue(row_id_));
-    return Decimal(precision, scale,
-                   static_cast<Decimal::int128_t>(decimal.high_bits()) << 64 | decimal.low_bits());
+    return Decimal(
+        precision, scale,
+        static_cast<Decimal::int128_t>(
+            static_cast<Decimal::uint128_t>(static_cast<uint64_t>(decimal.high_bits())) << 64 |
+            decimal.low_bits()));
 }
 
 Timestamp ColumnarRow::GetTimestamp(int32_t pos, int32_t precision) const {
@@ -57,6 +62,9 @@ Timestamp ColumnarRow::GetTimestamp(int32_t pos, int32_t precision) const {
 std::shared_ptr<InternalRow> ColumnarRow::GetRow(int32_t pos, int32_t num_fields) const {
     auto struct_array = arrow::internal::checked_cast<const arrow::StructArray*>(array_vec_[pos]);
     assert(struct_array);
+    // NOTE: For performance, the returned nested row does NOT hold shared ownership of the parent
+    // StructArray. Callers must ensure the parent ColumnarRow (or its underlying RecordBatch)
+    // outlives the returned row to avoid dangling pointers.
     return std::make_shared<ColumnarRow>(struct_array->fields(), pool_, row_id_);
 }
 

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

@@ -46,10 +46,19 @@ class MemoryPool;
 /// Columnar row to support access to vector column data. It is a row view in arrow::Array.
 class ColumnarRow : public InternalRow {
  public:
+    /// @brief Construct a ColumnarRow without holding ownership of the underlying arrays.
+    /// @warning The caller MUST ensure the data source (e.g., RecordBatch or parent StructArray)
+    /// outlives this ColumnarRow. The internal array_vec_ stores raw pointers only; if the
+    /// source is freed first, these pointers will dangle. This design is intentional for
+    /// performance—avoiding per-row shared_ptr ref-count overhead on the hot read path.
     ColumnarRow(const arrow::ArrayVector& array_vec, const std::shared_ptr<MemoryPool>& pool,
                 int64_t row_id)
         : ColumnarRow(/*struct_array holder*/ nullptr, array_vec, pool, row_id) {}
 
+    /// @brief Construct a ColumnarRow that holds shared ownership of a StructArray.
+    /// @note When struct_array is non-null it keeps the underlying buffers alive, making it safe
+    /// to outlive the original batch. Prefer this overload when the row may escape the scope of
+    /// its parent container.
     ColumnarRow(const std::shared_ptr<arrow::StructArray>& struct_array,
                 const arrow::ArrayVector& array_vec, const std::shared_ptr<MemoryPool>& pool,
                 int64_t row_id)

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

@@ -34,8 +34,11 @@ Decimal ColumnarRowRef::GetDecimal(int32_t pos, int32_t precision, int32_t scale
     auto array = arrow::internal::checked_cast<const ArrayType*>(ctx_->array_vec[pos].get());
     assert(array);
     arrow::Decimal128 decimal(array->GetValue(row_id_));
-    return Decimal(precision, scale,
-                   static_cast<Decimal::int128_t>(decimal.high_bits()) << 64 | decimal.low_bits());
+    return Decimal(
+        precision, scale,
+        static_cast<Decimal::int128_t>(
+            static_cast<Decimal::uint128_t>(static_cast<uint64_t>(decimal.high_bits())) << 64 |
+            decimal.low_bits()));
 }
 
 Timestamp ColumnarRowRef::GetTimestamp(int32_t pos, int32_t precision) const {

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

@@ -65,7 +65,17 @@ class ColumnarUtils {
             auto value_type_id = dict_type->value_type()->id();
             auto index_type_id = dict_type->index_type()->id();
             int64_t dict_index = -1;
-            if (index_type_id == arrow::Type::type::INT32) {
+            if (index_type_id == arrow::Type::type::INT8) {
+                auto indices =
+                    arrow::internal::checked_cast<arrow::Int8Array*>(typed_array->indices().get());
+                assert(indices);
+                dict_index = indices->Value(pos);
+            } else if (index_type_id == arrow::Type::type::INT16) {
+                auto indices =
+                    arrow::internal::checked_cast<arrow::Int16Array*>(typed_array->indices().get());
+                assert(indices);
+                dict_index = indices->Value(pos);
+            } else if (index_type_id == arrow::Type::type::INT32) {
                 auto indices =
                     arrow::internal::checked_cast<arrow::Int32Array*>(typed_array->indices().get());
                 assert(indices);

src/paimon/common/data/decimal.cpp

@@ -46,10 +46,10 @@ const int64_t Decimal::POWERS_OF_TEN[MAX_COMPACT_PRECISION + 1] = {1,
                                                                    10000000000000000l,
                                                                    100000000000000000l,
                                                                    1000000000000000000l};
-const Decimal::int128_t Decimal::INT128_MAXIMUM_VALUE =
-    static_cast<Decimal::int128_t>(0x7fffffffffffffff) << 64 | 0xffffffffffffffff;
+const Decimal::int128_t Decimal::INT128_MAXIMUM_VALUE = static_cast<Decimal::int128_t>(
+    static_cast<Decimal::uint128_t>(0x7fffffffffffffffULL) << 64 | 0xffffffffffffffff);
 const Decimal::int128_t Decimal::INT128_MINIMUM_VALUE =
-    static_cast<Decimal::int128_t>(0x8000000000000000) << 64;
+    static_cast<Decimal::int128_t>(static_cast<Decimal::uint128_t>(0x8000000000000000ULL) << 64);
 
 std::string Decimal::ToString() const {
     auto type = arrow::decimal128(Precision(), Scale());

src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp

@@ -60,7 +60,11 @@ Result<std::shared_ptr<BloomFilterFileIndexReader>> BloomFilterFileIndexReader::
     const std::shared_ptr<arrow::DataType>& arrow_type, const std::shared_ptr<Bytes>& bytes) {
     // compatible with java, little endian
     const char* data = bytes->data();
-    int32_t num_hash_functions = ((data[0] << 24) + (data[1] << 16) + (data[2] << 8) + data[3]);
+    auto num_hash_functions =
+        static_cast<int32_t>((static_cast<uint32_t>(static_cast<uint8_t>(data[0])) << 24) |
+                             (static_cast<uint32_t>(static_cast<uint8_t>(data[1])) << 16) |
+                             (static_cast<uint32_t>(static_cast<uint8_t>(data[2])) << 8) |
+                             static_cast<uint32_t>(static_cast<uint8_t>(data[3])));
     PAIMON_ASSIGN_OR_RAISE(FastHash::HashFunction hash_function,
                            FastHash::GetHashFunction(arrow_type));
     auto bit_set = std::make_unique<BloomFilter64::BitSet>(bytes, /*offset=*/sizeof(int32_t));

src/paimon/common/file_index/bloomfilter/fast_hash.cpp

@@ -94,14 +94,39 @@ Result<FastHash::HashFunction> FastHash::GetHashFunction(
 }
 
 int64_t FastHash::GetLongHash(int64_t key) {
-    key = (~key) + (key << 21);  // key = (key << 21) - key - 1;
-    key = key ^ (key >> 24);
-    key = (key + (key << 3)) + (key << 8);  // key * 265
-    key = key ^ (key >> 14);
-    key = (key + (key << 2)) + (key << 4);  // key * 21
-    key = key ^ (key >> 28);
-    key = key + (key << 31);
-    return key;
+    // NOTE: This hash function must produce results identical to the Java implementation.
+    // Java uses two's-complement wrapping arithmetic and arithmetic right-shift (>>).
+
+    auto to_unsigned = [](int64_t v) -> uint64_t {
+        uint64_t result;
+        std::memcpy(&result, &v, sizeof(result));
+        return result;
+    };
+    auto to_signed = [](uint64_t v) -> int64_t {
+        int64_t result;
+        std::memcpy(&result, &v, sizeof(result));
+        return result;
+    };
+    // Arithmetic right-shift: shift unsigned then sign-extend from the top.
+    auto asr = [](uint64_t v, int32_t shift) -> uint64_t {
+        bool sign = (v >> 63) != 0;
+        uint64_t shifted = v >> shift;
+        if (sign) {
+            // Fill the top `shift` bits with 1s.
+            shifted |= ~(~static_cast<uint64_t>(0) >> shift);
+        }
+        return shifted;
+    };
+
+    uint64_t k = to_unsigned(key);
+    k = (~k) + (k << 21);  // key = (key << 21) - key - 1;
+    k = k ^ asr(k, 24);
+    k = (k + (k << 3)) + (k << 8);  // key * 265
+    k = k ^ asr(k, 14);
+    k = (k + (k << 2)) + (k << 4);  // key * 21
+    k = k ^ asr(k, 28);
+    k = k + (k << 31);
+    return to_signed(k);
 }
 
 int64_t FastHash::Hash64(const char* data, size_t length) {

src/paimon/common/memory/memory_segment_utils.cpp

@@ -132,21 +132,21 @@ int32_t MemorySegmentUtils::ByteIndex(int32_t bit_index) {
 void MemorySegmentUtils::BitUnSet(MemorySegment* segment, int32_t base_offset, int32_t index) {
     int32_t offset = base_offset + ByteIndex(index);
     char current = segment->Get(offset);
-    current &= ~(1 << (index & BIT_BYTE_INDEX_MASK));
+    current &= static_cast<char>(~(1u << (index & BIT_BYTE_INDEX_MASK)));
     segment->Put(offset, current);
 }
 
 void MemorySegmentUtils::BitSet(MemorySegment* segment, int32_t base_offset, int32_t index) {
     int32_t offset = base_offset + ByteIndex(index);
     char current = segment->Get(offset);
-    current |= (1 << (index & BIT_BYTE_INDEX_MASK));
+    current |= static_cast<char>(1u << (index & BIT_BYTE_INDEX_MASK));
     segment->Put(offset, current);
 }
 
 bool MemorySegmentUtils::BitGet(const MemorySegment& segment, int32_t base_offset, int32_t index) {
     int32_t offset = base_offset + ByteIndex(index);
     char current = segment.Get(offset);
-    return (current & (1 << (index & BIT_BYTE_INDEX_MASK))) != 0;
+    return (current & static_cast<char>(1u << (index & BIT_BYTE_INDEX_MASK))) != 0;
 }
 
 void MemorySegmentUtils::BitSet(std::vector<MemorySegment>* segments, int32_t base_offset,
@@ -155,7 +155,7 @@ void MemorySegmentUtils::BitSet(std::vector<MemorySegment>* segments, int32_t ba
         int32_t offset = base_offset + ByteIndex(index);
         MemorySegment& segment = (*segments)[0];
         char current = segment.Get(offset);
-        current |= (1 << (index & BIT_BYTE_INDEX_MASK));
+        current |= static_cast<char>(1u << (index & BIT_BYTE_INDEX_MASK));
         segment.Put(offset, current);
     } else {
         BitSetMultiSegments(segments, base_offset, index);
@@ -171,15 +171,15 @@ void MemorySegmentUtils::BitSetMultiSegments(std::vector<MemorySegment>* segment
     MemorySegment& segment = (*segments)[seg_index];
 
     char current = segment.Get(seg_offset);
-    current |= (1 << (index & BIT_BYTE_INDEX_MASK));
+    current |= static_cast<char>(1u << (index & BIT_BYTE_INDEX_MASK));
     segment.Put(seg_offset, current);
 }
 
 bool MemorySegmentUtils::BitGet(const std::vector<MemorySegment>& segments, int32_t base_offset,
                                 int32_t index) {
     int32_t offset = base_offset + ByteIndex(index);
     char current = GetValue<char>(segments, offset);
-    return (current & (1 << (index & BIT_BYTE_INDEX_MASK))) != 0;
+    return (current & static_cast<char>(1u << (index & BIT_BYTE_INDEX_MASK))) != 0;
 }
 
 void MemorySegmentUtils::BitUnSet(std::vector<MemorySegment>* segments, int32_t base_offset,
@@ -188,7 +188,7 @@ void MemorySegmentUtils::BitUnSet(std::vector<MemorySegment>* segments, int32_t
         MemorySegment& segment = (*segments)[0];
         int32_t offset = base_offset + ByteIndex(index);
         char current = segment.Get(offset);
-        current &= ~(1 << (index & BIT_BYTE_INDEX_MASK));
+        current &= static_cast<char>(~(1u << (index & BIT_BYTE_INDEX_MASK)));
         segment.Put(offset, current);
     } else {
         BitUnSetMultiSegments(segments, base_offset, index);
@@ -204,7 +204,7 @@ void MemorySegmentUtils::BitUnSetMultiSegments(std::vector<MemorySegment>* segme
     MemorySegment& segment = (*segments)[seg_index];
 
     char current = segment.Get(seg_offset);
-    current &= ~(1 << (index & BIT_BYTE_INDEX_MASK));
+    current &= static_cast<char>(~(1u << (index & BIT_BYTE_INDEX_MASK)));
     segment.Put(seg_offset, current);
 }
 

src/paimon/common/memory/memory_segment_utils.h

@@ -20,6 +20,7 @@
 #include <cstdint>
 #include <cstring>
 #include <memory>
+#include <type_traits>
 #include <vector>
 
 #include "fmt/format.h"
@@ -429,22 +430,31 @@ template <typename T>
 inline T MemorySegmentUtils::GetValueSlowly(const std::vector<MemorySegment>& segments,
                                             int32_t seg_size, int32_t seg_num, int32_t seg_offset) {
     MemorySegment segment = segments[seg_num];
-    T ret = 0;
-    for (size_t i = 0; i < sizeof(T); i++) {
+    if constexpr (std::is_same_v<T, bool>) {
         if (seg_offset == seg_size) {
             segment = segments[++seg_num];
             seg_offset = 0;
         }
-        T unsigned_byte = segment.Get(seg_offset) & 0xff;
-        if (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
-            ret |= (unsigned_byte << (i * 8));
-        } else {
-            int32_t shift_count = sizeof(T) - 1;
-            ret |= (unsigned_byte << ((shift_count - i) * 8));
+        return static_cast<bool>(static_cast<uint8_t>(segment.Get(seg_offset)));
+    } else {
+        using UnsignedT = std::make_unsigned_t<T>;
+        UnsignedT ret = 0;
+        for (size_t i = 0; i < sizeof(T); i++) {
+            if (seg_offset == seg_size) {
+                segment = segments[++seg_num];
+                seg_offset = 0;
+            }
+            UnsignedT unsigned_byte = static_cast<uint8_t>(segment.Get(seg_offset));
+            if (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
+                ret |= (unsigned_byte << (i * 8));
+            } else {
+                int32_t shift_count = sizeof(T) - 1;
+                ret |= (unsigned_byte << ((shift_count - i) * 8));
+            }
+            seg_offset++;
         }
-        seg_offset++;
+        return static_cast<T>(ret);
     }
-    return ret;
 }
 
 inline Status MemorySegmentUtils::CopyToStream(const std::vector<MemorySegment>& segments,

src/paimon/common/memory/memory_slice_output.cpp

@@ -94,6 +94,8 @@ void MemorySliceOutput::EnsureSize(int32_t size) {
     int32_t capacity = segment_.Size();
     int32_t min_capacity = segment_.Size() + size;
     while (capacity < min_capacity) {
+        // capacity is always a power-of-two and <= INT32_MAX/2 in practice,
+        // so this shift does not overflow.
         capacity <<= 1;
     }
 

src/paimon/common/predicate/literal_converter.cpp

@@ -239,8 +239,9 @@ std::vector<Literal> LiteralConverter::GetLiteralFromDecimalArray(const arrow::A
             literals.emplace_back(FieldType::DECIMAL);
         } else {
             const arrow::Decimal128 decimal(array_.GetValue(i));
-            auto value =
-                static_cast<Decimal::int128_t>(decimal.high_bits()) << 64 | decimal.low_bits();
+            auto value = static_cast<Decimal::int128_t>(
+                static_cast<Decimal::uint128_t>(static_cast<uint64_t>(decimal.high_bits())) << 64 |
+                decimal.low_bits());
             literals.emplace_back(Decimal(precision, scale, value));
         }
     }