[BugFix] Fix column overflow in RowsetUpdateState and CompactionState (backport #33246)

be/src/column/binary_column.cpp

@@ -291,6 +291,10 @@ void BinaryColumnBase<T>::append_value_multiple_times(const void* value, size_t
 
 template <typename T>
 void BinaryColumnBase<T>::_build_slices() const {
+    if constexpr (std::is_same_v<T, uint32_t>) {
+        CHECK_LT(_bytes.size(), (size_t)UINT32_MAX) << "BinaryColumn size overflow";
+    }
+
     DCHECK(_offsets.size() > 0);
     _slices_cache = false;
     _slices.clear();

be/src/storage/lake/rowset_update_state.cpp

@@ -170,7 +170,7 @@ Status RowsetUpdateState::_do_load_upserts_deletes(const TxnLogPB_OpWrite& op_wr
     }
     Schema pkey_schema = ChunkHelper::convert_schema(tablet_schema, pk_columns);
     std::unique_ptr<Column> pk_column;
-    if (!PrimaryKeyEncoder::create_column(pkey_schema, &pk_column).ok()) {
+    if (!PrimaryKeyEncoder::create_column(pkey_schema, &pk_column, true).ok()) {
         CHECK(false) << "create column for primary key encoder failed";
     }
 

be/src/storage/lake/update_compaction_state.cpp

@@ -47,6 +47,8 @@ Status CompactionState::load_segments(Rowset* rowset, const TabletSchemaCSPtr& t
     return _load_segments(rowset, tablet_schema, segment_id);
 }
 
+static const size_t large_compaction_memory_threshold = 1000000000;
+
 Status CompactionState::_load_segments(Rowset* rowset, const TabletSchemaCSPtr& tablet_schema, uint32_t segment_id) {
     vector<uint32_t> pk_columns;
     for (size_t i = 0; i < tablet_schema->num_key_columns(); i++) {
@@ -56,7 +58,7 @@ Status CompactionState::_load_segments(Rowset* rowset, const TabletSchemaCSPtr&
     Schema pkey_schema = ChunkHelper::convert_schema(tablet_schema, pk_columns);
 
     std::unique_ptr<Column> pk_column;
-    CHECK(PrimaryKeyEncoder::create_column(pkey_schema, &pk_column).ok());
+    CHECK(PrimaryKeyEncoder::create_column(pkey_schema, &pk_column, true).ok());
 
     OlapReaderStatistics stats;
     auto res = rowset->get_each_segment_iterator(pkey_schema, &stats);
@@ -92,10 +94,8 @@ Status CompactionState::_load_segments(Rowset* rowset, const TabletSchemaCSPtr&
         itr->close();
     }
     dest = std::move(col);
-    dest->raw_data();
     _memory_usage += dest->memory_usage();
     _update_manager->compaction_state_mem_tracker()->consume(dest->memory_usage());
-
     return Status::OK();
 }
 

be/src/storage/persistent_index.cpp

@@ -19,6 +19,7 @@
 #include <utility>
 
 #include "fs/fs.h"
+#include "gutil/strings/escaping.h"
 #include "gutil/strings/substitute.h"
 #include "storage/chunk_helper.h"
 #include "storage/chunk_iterator.h"
@@ -3768,8 +3769,8 @@ Status PersistentIndex::try_replace(size_t n, const Slice* keys, const IndexValu
     RETURN_IF_ERROR(get(n, keys, found_values.data()));
     std::vector<size_t> replace_idxes;
     for (size_t i = 0; i < n; ++i) {
-        if (found_values[i].get_value() != NullIndexValue &&
-            ((uint32_t)(found_values[i].get_value() >> 32)) <= max_src_rssid) {
+        auto found_value = found_values[i].get_value();
+        if (found_value != NullIndexValue && ((uint32_t)(found_value >> 32)) <= max_src_rssid) {
             replace_idxes.emplace_back(i);
         } else {
             failed->emplace_back(values[i].get_value() & 0xFFFFFFFF);

be/src/storage/primary_index.cpp

@@ -1161,10 +1161,11 @@ Status PrimaryIndex::_build_persistent_values(uint32_t rssid, const vector<uint3
 
 const Slice* PrimaryIndex::_build_persistent_keys(const Column& pks, uint32_t idx_begin, uint32_t idx_end,
                                                   std::vector<Slice>* key_slices) const {
-    if (pks.is_binary()) {
+    if (pks.is_binary() || pks.is_large_binary()) {
         const Slice* vkeys = reinterpret_cast<const Slice*>(pks.raw_data());
         return vkeys + idx_begin;
     } else {
+        CHECK(_key_size > 0);
         const uint8_t* keys = pks.raw_data() + idx_begin * _key_size;
         for (size_t i = idx_begin; i < idx_end; i++) {
             key_slices->emplace_back(keys, _key_size);