fix: fix undefined behavior in bitwise left-shift operations across codebase (#308)

Author: lxy-9602

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);