[fix](agg)the output of window function's nullability should be consistent with output slot (#12607)

FE may force window function to output a nullable value in some case, be should follow this and change the nullability accordingly.

be/src/vec/exec/vanalytic_eval_node.cpp

@@ -160,6 +160,8 @@ Status VAnalyticEvalNode::prepare(RuntimeState* state) {
         SlotDescriptor* output_slot_desc = _output_tuple_desc->slots()[i];
         RETURN_IF_ERROR(_agg_functions[i]->prepare(state, child(0)->row_desc(), _mem_pool.get(),
                                                    intermediate_slot_desc, output_slot_desc));
+        _change_to_nullable_flags.push_back(output_slot_desc->is_nullable() &&
+                                            !_agg_functions[i]->data_type()->is_nullable());
     }
 
     _offsets_of_aggregate_states.resize(_agg_functions_size);
@@ -572,8 +574,15 @@ Status VAnalyticEvalNode::_output_current_block(Block* block) {
         block->erase_not_in(_origin_cols);
     }
 
+    DCHECK(_change_to_nullable_flags.size() == _result_window_columns.size());
     for (size_t i = 0; i < _result_window_columns.size(); ++i) {
-        block->insert({std::move(_result_window_columns[i]), _agg_functions[i]->data_type(), ""});
+        if (_change_to_nullable_flags[i]) {
+            block->insert({make_nullable(std::move(_result_window_columns[i])),
+                           make_nullable(_agg_functions[i]->data_type()), ""});
+        } else {
+            block->insert(
+                    {std::move(_result_window_columns[i]), _agg_functions[i]->data_type(), ""});
+        }
     }
 
     _output_block_index++;

be/src/vec/exec/vanalytic_eval_node.h

@@ -144,5 +144,7 @@ private:
     std::vector<int64_t> _origin_cols;
 
     RuntimeProfile::Counter* _evaluation_timer;
+
+    std::vector<bool> _change_to_nullable_flags;
 };
 } // namespace doris::vectorized

regression-test/data/correctness_p0/test_no_grouping_window.out

@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+23.0
+

regression-test/suites/correctness_p0/test_no_grouping_window.groovy

@@ -0,0 +1,45 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_no_grouping_window") {
+    sql """
+        drop table if exists outerjoin_A;
+    """
+
+    sql """
+        create table outerjoin_A ( a int not null )
+        ENGINE=OLAP
+        DISTRIBUTED BY HASH(a) BUCKETS 1
+        PROPERTIES (
+        "replication_allocation" = "tag.location.default: 1",
+        "in_memory" = "false",
+        "storage_format" = "V2"
+        );
+    """
+
+    sql """
+        insert into outerjoin_A values( 1 );
+    """
+
+    qt_select """
+        select avg( a ) over ( partition by a order by a ) * 23 from outerjoin_A;
+    """
+
+    sql """
+        drop table if exists outerjoin_A;
+    """
+}