branch-4.1: [fix](be) Materialize const columns before block merge #65770

be/src/core/block/block.h

@@ -447,6 +447,15 @@ class MutableBlock {
     DataTypes _data_types;
     std::vector<std::string> _names;
 
+    void materialize_const_column(size_t position) {
+        if (is_column_const(*_columns[position])) {
+            // ScopedMutableBlock can retain a const destination while merge materializes
+            // its source, so normalize the destination before appending full columns.
+            _columns[position] =
+                    IColumn::mutate(_columns[position]->convert_to_full_column_if_const());
+        }
+    }
+
 public:
     // Build from a consumed Block. This has no restore contract: the source
     // Block is left without columns and must not be used as a live output block.
@@ -586,6 +595,7 @@ class MutableBlock {
                                        dump_names(), dump_types(), block.dump_names(),
                                        block.dump_types());
             }
+            materialize_const_column(i);
             _columns[i]->insert_range_from_ignore_overflow(
                     *block.get_by_position(i).column->convert_to_full_column_if_const().get(), 0,
                     block.rows());
@@ -619,6 +629,7 @@ class MutableBlock {
                         block.dump_names(), block.dump_types());
             }
             for (int i = 0; i < _columns.size(); ++i) {
+                materialize_const_column(i);
                 if (!_data_types[i]->equals(*block.get_by_position(i).type)) {
                     DCHECK(_data_types[i]->is_nullable())
                             << " target type: " << _data_types[i]->get_name()

be/test/core/block/block_test.cpp

@@ -1145,6 +1145,37 @@ TEST(BlockTest, merge_impl) {
     EXPECT_ANY_THROW(st = mutable_block.merge_impl(std::move(block2)));
 }
 
+TEST(BlockTest, MergeMaterializesConstNullableDestination) {
+    for (bool ignore_overflow : {false, true}) {
+        auto nullable_type = make_nullable(std::make_shared<DataTypeInt64>());
+
+        auto const_data = nullable_type->create_column();
+        const_data->insert_default();
+        Block destination;
+        destination.insert(
+                {ColumnConst::create(std::move(const_data), 2), nullable_type, "lineage"});
+
+        auto source_column = nullable_type->create_column();
+        source_column->insert_default();
+        Block source;
+        source.insert({std::move(source_column), nullable_type, "lineage"});
+
+        {
+            ScopedMutableBlock scoped_destination(&destination);
+            auto status = ignore_overflow
+                                  ? scoped_destination.mutable_block().merge_ignore_overflow(source)
+                                  : scoped_destination.mutable_block().merge(source);
+            ASSERT_TRUE(status.ok()) << status.to_string();
+        }
+
+        ASSERT_EQ(destination.rows(), 3);
+        EXPECT_FALSE(is_column_const(*destination.get_by_position(0).column));
+        for (size_t i = 0; i < destination.rows(); ++i) {
+            EXPECT_TRUE(destination.get_by_position(0).column->is_null_at(i));
+        }
+    }
+}
+
 TEST(BlockTest, ctor) {
     TDescriptorTableBuilder builder;
     TTupleDescriptorBuilder tuple_builder;