[refine](Column)Disallow implicit conversion of ColumnPtr to IColumn* (…

…#45588)

### What problem does this PR solve?

Previously, we allowed ColumnPtr to be directly converted to Column*:  
```C++
    ColumnPtr column;  
    const IColumn* ptr = column;  
```
This can easily cause confusion.  
For example, in the following code: 
```C++
    ColumnPtr column;  
    const auto& const_column = check_and_get_column<ColumnConst>(column);  
```
The matched function is:  
```C++
template <>  
const doris::vectorized::ColumnConst* check_and_get_column<doris::vectorized::ColumnConst>(  
        const IColumn* column)  
```
However, the actual type of const_column is:  
```C++
const doris::vectorized::ColumnConst* const&
```

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [x] No need to test or manual test. Explain why:
- [x] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

be/src/exec/table_connector.cpp

@@ -118,16 +118,17 @@ Status TableConnector::convert_column_data(const vectorized::ColumnPtr& column_p
             fmt::format_to(_insert_stmt_buffer, "\"{}\"", str);
         }
     };
-    const vectorized::IColumn* column = column_ptr;
+    const vectorized::IColumn* column = column_ptr.get();
     if (type_ptr->is_nullable()) {
-        auto nullable_column = assert_cast<const vectorized::ColumnNullable*>(column_ptr.get());
+        const auto* nullable_column =
+                assert_cast<const vectorized::ColumnNullable*>(column_ptr.get());
         if (nullable_column->is_null_at(row)) {
             fmt::format_to(_insert_stmt_buffer, "{}", "NULL");
             return Status::OK();
         }
         column = nullable_column->get_nested_column_ptr().get();
     } else {
-        column = column_ptr;
+        column = column_ptr.get();
     }
     auto [item, size] = column->get_data_at(row);
     switch (type.type) {

be/src/olap/push_handler.cpp

@@ -518,7 +518,7 @@ Status PushBrokerReader::_convert_to_output_block(vectorized::Block* block) {
         column_ptr = _src_block.get_by_position(result_column_id).column;
         // column_ptr maybe a ColumnConst, convert it to a normal column
         column_ptr = column_ptr->convert_to_full_column_if_const();
-        DCHECK(column_ptr != nullptr);
+        DCHECK(column_ptr);
 
         // because of src_slot_desc is always be nullable, so the column_ptr after do dest_expr
         // is likely to be nullable

be/src/olap/rowset/segment_v2/column_reader.cpp

@@ -1267,8 +1267,8 @@ Status FileColumnIterator::next_batch(size_t* n, vectorized::MutableColumnPtr& d
                     DCHECK_EQ(this_run, num_rows);
                 } else {
                     *has_null = true;
-                    auto* null_col =
-                            vectorized::check_and_get_column<vectorized::ColumnNullable>(dst);
+                    const auto* null_col =
+                            vectorized::check_and_get_column<vectorized::ColumnNullable>(dst.get());
                     if (null_col != nullptr) {
                         const_cast<vectorized::ColumnNullable*>(null_col)->insert_null_elements(
                                 this_run);
@@ -1328,8 +1328,9 @@ Status FileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t co
                 auto origin_index = _page.data_decoder->current_index();
                 if (this_read_count > 0) {
                     if (is_null) {
-                        auto* null_col =
-                                vectorized::check_and_get_column<vectorized::ColumnNullable>(dst);
+                        const auto* null_col =
+                                vectorized::check_and_get_column<vectorized::ColumnNullable>(
+                                        dst.get());
                         if (UNLIKELY(null_col == nullptr)) {
                             return Status::InternalError("unexpected column type in column reader");
                         }
@@ -1710,9 +1711,9 @@ Status DefaultNestedColumnIterator::next_batch(size_t* n, vectorized::MutableCol
 static void fill_nested_with_defaults(vectorized::MutableColumnPtr& dst,
                                       vectorized::MutableColumnPtr& sibling_column, size_t nrows) {
     const auto* sibling_array = vectorized::check_and_get_column<vectorized::ColumnArray>(
-            remove_nullable(sibling_column->get_ptr()));
+            remove_nullable(sibling_column->get_ptr()).get());
     const auto* dst_array = vectorized::check_and_get_column<vectorized::ColumnArray>(
-            remove_nullable(dst->get_ptr()));
+            remove_nullable(dst->get_ptr()).get());
     if (!dst_array || !sibling_array) {
         throw doris::Exception(ErrorCode::INTERNAL_ERROR,
                                "Expected array column, but met %s and %s", dst->get_name(),

be/src/olap/rowset/segment_v2/hierarchical_data_reader.h

@@ -165,8 +165,8 @@ class HierarchicalDataReader : public ColumnIterator {
         // will type the type of ColumnObject::NESTED_TYPE, whih is Nullable<ColumnArray<NULLABLE(ColumnObject)>>.
         for (auto& entry : nested_subcolumns) {
             MutableColumnPtr nested_object = ColumnObject::create(true, false);
-            const auto* base_array =
-                    check_and_get_column<ColumnArray>(remove_nullable(entry.second[0].column));
+            const auto* base_array = check_and_get_column<ColumnArray>(
+                    remove_nullable(entry.second[0].column).get());
             MutableColumnPtr offset = base_array->get_offsets_ptr()->assume_mutable();
             auto* nested_object_ptr = assert_cast<ColumnObject*>(nested_object.get());
             // flatten nested arrays

be/src/olap/rowset/segment_v2/segment_iterator.cpp

@@ -1955,8 +1955,7 @@ Status SegmentIterator::next_batch(vectorized::Block* block) {
 
 Status SegmentIterator::_convert_to_expected_type(const std::vector<ColumnId>& col_ids) {
     for (ColumnId i : col_ids) {
-        if (_current_return_columns[i] == nullptr || _converted_column_ids[i] ||
-            _is_pred_column[i]) {
+        if (!_current_return_columns[i] || _converted_column_ids[i] || _is_pred_column[i]) {
             continue;
         }
         if (!_segment->same_with_storage_type(
@@ -1999,7 +1998,7 @@ Status SegmentIterator::copy_column_data_by_selector(vectorized::IColumn* input_
         return Status::RuntimeError("copy_column_data_by_selector nullable mismatch");
     }
 
-    return input_col_ptr->filter_by_selector(sel_rowid_idx, select_size, output_col);
+    return input_col_ptr->filter_by_selector(sel_rowid_idx, select_size, output_col.get());
 }
 
 void SegmentIterator::_clear_iterators() {

be/src/olap/schema_change.cpp

@@ -337,7 +337,7 @@ Status BlockChanger::change_block(vectorized::Block* ref_block,
             int result_tmp_column_idx = -1;
             RETURN_IF_ERROR(ctx->execute(ref_block, &result_tmp_column_idx));
             auto& result_tmp_column_def = ref_block->get_by_position(result_tmp_column_idx);
-            if (result_tmp_column_def.column == nullptr) {
+            if (!result_tmp_column_def.column) {
                 return Status::Error<ErrorCode::INTERNAL_ERROR>(
                         "result column={} is nullptr, input expr={}", result_tmp_column_def.name,
                         apache::thrift::ThriftDebugString(*expr));
@@ -430,7 +430,7 @@ Status BlockChanger::_check_cast_valid(vectorized::ColumnPtr input_column,
     if (input_column->is_nullable() != output_column->is_nullable()) {
         if (input_column->is_nullable()) {
             const auto* ref_null_map =
-                    vectorized::check_and_get_column<vectorized::ColumnNullable>(input_column)
+                    vectorized::check_and_get_column<vectorized::ColumnNullable>(input_column.get())
                             ->get_null_map_column()
                             .get_data()
                             .data();
@@ -446,10 +446,12 @@ Status BlockChanger::_check_cast_valid(vectorized::ColumnPtr input_column,
             }
         } else {
             const auto& null_map_column =
-                    vectorized::check_and_get_column<vectorized::ColumnNullable>(output_column)
+                    vectorized::check_and_get_column<vectorized::ColumnNullable>(
+                            output_column.get())
                             ->get_null_map_column();
             const auto& nested_column =
-                    vectorized::check_and_get_column<vectorized::ColumnNullable>(output_column)
+                    vectorized::check_and_get_column<vectorized::ColumnNullable>(
+                            output_column.get())
                             ->get_nested_column();
             const auto* new_null_map = null_map_column.get_data().data();
 
@@ -481,12 +483,12 @@ Status BlockChanger::_check_cast_valid(vectorized::ColumnPtr input_column,
 
     if (input_column->is_nullable() && output_column->is_nullable()) {
         const auto* ref_null_map =
-                vectorized::check_and_get_column<vectorized::ColumnNullable>(input_column)
+                vectorized::check_and_get_column<vectorized::ColumnNullable>(input_column.get())
                         ->get_null_map_column()
                         .get_data()
                         .data();
         const auto* new_null_map =
-                vectorized::check_and_get_column<vectorized::ColumnNullable>(output_column)
+                vectorized::check_and_get_column<vectorized::ColumnNullable>(output_column.get())
                         ->get_null_map_column()
                         .get_data()
                         .data();

be/src/pipeline/exec/hashjoin_build_sink.cpp

@@ -254,7 +254,7 @@ Status HashJoinBuildSinkLocalState::_extract_join_column(
             // update nulllmap and split nested out of ColumnNullable when serialize_null_into_key is false and column is nullable
             const auto& col_nested = nullable->get_nested_column();
             const auto& col_nullmap = nullable->get_null_map_data();
-            DCHECK(null_map != nullptr);
+            DCHECK(null_map);
             vectorized::VectorizedUtils::update_null_map(null_map->get_data(), col_nullmap);
             raw_ptrs[i] = &col_nested;
         } else {

be/src/pipeline/exec/hashjoin_probe_operator.cpp

@@ -371,7 +371,7 @@ Status HashJoinProbeLocalState::_extract_join_column(vectorized::Block& block,
         _need_null_map_for_probe = _need_probe_null_map(block, res_col_ids);
     }
     if (_need_null_map_for_probe) {
-        if (_null_map_column == nullptr) {
+        if (!_null_map_column) {
             _null_map_column = vectorized::ColumnUInt8::create();
         }
         _null_map_column->get_data().assign(block.rows(), (uint8_t)0);
@@ -389,7 +389,7 @@ Status HashJoinProbeLocalState::_extract_join_column(vectorized::Block& block,
             // update nulllmap and split nested out of ColumnNullable when serialize_null_into_key is false and column is nullable
             const auto& col_nested = nullable->get_nested_column();
             const auto& col_nullmap = nullable->get_null_map_data();
-            DCHECK(_null_map_column != nullptr);
+            DCHECK(_null_map_column);
             vectorized::VectorizedUtils::update_null_map(_null_map_column->get_data(), col_nullmap);
             _probe_columns[i] = &col_nested;
         } else {

be/src/pipeline/exec/join_probe_operator.cpp

@@ -150,7 +150,7 @@ Status JoinProbeLocalState<SharedStateArg, Derived>::_build_output_block(
                 /// TODO: maybe need a method to check if a column need to be converted to full
                 /// column.
                 if (is_column_const(*origin_column) ||
-                    check_column<vectorized::ColumnArray>(origin_column)) {
+                    check_column<vectorized::ColumnArray>(origin_column.get())) {
                     auto column_ptr = origin_column->convert_to_full_column_if_const();
                     insert_column_datas(mutable_columns[i], column_ptr, rows);
                 } else {

be/src/pipeline/exec/olap_scan_operator.cpp

@@ -246,9 +246,8 @@ Status OlapScanLocalState::_should_push_down_function_filter(vectorized::Vectori
             DCHECK(children[1 - i]->type().is_string_type());
             std::shared_ptr<ColumnPtrWrapper> const_col_wrapper;
             RETURN_IF_ERROR(children[1 - i]->get_const_col(expr_ctx, &const_col_wrapper));
-            if (const vectorized::ColumnConst* const_column =
-                        check_and_get_column<vectorized::ColumnConst>(
-                                const_col_wrapper->column_ptr)) {
+            if (const auto* const_column = check_and_get_column<vectorized::ColumnConst>(
+                        const_col_wrapper->column_ptr.get())) {
                 *constant_str = const_column->get_data_at(0);
             } else {
                 pdt = PushDownType::UNACCEPTABLE;

be/src/pipeline/exec/scan_operator.cpp

@@ -520,8 +520,8 @@ Status ScanLocalState<Derived>::_eval_const_conjuncts(vectorized::VExpr* vexpr,
     if (vexpr->is_constant()) {
         std::shared_ptr<ColumnPtrWrapper> const_col_wrapper;
         RETURN_IF_ERROR(vexpr->get_const_col(expr_ctx, &const_col_wrapper));
-        if (const auto* const_column =
-                    check_and_get_column<vectorized::ColumnConst>(const_col_wrapper->column_ptr)) {
+        if (const auto* const_column = check_and_get_column<vectorized::ColumnConst>(
+                    const_col_wrapper->column_ptr.get())) {
             constant_val = const_cast<char*>(const_column->get_data_at(0).data);
             if (constant_val == nullptr || !*reinterpret_cast<bool*>(constant_val)) {
                 *pdt = PushDownType::ACCEPTABLE;
@@ -530,7 +530,7 @@ Status ScanLocalState<Derived>::_eval_const_conjuncts(vectorized::VExpr* vexpr,
             }
         } else if (const auto* bool_column =
                            check_and_get_column<vectorized::ColumnVector<vectorized::UInt8>>(
-                                   const_col_wrapper->column_ptr)) {
+                                   const_col_wrapper->column_ptr.get())) {
             // TODO: If `vexpr->is_constant()` is true, a const column is expected here.
             //  But now we still don't cover all predicates for const expression.
             //  For example, for query `SELECT col FROM tbl WHERE 'PROMOTION' LIKE 'AAA%'`,
@@ -690,7 +690,7 @@ Status ScanLocalState<Derived>::_should_push_down_binary_predicate(
             std::shared_ptr<ColumnPtrWrapper> const_col_wrapper;
             RETURN_IF_ERROR(children[1 - i]->get_const_col(expr_ctx, &const_col_wrapper));
             if (const auto* const_column = check_and_get_column<vectorized::ColumnConst>(
-                        const_col_wrapper->column_ptr)) {
+                        const_col_wrapper->column_ptr.get())) {
                 *slot_ref_child = i;
                 *constant_val = const_column->get_data_at(0);
             } else {

be/src/vec/aggregate_functions/aggregate_function_window.h

@@ -402,7 +402,7 @@ struct LeadLagData {
                 if (nullable_column->is_null_at(0)) {
                     _default_value.reset();
                 } else {
-                    _default_value.set_value(nullable_column->get_nested_column_ptr(), 0);
+                    _default_value.set_value(nullable_column->get_nested_column_ptr().get(), 0);
                 }
             } else {
                 _default_value.set_value(column, 0);

be/src/vec/columns/column_object.cpp

@@ -1484,7 +1484,7 @@ Status ColumnObject::serialize_one_row_to_json_format(size_t row, rapidjson::Str
 #endif
     for (const auto& subcolumn : subcolumns) {
         RETURN_IF_ERROR(find_and_set_leave_value(
-                subcolumn->data.get_finalized_column_ptr(), subcolumn->path,
+                subcolumn->data.get_finalized_column_ptr().get(), subcolumn->path,
                 subcolumn->data.get_least_common_type_serde(),
                 subcolumn->data.get_least_common_type(),
                 subcolumn->data.least_common_type.get_base_type_id(), root,
@@ -1558,7 +1558,7 @@ Status ColumnObject::merge_sparse_to_root_column() {
                 continue;
             }
             bool succ = find_and_set_leave_value(
-                    column, subcolumn->path, subcolumn->data.get_least_common_type_serde(),
+                    column.get(), subcolumn->path, subcolumn->data.get_least_common_type_serde(),
                     subcolumn->data.get_least_common_type(),
                     subcolumn->data.least_common_type.get_base_type_id(), root,
                     doc_structure->GetAllocator(), mem_pool, i);
@@ -1705,7 +1705,7 @@ bool ColumnObject::empty() const {
 }
 
 ColumnPtr get_base_column_of_array(const ColumnPtr& column) {
-    if (const auto* column_array = check_and_get_column<ColumnArray>(column)) {
+    if (const auto* column_array = check_and_get_column<ColumnArray>(column.get())) {
         return column_array->get_data_ptr();
     }
     return column;

be/src/vec/common/cow.h

@@ -203,8 +203,6 @@ class COW {
 
         operator bool() const { return t != nullptr; }
 
-        operator T*() const { return t; }
-
     private:
         T* t = nullptr;
     };
@@ -346,8 +344,8 @@ class COW {
         operator const immutable_ptr<T>&() const { return value; }
         operator immutable_ptr<T>&() { return value; }
 
-        operator bool() const { return value != nullptr; }
-        bool operator!() const { return value == nullptr; }
+        operator bool() const { return value.get() != nullptr; }
+        bool operator!() const { return value.get() == nullptr; }
 
         bool operator==(const chameleon_ptr& rhs) const { return value == rhs.value; }
         bool operator!=(const chameleon_ptr& rhs) const { return value != rhs.value; }

be/src/vec/exec/format/column_type_convert.cpp

@@ -99,7 +99,7 @@ ColumnPtr ColumnTypeConverter::get_column(const TypeDescriptor& src_type, Column
         return dst_column;
     }
 
-    if (_cached_src_column == nullptr) {
+    if (!_cached_src_column) {
         _cached_src_type =
                 DataTypeFactory::instance().create_data_type(src_type, dst_type->is_nullable());
         _cached_src_column =

be/src/vec/exec/format/csv/csv_reader.cpp

@@ -657,7 +657,7 @@ Status CsvReader::_fill_dest_columns(const Slice& line, Block* block,
                 col_idx < _split_values.size() ? _split_values[col_idx] : _s_null_slice;
         Slice slice {value.data, value.size};
 
-        IColumn* col_ptr = columns[i];
+        IColumn* col_ptr = columns[i].get();
         if (!_is_load) {
             col_ptr = const_cast<IColumn*>(
                     block->get_by_position(_file_slot_idx_map[i]).column.get());
@@ -700,7 +700,7 @@ Status CsvReader::_fill_dest_columns(const Slice& line, Block* block,
 Status CsvReader::_fill_empty_line(Block* block, std::vector<MutableColumnPtr>& columns,
                                    size_t* rows) {
     for (int i = 0; i < _file_slot_descs.size(); ++i) {
-        IColumn* col_ptr = columns[i];
+        IColumn* col_ptr = columns[i].get();
         if (!_is_load) {
             col_ptr = const_cast<IColumn*>(
                     block->get_by_position(_file_slot_idx_map[i]).column.get());