[Refactor](BE) Uniform Expr, Block, Columns size-relative interface (Part I)

be/src/exprs/bitmapfilter_predicate.h

@@ -37,7 +37,7 @@ class BitmapFilterFuncBase : public RuntimeFilterFuncBase {
     virtual void light_copy(BitmapFilterFuncBase* other) { _not_in = other->_not_in; }
     virtual uint16_t find_fixed_len_olap_engine(const char* data, const uint8* nullmap,
                                                 uint16_t* offsets, int number) = 0;
-    virtual void find_batch(const char* data, const uint8* nullmap, int number,
+    virtual void find_batch(const char* data, const uint8* nullmap, size_t number,
                             uint8* results) const = 0;
     virtual size_t size() const = 0;
     bool is_not_in() const { return _not_in; }
@@ -65,7 +65,7 @@ class BitmapFilterFunc : public BitmapFilterFuncBase {
     uint16_t find_fixed_len_olap_engine(const char* data, const uint8* nullmap, uint16_t* offsets,
                                         int number) override;
 
-    void find_batch(const char* data, const uint8* nullmap, int number,
+    void find_batch(const char* data, const uint8* nullmap, size_t number,
                     uint8* results) const override;
 
     bool empty() override { return _bitmap_value->empty(); }
@@ -133,9 +133,9 @@ uint16_t BitmapFilterFunc<type>::find_fixed_len_olap_engine(const char* data, co
 }
 
 template <PrimitiveType type>
-void BitmapFilterFunc<type>::find_batch(const char* data, const uint8* nullmap, int number,
+void BitmapFilterFunc<type>::find_batch(const char* data, const uint8* nullmap, size_t number,
                                         uint8* results) const {
-    for (int i = 0; i < number; i++) {
+    for (size_t i = 0; i < number; i++) {
         results[i] = false;
         if (nullmap != nullptr && nullmap[i]) {
             continue;

be/src/olap/column_predicate.h

@@ -325,7 +325,7 @@ class ColumnPredicate {
         _judge_filter_rows = 0;
     }
 
-    void do_judge_selectivity(int64_t filter_rows, int64_t input_rows) const {
+    void do_judge_selectivity(uint64_t filter_rows, uint64_t input_rows) const {
         if ((_judge_counter--) == 0) {
             reset_judge_selectivity();
         }
@@ -352,8 +352,8 @@ class ColumnPredicate {
     // without recalculating. At the beginning of the next period,
     // reset_judge_selectivity is used to reset these variables.
     mutable int _judge_counter = 0;
-    mutable int _judge_input_rows = 0;
-    mutable int _judge_filter_rows = 0;
+    mutable uint64_t _judge_input_rows = 0;
+    mutable uint64_t _judge_filter_rows = 0;
     mutable bool _always_true = false;
 };
 

be/src/olap/push_handler.cpp

@@ -471,7 +471,7 @@ Status PushBrokerReader::_init_src_block() {
 }
 
 Status PushBrokerReader::_cast_to_input_block() {
-    size_t idx = 0;
+    uint32_t idx = 0;
     for (auto& slot_desc : _src_slot_descs) {
         if (_name_to_col_type.find(slot_desc->col_name()) == _name_to_col_type.end()) {
             continue;

be/src/pipeline/exec/aggregation_sink_operator.cpp

@@ -20,6 +20,7 @@
 #include <memory>
 #include <string>
 
+#include "common/cast_set.h"
 #include "common/status.h"
 #include "pipeline/exec/operator.h"
 #include "runtime/primitive_type.h"
@@ -814,7 +815,8 @@ Status AggSinkOperatorX::open(RuntimeState* state) {
     // check output type
     if (_needs_finalize) {
         RETURN_IF_ERROR(vectorized::AggFnEvaluator::check_agg_fn_output(
-                _probe_expr_ctxs.size(), _aggregate_evaluators, _agg_fn_output_row_descriptor));
+                cast_set<uint32_t>(_probe_expr_ctxs.size()), _aggregate_evaluators,
+                _agg_fn_output_row_descriptor));
     }
     RETURN_IF_ERROR(vectorized::VExpr::open(_probe_expr_ctxs, state));
 

be/src/pipeline/exec/scan_operator.cpp

@@ -375,7 +375,7 @@ Status ScanLocalState<Derived>::_normalize_bloom_filter(vectorized::VExpr* expr,
                                                         vectorized::VExprContext* expr_ctx,
                                                         SlotDescriptor* slot, PushDownType* pdt) {
     if (TExprNodeType::BLOOM_PRED == expr->node_type()) {
-        DCHECK(expr->children().size() == 1);
+        DCHECK(expr->get_num_children() == 1);
         PushDownType temp_pdt = _should_push_down_bloom_filter();
         if (temp_pdt != PushDownType::UNACCEPTABLE) {
             _filter_predicates.bloom_filters.emplace_back(slot->col_name(),
@@ -391,7 +391,7 @@ Status ScanLocalState<Derived>::_normalize_bitmap_filter(vectorized::VExpr* expr
                                                          vectorized::VExprContext* expr_ctx,
                                                          SlotDescriptor* slot, PushDownType* pdt) {
     if (TExprNodeType::BITMAP_PRED == expr->node_type()) {
-        DCHECK(expr->children().size() == 1);
+        DCHECK(expr->get_num_children() == 1);
         PushDownType temp_pdt = _should_push_down_bitmap_filter();
         if (temp_pdt != PushDownType::UNACCEPTABLE) {
             _filter_predicates.bitmap_filters.emplace_back(slot->col_name(),
@@ -620,7 +620,7 @@ Status ScanLocalState<Derived>::_normalize_in_and_eq_predicate(vectorized::VExpr
         range.intersection(temp_range);
         *pdt = PushDownType::ACCEPTABLE;
     } else if (TExprNodeType::BINARY_PRED == expr->node_type()) {
-        DCHECK(expr->children().size() == 2);
+        DCHECK(expr->get_num_children() == 2);
         auto eq_checker = [](const std::string& fn_name) { return fn_name == "eq"; };
 
         StringRef value;
@@ -769,7 +769,7 @@ Status ScanLocalState<Derived>::_normalize_not_in_and_not_eq_predicate(
             iter->next();
         }
     } else if (TExprNodeType::BINARY_PRED == expr->node_type()) {
-        DCHECK(expr->children().size() == 2);
+        DCHECK(expr->get_num_children() == 2);
 
         auto ne_checker = [](const std::string& fn_name) { return fn_name == "ne"; };
         StringRef value;
@@ -924,7 +924,7 @@ Status ScanLocalState<Derived>::_normalize_noneq_binary_predicate(
         vectorized::VExpr* expr, vectorized::VExprContext* expr_ctx, SlotDescriptor* slot,
         ColumnValueRange<T>& range, PushDownType* pdt) {
     if (TExprNodeType::BINARY_PRED == expr->node_type()) {
-        DCHECK(expr->children().size() == 2);
+        DCHECK(expr->get_num_children() == 2);
 
         auto noneq_checker = [](const std::string& fn_name) {
             return fn_name != "ne" && fn_name != "eq" && fn_name != "eq_for_null";

be/src/pipeline/exec/streaming_aggregation_operator.cpp

@@ -22,6 +22,7 @@
 #include <memory>
 #include <utility>
 
+#include "common/cast_set.h"
 #include "common/compiler_util.h" // IWYU pragma: keep
 #include "pipeline/exec/operator.h"
 #include "vec/exprs/vectorized_agg_fn.h"
@@ -1217,7 +1218,8 @@ Status StreamingAggOperatorX::open(RuntimeState* state) {
     // check output type
     if (_needs_finalize) {
         RETURN_IF_ERROR(vectorized::AggFnEvaluator::check_agg_fn_output(
-                _probe_expr_ctxs.size(), _aggregate_evaluators, _agg_fn_output_row_descriptor));
+                cast_set<uint32_t>(_probe_expr_ctxs.size()), _aggregate_evaluators,
+                _agg_fn_output_row_descriptor));
     }
     RETURN_IF_ERROR(vectorized::VExpr::open(_probe_expr_ctxs, state));
 

be/src/util/simd/bits.h

@@ -19,6 +19,7 @@
 
 #include <cstdint>
 #include <cstring>
+#include <type_traits>
 #include <vector>
 
 #if defined(__ARM_NEON) && defined(__aarch64__)
@@ -27,8 +28,7 @@
 
 #include "util/sse_util.hpp"
 
-namespace doris {
-namespace simd {
+namespace doris::simd {
 
 consteval auto bits_mask_length() {
 #if defined(__ARM_NEON) && defined(__aarch64__)
@@ -70,7 +70,7 @@ inline uint64_t bytes16_mask_to_bits64_mask(const uint8_t* data) {
 inline uint32_t bytes32_mask_to_bits32_mask(const uint8_t* data) {
 #ifdef __AVX2__
     auto zero32 = _mm256_setzero_si256();
-    uint32_t mask = static_cast<uint32_t>(_mm256_movemask_epi8(
+    auto mask = static_cast<uint32_t>(_mm256_movemask_epi8(
             _mm256_cmpgt_epi8(_mm256_loadu_si256(reinterpret_cast<const __m256i*>(data)), zero32)));
 #elif defined(__SSE2__)
     auto zero16 = _mm_setzero_si128();
@@ -125,8 +125,10 @@ void iterate_through_bits_mask(Func func, decltype(bytes_mask_to_bits_mask(nullp
 #endif
 }
 
-inline size_t count_zero_num(const int8_t* __restrict data, size_t size) {
-    size_t num = 0;
+template <typename T>
+    requires requires { std::is_unsigned_v<T>; }
+inline T count_zero_num(const int8_t* __restrict data, T size) {
+    T num = 0;
     const int8_t* end = data + size;
 #if defined(__SSE2__) && defined(__POPCNT__)
     const __m128i zero16 = _mm_setzero_si128();
@@ -138,13 +140,13 @@ inline size_t count_zero_num(const int8_t* __restrict data, size_t size) {
                         _mm_loadu_si128(reinterpret_cast<const __m128i*>(data)), zero16))) |
                 (static_cast<uint64_t>(_mm_movemask_epi8(_mm_cmpeq_epi8(
                          _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + 16)), zero16)))
-                 << 16u) |
+                 << 16U) |
                 (static_cast<uint64_t>(_mm_movemask_epi8(_mm_cmpeq_epi8(
                          _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + 32)), zero16)))
-                 << 32u) |
+                 << 32U) |
                 (static_cast<uint64_t>(_mm_movemask_epi8(_mm_cmpeq_epi8(
                          _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + 48)), zero16)))
-                 << 48u));
+                 << 48U));
     }
 #endif
     for (; data < end; ++data) {
@@ -153,9 +155,10 @@ inline size_t count_zero_num(const int8_t* __restrict data, size_t size) {
     return num;
 }
 
-inline size_t count_zero_num(const int8_t* __restrict data, const uint8_t* __restrict null_map,
-                             size_t size) {
-    size_t num = 0;
+template <typename T>
+    requires requires { std::is_unsigned_v<T>; }
+inline T count_zero_num(const int8_t* __restrict data, const uint8_t* __restrict null_map, T size) {
+    T num = 0;
     const int8_t* end = data + size;
 #if defined(__SSE2__) && defined(__POPCNT__)
     const __m128i zero16 = _mm_setzero_si128();
@@ -172,19 +175,19 @@ inline size_t count_zero_num(const int8_t* __restrict data, const uint8_t* __res
                                  _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + 16)),
                                  zero16),
                          _mm_loadu_si128(reinterpret_cast<const __m128i*>(null_map + 16)))))
-                 << 16u) |
+                 << 16U) |
                 (static_cast<uint64_t>(_mm_movemask_epi8(_mm_or_si128(
                          _mm_cmpeq_epi8(
                                  _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + 32)),
                                  zero16),
                          _mm_loadu_si128(reinterpret_cast<const __m128i*>(null_map + 32)))))
-                 << 32u) |
+                 << 32U) |
                 (static_cast<uint64_t>(_mm_movemask_epi8(_mm_or_si128(
                          _mm_cmpeq_epi8(
                                  _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + 48)),
                                  zero16),
                          _mm_loadu_si128(reinterpret_cast<const __m128i*>(null_map + 48)))))
-                 << 48u));
+                 << 48U));
     }
 #endif
     for (; data < end; ++data, ++null_map) {
@@ -235,5 +238,4 @@ inline size_t find_zero(const std::vector<uint8_t>& vec, size_t start) {
     return find_byte<uint8_t>(vec, start, 0);
 }
 
-} // namespace simd
-} // namespace doris
+} // namespace doris::simd

be/src/vec/core/block.h

@@ -194,7 +194,9 @@ class Block {
     // Skip the rows in block, use in OFFSET, LIMIT operation
     void skip_num_rows(int64_t& offset);
 
-    size_t columns() const { return data.size(); }
+    /// As the assumption we used around, the number of columns won't exceed int16 range. so no need to worry when we
+    ///  assign it to int32.
+    uint32_t columns() const { return static_cast<uint32_t>(data.size()); }
 
     /// Checks that every column in block is not nullptr and has same number of elements.
     void check_number_of_rows(bool allow_null_columns = false) const;

be/src/vec/core/column_numbers.h

@@ -20,11 +20,10 @@
 
 #pragma once
 
-#include <string>
+#include <cstdint>
 #include <vector>
 
 namespace doris::vectorized {
 
-using ColumnNumbers = std::vector<size_t>;
-
+using ColumnNumbers = std::vector<uint32_t>;
 }

be/src/vec/core/decimal_comparison.h

@@ -79,15 +79,15 @@ class DecimalComparison {
     using ArrayA = typename ColVecA::Container;
     using ArrayB = typename ColVecB::Container;
 
-    DecimalComparison(Block& block, size_t result, const ColumnWithTypeAndName& col_left,
+    DecimalComparison(Block& block, uint32_t result, const ColumnWithTypeAndName& col_left,
                       const ColumnWithTypeAndName& col_right) {
         if (!apply(block, result, col_left, col_right)) {
             LOG(FATAL) << fmt::format("Wrong decimal comparison with {} and {}",
                                       col_left.type->get_name(), col_right.type->get_name());
         }
     }
 
-    static bool apply(Block& block, size_t result [[maybe_unused]],
+    static bool apply(Block& block, uint32_t result [[maybe_unused]],
                       const ColumnWithTypeAndName& col_left,
                       const ColumnWithTypeAndName& col_right) {
         if constexpr (_actual) {

be/src/vec/exec/format/orc/vorc_reader.cpp

@@ -773,7 +773,7 @@ Status OrcReader::set_fill_columns(
                     visit_slot(child.get());
                 }
             } else if (VInPredicate* in_predicate = typeid_cast<VInPredicate*>(filter_impl)) {
-                if (in_predicate->children().size() > 0) {
+                if (in_predicate->get_num_children() > 0) {
                     visit_slot(in_predicate->children()[0].get());
                 }
             } else {

be/src/vec/exec/format/parquet/vparquet_reader.cpp

@@ -412,7 +412,7 @@ Status ParquetReader::set_fill_columns(
                     visit_slot(child.get());
                 }
             } else if (VInPredicate* in_predicate = typeid_cast<VInPredicate*>(filter_impl)) {
-                if (in_predicate->children().size() > 0) {
+                if (in_predicate->get_num_children() > 0) {
                     visit_slot(in_predicate->children()[0].get());
                 }
             } else {

be/src/vec/exec/scan/vfile_scanner.cpp

@@ -428,7 +428,7 @@ Status VFileScanner::_cast_to_input_block(Block* block) {
     }
     SCOPED_TIMER(_cast_to_input_block_timer);
     // cast primitive type(PT0) to primitive type(PT1)
-    size_t idx = 0;
+    uint32_t idx = 0;
     for (auto& slot_desc : _input_tuple_desc->slots()) {
         if (_name_to_col_type.find(slot_desc->col_name()) == _name_to_col_type.end()) {
             // skip columns which does not exist in file

be/src/vec/exec/vjdbc_connector.cpp

@@ -259,8 +259,8 @@ Status JdbcConnector::get_next(bool* eos, Block* block, int batch_size) {
     RETURN_IF_ERROR(JniUtil::GetJniExceptionMsg(env));
     env->DeleteLocalRef(map);
 
-    std::vector<size_t> all_columns;
-    for (size_t i = 0; i < column_size; ++i) {
+    std::vector<uint32_t> all_columns;
+    for (uint32_t i = 0; i < column_size; ++i) {
         all_columns.push_back(i);
     }
     SCOPED_RAW_TIMER(&_jdbc_statistic._fill_block_timer);

be/src/vec/exprs/lambda_function/varray_filter_function.cpp

@@ -20,7 +20,6 @@
 #include <memory>
 #include <string>
 #include <utility>
-#include <vector>
 
 #include "common/status.h"
 #include "vec/aggregate_functions/aggregate_function.h"
@@ -36,16 +35,11 @@
 #include "vec/data_types/data_type.h"
 #include "vec/exprs/lambda_function/lambda_function.h"
 #include "vec/exprs/lambda_function/lambda_function_factory.h"
-#include "vec/exprs/vexpr.h"
 #include "vec/utils/util.hpp"
 
-namespace doris {
-namespace vectorized {
-class VExprContext;
-} // namespace vectorized
-} // namespace doris
-
 namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+class VExprContext;
 
 class ArrayFilterFunction : public LambdaFunction {
     ENABLE_FACTORY_CREATOR(ArrayFilterFunction);
@@ -78,7 +72,7 @@ class ArrayFilterFunction : public LambdaFunction {
         auto second_column =
                 block->get_by_position(arguments[1]).column->convert_to_full_column_if_const();
 
-        int input_rows = first_column->size();
+        auto input_rows = first_column->size();
         auto first_outside_null_map = ColumnUInt8::create(input_rows, 0);
         auto first_arg_column = first_column;
         if (first_arg_column->is_nullable()) {
@@ -89,7 +83,7 @@ class ArrayFilterFunction : public LambdaFunction {
             VectorizedUtils::update_null_map(first_outside_null_map->get_data(),
                                              column_array_nullmap.get_data());
         }
-        const ColumnArray& first_col_array = assert_cast<const ColumnArray&>(*first_arg_column);
+        const auto& first_col_array = assert_cast<const ColumnArray&>(*first_arg_column);
         const auto& first_off_data =
                 assert_cast<const ColumnArray::ColumnOffsets&>(first_col_array.get_offsets_column())
                         .get_data();
@@ -113,7 +107,7 @@ class ArrayFilterFunction : public LambdaFunction {
             VectorizedUtils::update_null_map(second_outside_null_map->get_data(),
                                              column_array_nullmap.get_data());
         }
-        const ColumnArray& second_col_array = assert_cast<const ColumnArray&>(*second_arg_column);
+        const auto& second_col_array = assert_cast<const ColumnArray&>(*second_arg_column);
         const auto& second_off_data = assert_cast<const ColumnArray::ColumnOffsets&>(
                                               second_col_array.get_offsets_column())
                                               .get_data();
@@ -180,4 +174,6 @@ class ArrayFilterFunction : public LambdaFunction {
 void register_function_array_filter(doris::vectorized::LambdaFunctionFactory& factory) {
     factory.register_function<ArrayFilterFunction>();
 }
+
+#include "common/compile_check_end.h"
 } // namespace doris::vectorized