[bug](analytic) fix column string overflow in analytic operator

Author: zhangstar333

be/src/pipeline/exec/analytic_sink_operator.cpp

@@ -81,8 +81,8 @@ Status AnalyticSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& inf
             if (b.__isset.rows_offset_value) { //[offset     ,   ]
                 _rows_start_offset = b.rows_offset_value;
                 if (b.type == TAnalyticWindowBoundaryType::PRECEDING) {
-                    _rows_start_offset *= -1;                                //preceding--> negative
-                }                                                            //current_row  0
+                    _rows_start_offset *= -1; //preceding--> negative
+                } //current_row  0
             } else {                                                         //following    positive
                 DCHECK_EQ(b.type, TAnalyticWindowBoundaryType::CURRENT_ROW); //[current row,   ]
                 _rows_start_offset = 0;
@@ -784,6 +784,9 @@ Status AnalyticSinkOperatorX::_add_input_block(doris::RuntimeState* state,
 }
 
 void AnalyticSinkLocalState::_remove_unused_rows() {
+    // test column overflow 4G
+    DBUG_EXECUTE_IF("AnalyticSinkLocalState._remove_unused_rows", { return; });
+
     const size_t block_num = 256;
     if (_removed_block_index + block_num + 1 >= _input_block_first_row_positions.size()) {
         return;
@@ -845,7 +848,9 @@ Status AnalyticSinkOperatorX::_insert_range_column(vectorized::Block* block,
     RETURN_IF_ERROR(expr->execute(block, &result_col_id));
     DCHECK_GE(result_col_id, 0);
     auto column = block->get_by_position(result_col_id).column->convert_to_full_column_if_const();
-    dst_column->insert_range_from(*column, 0, length);
+    // iff dst_column is string, maybe overflow of 4G, so need ignore overflow
+    // the column is used by compare_at self to find the range, it's need convert it when overflow?
+    dst_column->insert_range_from_ignore_overflow(*column, 0, length);
     return Status::OK();
 }
 

regression-test/data/query_p0/sql_functions/window_functions/test_column_boundary.out

@@ -1,4 +1,7 @@
 -- This file is automatically generated. You should know what you did if you want to edit this
 -- !sql_1 --
-251000000
+270000000
+
+-- !sql_2 --
+69324304568
 

regression-test/suites/query_p0/sql_functions/window_functions/test_column_boundary.groovy

@@ -15,12 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-suite("test_column_boundary") {
+suite("test_column_boundary","nonConcurrent") {
     sql """ DROP TABLE IF EXISTS test_column_boundary """
     sql """
         CREATE TABLE IF NOT EXISTS test_column_boundary (
             u_id int NULL COMMENT "",
-            u_city varchar(20) NULL COMMENT ""
+            u_city varchar(40) NULL COMMENT ""
         ) ENGINE=OLAP
         DUPLICATE KEY(`u_id`, `u_city`)
         DISTRIBUTED BY HASH(`u_id`, `u_city`) BUCKETS 1
@@ -31,24 +31,40 @@ suite("test_column_boundary") {
     );
     """
 
-    sql """ insert into test_column_boundary select number, number + random() from numbers("number" = "1000000"); """
+    sql """ DROP TABLE IF EXISTS test_column_boundary2 """
+    sql """
+        CREATE TABLE IF NOT EXISTS test_column_boundary2 (
+            u_id int NULL COMMENT "",
+            u_city varchar(40) NULL COMMENT ""
+        ) ENGINE=OLAP
+        DUPLICATE KEY(`u_id`, `u_city`)
+        DISTRIBUTED BY HASH(`u_id`, `u_city`) BUCKETS 1
+        PROPERTIES (
+        "replication_allocation" = "tag.location.default: 1",
+        "in_memory" = "false",
+        "storage_format" = "V2"
+    );
+    """
+
+    sql """ insert into test_column_boundary2 select number, number + random() from numbers("number" = "1000000"); """
     Integer count = 0;
-    Integer maxCount = 25;
+    Integer maxCount = 270;
     while (count < maxCount) {
-        sql """ insert into test_column_boundary select number, number + random() from numbers("number" = "10000000"); """
+        log.info("count: ${count}")
+        sql """ insert into test_column_boundary select * from test_column_boundary2; """
         count++
         sleep(100);
     }
     sql """ set parallel_pipeline_task_num = 1; """
 
     qt_sql_1 """ select count() from test_column_boundary; """ // 256000000 rows
-    test {
-        // column size is too large
-        sql """ select sum(res) from (select count() over(partition by u_city) as res from test_column_boundary) as t; """
-        exception "string column length is too large"
+
+    try {
+        GetDebugPoint().enableDebugPointForAllBEs("AnalyticSinkLocalState._remove_unused_rows")
+        // before column size will be too large 
+        qt_sql_2 """ select sum(res) from (select count() over(partition by u_city) as res from test_column_boundary) as t; """
+    } finally {
+        GetDebugPoint().disableDebugPointForAllBEs("AnalyticSinkLocalState._remove_unused_rows")
     }
     sql """ DROP TABLE IF EXISTS test_column_boundary """
-}
-
-
-
+}