[fix](be) Backport BE fixes and refinements to branch-4.1

be/benchmark/benchmark_main.cpp

@@ -17,6 +17,7 @@
 
 #include <benchmark/benchmark.h>
 
+#include "benchmark_arrow_validation.hpp"
 #include "benchmark_bit_pack.hpp"
 #include "benchmark_column_array_view.hpp"
 #include "benchmark_column_array_view_distance.hpp"

be/src/common/config.cpp

@@ -71,6 +71,9 @@ DEFINE_Int32(brpc_port, "8060");
 
 DEFINE_Int32(arrow_flight_sql_port, "8050");
 
+// Validate Arrow input buffers in opted-in Arrow readers before converting them to Doris columns.
+DEFINE_Bool(enable_arrow_input_validation, "true");
+
 DEFINE_Int32(cdc_client_port, "9096");
 
 DEFINE_String(cdc_client_java_opts, "");

be/src/common/config.h

@@ -119,6 +119,9 @@ DECLARE_Int32(brpc_port);
 // Default -1, do not start arrow flight sql server.
 DECLARE_Int32(arrow_flight_sql_port);
 
+// Validate Arrow input buffers in opted-in Arrow readers before converting them to Doris columns.
+DECLARE_Bool(enable_arrow_input_validation);
+
 // port for cdc client scan oltp cdc data
 DECLARE_Int32(cdc_client_port);
 

be/src/core/block/block.cpp

@@ -341,6 +341,21 @@ Status Block::check_type_and_column() const {
     return Status::OK();
 }
 
+Status Block::check_column_and_type_not_null() const {
+    for (size_t i = 0; i != data.size(); ++i) {
+        const auto& elem = data[i];
+        if (!elem.column) {
+            return Status::InternalError("Column in block is nullptr, column index: {}, name: {}",
+                                         i, elem.name);
+        }
+        if (!elem.type) {
+            return Status::InternalError("Type in block is nullptr, column index: {}, name: {}", i,
+                                         elem.name);
+        }
+    }
+    return Status::OK();
+}
+
 size_t Block::rows() const {
     for (const auto& elem : data) {
         if (elem.column) {

be/src/core/block/block.h

@@ -186,6 +186,8 @@ class Block {
 
     Status check_type_and_column() const;
 
+    Status check_column_and_type_not_null() const;
+
     /// Approximate number of bytes used by column data in memory.
     /// This reflects the actual data footprint (e.g. string contents, numeric arrays)
     /// and is the metric used by adaptive batch size byte budgets.

be/src/core/column/column.h

@@ -446,7 +446,8 @@ class IColumn : public COW<IColumn> {
      *  // nullable -> predict_column
      *  // string (dictionary) -> column_dictionary
      */
-    virtual Status filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) {
+    virtual Status filter_by_selector(const uint16_t* sel, size_t sel_size,
+                                      IColumn* col_ptr) const {
         throw doris::Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
                                "Method filter_by_selector is not supported for {}, only "
                                "column_nullable, column_dictionary and predict_column support",

be/src/core/column/column_decimal.h

@@ -149,7 +149,8 @@ class ColumnDecimal final : public COWHelper<IColumn, ColumnDecimal<T>> {
         memset(data.data() + old_size, 0, length * sizeof(data[0]));
     }
 
-    Status filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) override {
+    Status filter_by_selector(const uint16_t* sel, size_t sel_size,
+                              IColumn* col_ptr) const override {
         Self* output = assert_cast<Self*>(col_ptr);
         auto& res_data = output->get_data();
         DCHECK(res_data.empty())

be/src/core/column/column_dictionary.h

@@ -153,19 +153,23 @@ class ColumnDictI32 final : public COWHelper<IColumn, ColumnDictI32> {
                                "permute not supported in ColumnDictionary");
     }
 
-    Status filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) override {
+    Status filter_by_selector(const uint16_t* sel, size_t sel_size,
+                              IColumn* col_ptr) const override {
         auto* res_col = assert_cast<ColumnString*>(col_ptr);
-        _strings.resize(sel_size);
+        if (sel_size == 0) {
+            return Status::OK();
+        }
+        std::vector<StringRef> strings(sel_size);
         size_t length = 0;
         for (size_t i = 0; i != sel_size; ++i) {
-            auto& value = _dict.get_value(_codes[sel[i]]);
-            _strings[i].data = value.data;
-            _strings[i].size = value.size;
+            const auto& value = _dict.get_value(_codes[sel[i]]);
+            strings[i].data = value.data;
+            strings[i].size = value.size;
             length += value.size;
         }
         res_col->get_offsets().reserve(sel_size + res_col->get_offsets().size());
         res_col->get_chars().reserve(length + res_col->get_chars().size());
-        res_col->insert_many_strings_without_reserve(_strings.data(), sel_size);
+        res_col->insert_many_strings_without_reserve(strings.data(), sel_size);
         return Status::OK();
     }
 
@@ -455,7 +459,6 @@ class ColumnDictI32 final : public COWHelper<IColumn, ColumnDictI32> {
     Dictionary _dict;
     Container _codes;
     std::pair<RowsetId, uint32_t> _rowset_segment_id;
-    std::vector<StringRef> _strings;
 };
 
 } // namespace doris

be/src/core/column/column_nullable.cpp

@@ -471,17 +471,15 @@ size_t ColumnNullable::filter(const Filter& filter) {
     return data_result_size;
 }
 
-Status ColumnNullable::filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) {
+Status ColumnNullable::filter_by_selector(const uint16_t* sel, size_t sel_size,
+                                          IColumn* col_ptr) const {
     auto* nullable_col_ptr = assert_cast<ColumnNullable*>(col_ptr);
-    // Access the nested column via const path to avoid assert_mutable_ref (which requires
-    // exclusive ownership). The output col_ptr was just created, so its nested column is exclusive.
-    auto nest_col_raw = const_cast<IColumn*>(
-            static_cast<const IColumn::WrappedPtr&>(nullable_col_ptr->_nested_column).get());
+    auto nested_column = nullable_col_ptr->get_nested_column_ptr();
 
     /// `get_null_map_data` will set `_need_update_has_null` to true
     auto& res_nullmap = nullable_col_ptr->get_null_map_data();
 
-    RETURN_IF_ERROR(get_nested_column().filter_by_selector(sel, sel_size, nest_col_raw));
+    RETURN_IF_ERROR(get_nested_column().filter_by_selector(sel, sel_size, nested_column.get()));
     DCHECK(res_nullmap.empty());
     res_nullmap.resize(sel_size);
     auto& cur_nullmap = get_null_map_column().get_data();

be/src/core/column/column_nullable.h

@@ -206,7 +206,8 @@ class ColumnNullable final : public COWHelper<IColumn, ColumnNullable> {
 
     size_t filter(const Filter& filter) override;
 
-    Status filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) override;
+    Status filter_by_selector(const uint16_t* sel, size_t sel_size,
+                              IColumn* col_ptr) const override;
     MutableColumnPtr permute(const Permutation& perm, size_t limit) const override;
     //    ColumnPtr index(const IColumn & indexes, size_t limit) const override;
     int compare_at(size_t n, size_t m, const IColumn& rhs_, int null_direction_hint) const override;

be/src/core/column/column_string.cpp

@@ -372,7 +372,8 @@ size_t ColumnStr<T>::filter(const IColumn::Filter& filter) {
 }
 
 template <typename T>
-Status ColumnStr<T>::filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) {
+Status ColumnStr<T>::filter_by_selector(const uint16_t* sel, size_t sel_size,
+                                        IColumn* col_ptr) const {
     if constexpr (std::is_same_v<UInt32, T>) {
         auto* col = static_cast<ColumnStr<T>*>(col_ptr);
         Chars& res_chars = col->chars;

be/src/core/column/column_string.h

@@ -482,7 +482,8 @@ class ColumnStr final : public COWHelper<IColumn, ColumnStr<T>> {
     ColumnPtr filter(const IColumn::Filter& filt, ssize_t result_size_hint) const override;
     size_t filter(const IColumn::Filter& filter) override;
 
-    Status filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) override;
+    Status filter_by_selector(const uint16_t* sel, size_t sel_size,
+                              IColumn* col_ptr) const override;
 
     MutableColumnPtr permute(const IColumn::Permutation& perm, size_t limit) const override;
 

be/src/core/column/column_vector.h

@@ -271,7 +271,8 @@ class ColumnVector final : public COWHelper<IColumn, ColumnVector<T>> {
 
     void insert_value(const value_type value) { data.push_back(value); }
 
-    Status filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) override {
+    Status filter_by_selector(const uint16_t* sel, size_t sel_size,
+                              IColumn* col_ptr) const override {
         Self* output = assert_cast<Self*>(col_ptr);
         auto& res_data = output->get_data();
         DCHECK(res_data.empty())