[opt](be) Backport nullable and projection optimizations to branch-4.1

be/src/core/block/column_with_type_and_name.cpp

@@ -30,10 +30,10 @@
 #include "core/column/column.h"
 #include "core/column/column_const.h"
 #include "core/column/column_nothing.h"
+#include "core/column/column_nullable.h"
 #include "core/data_type/data_type.h"
 #include "core/data_type/data_type_nullable.h"
 #include "core/types.h"
-#include "util/simd/bits.h"
 
 namespace doris {
 
@@ -105,41 +105,67 @@ void ColumnWithTypeAndName::to_pb_column_meta(PColumnMeta* col_meta) const {
     type->to_pb_column_meta(col_meta);
 }
 
+const ColumnNullable& ColumnWithTypeAndName::get_nullable_column() const {
+    DCHECK(type->is_nullable());
+    DCHECK(column);
+    const auto& [physical_column, _] = unpack_if_const(column);
+    return assert_cast<const ColumnNullable&, TypeCheckOnRelease::DISABLE>(*physical_column);
+}
+
+const ColumnUInt8::Ptr& ColumnWithTypeAndName::get_nullable_null_map_column() const {
+    return get_nullable_column().get_null_map_column_ptr();
+}
+
+NullableColumnInfo ColumnWithTypeAndName::get_nullable_column_info() const {
+    DCHECK(type->is_nullable());
+    DCHECK(column);
+
+    const auto [has_null, only_null] = get_nullable_column().get_null_map_state();
+    return {.has_null = has_null,
+            .only_null = only_null,
+            .is_const = is_column_const(*column),
+            .is_nullable = true};
+}
+
 ColumnWithTypeAndName ColumnWithTypeAndName::unnest_nullable(
         bool replace_null_data_to_default) const {
+    NullableColumnInfo info;
     if (type->is_nullable()) {
-        auto nested_type =
-                assert_cast<const DataTypeNullable*, TypeCheckOnRelease::DISABLE>(type.get())
-                        ->get_nested_type();
-        ColumnPtr nested_column = column;
-        if (column) {
-            // A column_ptr is needed here to ensure that the column in convert_to_full_column_if_const is not released.
-            auto [column_ptr, is_const] = unpack_if_const(column);
-            const auto* source_column =
-                    assert_cast<const ColumnNullable*, TypeCheckOnRelease::DISABLE>(
-                            column_ptr.get());
-            if (is_const) {
-                nested_column =
-                        ColumnConst::create(source_column->get_nested_column_ptr(), column->size());
-            } else {
-                nested_column = source_column->get_nested_column_ptr();
-            }
-
-            if (replace_null_data_to_default) {
-                const auto& null_map = source_column->get_null_map_data();
-                // only need to mutate nested column, avoid to copy nullmap
-                auto mutable_nested_col = (*std::move(nested_column)).mutate();
-                if (simd::contain_one(null_map.data(), null_map.size())) {
-                    mutable_nested_col->replace_column_null_data(null_map.data());
-                }
-
-                return {std::move(mutable_nested_col), nested_type, ""};
-            }
-        }
-        return {nested_column, nested_type, ""};
-    } else {
+        info = get_nullable_column_info();
+    }
+    return unnest_nullable(info, replace_null_data_to_default);
+}
+
+ColumnWithTypeAndName ColumnWithTypeAndName::unnest_nullable(
+        const NullableColumnInfo& info, bool replace_null_data_to_default) const {
+    if (!type->is_nullable()) {
         return {column, type, ""};
     }
+    DCHECK(info.is_nullable);
+
+    const auto& nullable_column = get_nullable_column();
+    const auto get_nested_column = [&]() -> ColumnPtr {
+        const auto& nested_column = nullable_column.get_nested_column_ptr();
+        if (info.is_const) {
+            return ColumnConst::create(nested_column, column->size());
+        }
+        return nested_column;
+    };
+
+    auto nested_type = assert_cast<const DataTypeNullable*, TypeCheckOnRelease::DISABLE>(type.get())
+                               ->get_nested_type();
+    if (replace_null_data_to_default && info.has_null) {
+        if (column->try_replace_null_payload_with_default_without_cow()) {
+            return {get_nested_column(), nested_type, ""};
+        }
+
+        // Only copy the nested column because the original nullable column must remain unchanged.
+        const auto nested_column = get_nested_column();
+        auto mutable_nested_col = nested_column->clone_resized(nested_column->size());
+        mutable_nested_col->replace_column_null_data(nullable_column.get_null_map_data().data());
+        return {std::move(mutable_nested_col), nested_type, ""};
+    }
+    return {get_nested_column(), nested_type, ""};
 }
 
 Status ColumnWithTypeAndName::check_type_and_column_match() const {

be/src/core/block/column_with_type_and_name.h

@@ -26,17 +26,29 @@
 #include <memory>
 #include <string>
 #include <utility>
+#include <vector>
 
+#include "core/column/column_vector.h"
 #include "core/data_type/data_type.h"
 #include "core/data_type_serde/data_type_serde.h"
 #include "core/types.h"
 
 namespace doris {
+class ColumnNullable;
 class PColumnMeta;
 } // namespace doris
 
 namespace doris {
 
+struct NullableColumnInfo {
+    bool has_null = false;
+    bool only_null = false;
+    bool is_const = false;
+    bool is_nullable = false;
+};
+
+using NullableColumnInfos = std::vector<NullableColumnInfo>;
+
 // class WriteBuffer;
 
 /** Column data along with its data type and name.
@@ -69,9 +81,16 @@ struct ColumnWithTypeAndName {
 
     void to_pb_column_meta(PColumnMeta* col_meta) const;
 
+    const ColumnUInt8::Ptr& get_nullable_null_map_column() const;
+    NullableColumnInfo get_nullable_column_info() const;
     ColumnWithTypeAndName unnest_nullable(bool replace_null_data_to_default = false) const;
+    ColumnWithTypeAndName unnest_nullable(const NullableColumnInfo& info,
+                                          bool replace_null_data_to_default) const;
 
     Status check_type_and_column_match() const;
+
+private:
+    const ColumnNullable& get_nullable_column() const;
 };
 
 } // namespace doris

be/src/core/column/column.h

@@ -724,6 +724,17 @@ class IColumn : public COW<IColumn> {
     // column_vector and column_decimal override this method to return true
     virtual bool support_replace_column_null_data() const { return false; }
 
+    /**
+     * Try to replace the payload of NULL rows with the nested column's default value without
+     * going through COW. Implementations must return false without modifying data unless the
+     * complete column ownership chain is exclusive. This is only safe because payloads of rows
+     * that are already NULL are not observable through the nullable column. In particular, a
+     * shared nested column may belong to another nullable column with a different null map.
+     *
+     * This bypasses the normal COW mutation path. Do not use it for general column mutation.
+     */
+    virtual bool try_replace_null_payload_with_default_without_cow() const { return false; }
+
     // For float/double types, replace -0.0 with 0.0, set NaN to quiet NaN,
     // used to ensure data hash equality for -0.0 and +0.0, e.g. aggregate and join
     virtual void replace_float_special_values() {}

be/src/core/column/column_const.h

@@ -121,6 +121,8 @@ class ColumnConst final : public COWHelper<IColumn, ColumnConst> {
 
     bool is_variable_length() const override { return data->is_variable_length(); }
 
+    bool is_exclusive() const override { return IColumn::is_exclusive() && data->is_exclusive(); }
+
     std::string get_name() const override { return "Const(" + data->get_name() + ")"; }
 
     void resize(size_t new_size) override { s = new_size; }
@@ -307,6 +309,13 @@ class ColumnConst final : public COWHelper<IColumn, ColumnConst> {
         return data->support_replace_column_null_data();
     }
 
+    bool try_replace_null_payload_with_default_without_cow() const override {
+        if (!IColumn::is_exclusive()) {
+            return false;
+        }
+        return data->try_replace_null_payload_with_default_without_cow();
+    }
+
     void finalize() override { data->finalize(); }
 
     void erase(size_t start, size_t length) override {

be/src/core/column/column_nullable.cpp

@@ -699,6 +699,29 @@ bool ColumnNullable::only_null() const {
     return !simd::contain_zero(get_null_map_data().data(), size());
 }
 
+ColumnNullable::NullMapState ColumnNullable::get_null_map_state() const {
+    const auto& null_map = get_null_map_data();
+    if (null_map.empty()) {
+        return {.has_null = false, .only_null = true};
+    }
+
+    if (null_map[0]) {
+        return {.has_null = true,
+                .only_null = !simd::contain_zero(null_map.data() + 1, null_map.size() - 1)};
+    }
+    return {.has_null = simd::contain_one(null_map.data() + 1, null_map.size() - 1),
+            .only_null = false};
+}
+
+bool ColumnNullable::try_replace_null_payload_with_default_without_cow() const {
+    if (!is_exclusive()) {
+        return false;
+    }
+
+    const_cast<IColumn&>(get_nested_column()).replace_column_null_data(get_null_map_data().data());
+    return true;
+}
+
 bool ColumnNullable::has_null(size_t begin, size_t end) const {
     return simd::contain_one(get_null_map_data().data() + begin, end - begin);
 }

be/src/core/column/column_nullable.h

@@ -64,6 +64,11 @@ class ColumnNullable final : public COWHelper<IColumn, ColumnNullable> {
     ColumnNullable(const ColumnNullable&) = default;
 
 public:
+    struct NullMapState {
+        bool has_null;
+        bool only_null;
+    };
+
     /** Create a column from immutable/shared subcolumns without cloning them.
       * Call IColumn::mutate before modifying the returned column tree.
       */
@@ -270,7 +275,10 @@ class ColumnNullable final : public COWHelper<IColumn, ColumnNullable> {
                get_null_map_column().is_exclusive();
     }
 
+    bool try_replace_null_payload_with_default_without_cow() const override;
+
     bool only_null() const override;
+    NullMapState get_null_map_state() const;
 
     // used in schema change
     void change_nested_column(ColumnPtr& other) { ((ColumnPtr&)_nested_column) = other; }

be/src/core/column/column_variant.cpp

@@ -832,6 +832,20 @@ size_t ColumnVariant::allocated_bytes() const {
     return res;
 }
 
+bool ColumnVariant::is_exclusive() const {
+    if (!IColumn::is_exclusive()) {
+        return false;
+    }
+    for (const auto& entry : subcolumns) {
+        for (const auto& part : entry->data.data) {
+            if (!part->is_exclusive()) {
+                return false;
+            }
+        }
+    }
+    return serialized_sparse_column->is_exclusive() && serialized_doc_value_column->is_exclusive();
+}
+
 void ColumnVariant::for_each_subcolumn(ColumnCallback callback) {
     for (auto& entry : subcolumns) {
         for (auto& part : entry->data.data) {

be/src/core/column/column_variant.h

@@ -469,6 +469,8 @@ class ColumnVariant final : public COWHelper<IColumn, ColumnVariant> {
 
     bool has_enough_capacity(const IColumn& src) const override { return false; }
 
+    bool is_exclusive() const override;
+
     void for_each_subcolumn(ColumnCallback callback) override;
 
     // Do nothing, call try_insert instead

be/src/exec/operator/operator.cpp

@@ -333,58 +333,54 @@ Status OperatorXBase::do_projections(RuntimeState* state, Block* origin_block,
     if (rows == 0) {
         return Status::OK();
     }
-    Block input_block = *origin_block;
-
-    size_t bytes_usage = 0;
-    ColumnsWithTypeAndName new_columns;
-    for (const auto& projections : local_state->_intermediate_projections) {
-        new_columns.resize(projections.size());
-        for (int i = 0; i < projections.size(); i++) {
-            RETURN_IF_ERROR(projections[i]->execute(&input_block, new_columns[i]));
-        }
-        Block tmp_block {new_columns};
-        bytes_usage += tmp_block.allocated_bytes();
-        input_block.swap(tmp_block);
-    }
-
-    DCHECK_EQ(rows, input_block.rows());
-    auto insert_column_datas = [&](auto& to, ColumnPtr& from, size_t rows) {
-        if (to->is_nullable() && !from->is_nullable()) {
-            if (_keep_origin || !from->is_exclusive()) {
-                auto& null_column = reinterpret_cast<ColumnNullable&>(*to);
-                null_column.get_nested_column().insert_range_from(*from, 0, rows);
-                null_column.get_null_map_column().get_data().resize_fill(rows, 0);
-                bytes_usage += null_column.allocated_bytes();
-            } else {
-                to = ColumnNullable::create(IColumn::mutate(std::move(from)),
-                                            ColumnUInt8::create(rows, 0));
-            }
-        } else {
-            if (_keep_origin || !from->is_exclusive()) {
-                to->insert_range_from(*from, 0, rows);
-                bytes_usage += from->allocated_bytes();
-            } else {
-                to = IColumn::mutate(std::move(from));
+    SCOPED_PEAK_MEM(&local_state->_estimate_memory_usage);
+
+    {
+        Block input_block = *origin_block;
+
+        ColumnsWithTypeAndName new_columns;
+        for (const auto& projections : local_state->_intermediate_projections) {
+            new_columns.resize(projections.size());
+            for (int i = 0; i < projections.size(); i++) {
+                RETURN_IF_ERROR(projections[i]->execute(&input_block, new_columns[i]));
             }
+            Block tmp_block {new_columns};
+            input_block.swap(tmp_block);
         }
-    };
 
-    auto scoped_mutable_block = VectorizedUtils::build_scoped_mutable_mem_reuse_block(
-            output_block, *_output_row_descriptor);
-    auto& mutable_block = scoped_mutable_block.mutable_block();
-    auto& mutable_columns = mutable_block.mutable_columns();
-    if (rows != 0) {
+        DCHECK_EQ(rows, input_block.rows());
+
+        auto scoped_mutable_block = VectorizedUtils::build_scoped_mutable_mem_reuse_block(
+                output_block, *_output_row_descriptor);
+        auto& mutable_columns = scoped_mutable_block.mutable_columns();
         DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) << debug_string();
+        Columns shared_columns(mutable_columns.size());
+
         for (int i = 0; i < mutable_columns.size(); ++i) {
             ColumnPtr column_ptr;
             RETURN_IF_ERROR(local_state->_projections[i]->execute(&input_block, column_ptr));
             column_ptr = column_ptr->convert_to_full_column_if_const();
-            bytes_usage += column_ptr->allocated_bytes();
-            insert_column_datas(mutable_columns[i], column_ptr, rows);
+            if (is_column_nullable(*mutable_columns[i]) && !is_column_nullable(*column_ptr)) {
+                column_ptr = make_nullable(column_ptr, false);
+            }
+            if (column_ptr->is_exclusive()) {
+                mutable_columns[i] = IColumn::mutate(std::move(column_ptr));
+            } else {
+                shared_columns[i] = std::move(column_ptr);
+            }
+        }
+
+        scoped_mutable_block.restore();
+        for (int i = 0; i < shared_columns.size(); ++i) {
+            if (shared_columns[i]) {
+                output_block->replace_by_position(i, std::move(shared_columns[i]));
+            }
         }
-        DCHECK(mutable_block.rows() == rows);
     }
-    local_state->_estimate_memory_usage += bytes_usage;
+
+    origin_block->clear_column_data(
+            local_state->_parent->intermediate_row_desc().num_materialized_slots());
+    DCHECK_EQ(output_block->rows(), rows);
 
     return Status::OK();
 }

be/src/exec/operator/operator.h

@@ -1020,10 +1020,6 @@ class OperatorXBase : public OperatorBase {
     std::string _op_name;
     int _parallel_tasks = 0;
 
-    //_keep_origin is used to avoid copying during projection,
-    // currently set to false only in the nestloop join.
-    bool _keep_origin = true;
-
     // _blockable is true if the operator contains expressions that may block execution
     bool _blockable = false;
 };