[BugFix] fix upgrade_if_overflow ASAN crash (backport #56261)

be/src/column/binary_column.cpp

@@ -130,6 +130,11 @@ StatusOr<ColumnPtr> BinaryColumnBase<T>::replicate(const Buffer<uint32_t>& offse
         auto bytes_size = _offsets[i + 1] - _offsets[i];
         total_size += bytes_size * (offsets[i + 1] - offsets[i]);
     }
+
+    if (total_size >= Column::MAX_CAPACITY_LIMIT) {
+        return Status::InternalError("replicated column size will exceed the limit");
+    }
+
     dest_bytes.resize(total_size);
     dest_offsets.resize(dest_offsets.size() + offsets.back());
 
@@ -143,13 +148,6 @@ StatusOr<ColumnPtr> BinaryColumnBase<T>::replicate(const Buffer<uint32_t>& offse
         }
     }
 
-    auto ret = dest->upgrade_if_overflow();
-    if (!ret.ok()) {
-        return ret.status();
-    } else if (ret.value() != nullptr) {
-        return ret.value();
-    }
-
     return dest;
 }
 
@@ -664,8 +662,10 @@ std::string BinaryColumnBase<T>::raw_item_value(size_t idx) const {
     return s;
 }
 
+// find first overflow point in offsets[start,end)
+// return the first overflow point or end if not found
 size_t find_first_overflow_point(const BinaryColumnBase<uint32_t>::Offsets& offsets, size_t start, size_t end) {
-    for (size_t i = start; i < end; i++) {
+    for (size_t i = start; i < end - 1; i++) {
         if (offsets[i] > offsets[i + 1]) {
             return i + 1;
         }

be/src/exprs/array_map_expr.cpp

@@ -173,9 +173,11 @@ StatusOr<ColumnPtr> ArrayMapExpr::evaluate_lambda_expr(ExprContext* context, Chu
                 auto view_column = ArrayViewColumn::from_array_column(captured_column);
                 ASSIGN_OR_RETURN(auto replicated_view_column, view_column->replicate(aligned_offsets->get_data()));
                 cur_chunk->append_column(replicated_view_column, slot_id);
+                RETURN_IF_ERROR(view_column->capacity_limit_reached());
             } else {
                 ASSIGN_OR_RETURN(auto replicated_column, captured_column->replicate(aligned_offsets->get_data()));
                 cur_chunk->append_column(replicated_column, slot_id);
+                RETURN_IF_ERROR(replicated_column->capacity_limit_reached());
             }
         }
     }