[spill](regression) add regression test cases for spill (#37319)

## Proposed changes

Issue Number: close #xxx

<!--Describe your changes.-->

be/src/pipeline/exec/aggregation_sink_operator.cpp

@@ -748,7 +748,6 @@ Status AggSinkOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
     }
 
     const auto& agg_functions = tnode.agg_node.aggregate_functions;
-    _external_agg_bytes_threshold = state->external_agg_bytes_threshold();
 
     _is_merge = std::any_of(agg_functions.cbegin(), agg_functions.cend(),
                             [](const auto& e) { return e.nodes[0].agg_expr.is_merge_agg; });

be/src/pipeline/exec/aggregation_sink_operator.h

@@ -189,7 +189,6 @@ class AggSinkOperatorX final : public DataSinkOperatorX<AggSinkLocalState> {
     /// The total size of the row from the aggregate functions.
     size_t _total_size_of_aggregate_states = 0;
 
-    size_t _external_agg_bytes_threshold;
     // group by k1,k2
     vectorized::VExprContextSPtrs _probe_expr_ctxs;
     ObjectPool* _pool = nullptr;

be/src/runtime/runtime_state.h

@@ -344,8 +344,6 @@ class RuntimeState {
         return _query_options.disable_stream_preaggregations;
     }
 
-    bool enable_spill() const { return _query_options.enable_spilling; }
-
     int32_t runtime_filter_wait_time_ms() const {
         return _query_options.runtime_filter_wait_time_ms;
     }
@@ -511,12 +509,6 @@ class RuntimeState {
 
     void set_be_exec_version(int32_t version) noexcept { _query_options.be_exec_version = version; }
 
-    int64_t external_agg_bytes_threshold() const {
-        return _query_options.__isset.external_agg_bytes_threshold
-                       ? _query_options.external_agg_bytes_threshold
-                       : 0;
-    }
-
     inline bool enable_delete_sub_pred_v2() const {
         return _query_options.__isset.enable_delete_sub_predicate_v2 &&
                _query_options.enable_delete_sub_predicate_v2;
@@ -589,9 +581,9 @@ class RuntimeState {
 
     int64_t min_revocable_mem() const {
         if (_query_options.__isset.min_revocable_mem) {
-            return _query_options.min_revocable_mem;
+            return std::max(_query_options.min_revocable_mem, (int64_t)1);
         }
-        return 0;
+        return 1;
     }
 
     void set_max_operator_id(int max_operator_id) { _max_operator_id = max_operator_id; }

be/src/vec/spill/spill_stream.cpp

@@ -119,10 +119,10 @@ Status SpillStream::spill_eof() {
     DBUG_EXECUTE_IF("fault_inject::spill_stream::spill_eof", {
         return Status::Error<INTERNAL_ERROR>("fault_inject spill_stream spill_eof failed");
     });
-    RETURN_IF_ERROR(writer_->close());
+    auto status = writer_->close();
     total_written_bytes_ = writer_->get_written_bytes();
     writer_.reset();
-    return Status::OK();
+    return status;
 }
 
 Status SpillStream::read_next_block_sync(Block* block, bool* eos) {

fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java

@@ -311,9 +311,6 @@ private PlanNode createQueryPlan(QueryStmt stmt, Analyzer analyzer, long default
             // TODO: External sort could be used for very large limits
             // not just unlimited order-by
             boolean useTopN = true;
-            if (limit == -1 && analyzer.getContext().getSessionVariable().enableSpilling) {
-                useTopN = false;
-            }
             root = new SortNode(ctx.getNextNodeId(), root, stmt.getSortInfo(),
                     useTopN);
             ((SortNode) root).setDefaultLimit(limit == -1);

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -139,7 +139,6 @@ public class SessionVariable implements Serializable, Writable {
     public static final String MAX_INSTANCE_NUM = "max_instance_num";
     public static final String ENABLE_INSERT_STRICT = "enable_insert_strict";
     public static final String INSERT_MAX_FILTER_RATIO = "insert_max_filter_ratio";
-    public static final String ENABLE_SPILLING = "enable_spilling";
 
     public static final String ENABLE_SERVER_SIDE_PREPARED_STATEMENT = "enable_server_side_prepared_statement";
     public static final String MAX_PREPARED_STMT_COUNT = "max_prepared_stmt_count";
@@ -537,7 +536,6 @@ public class SessionVariable implements Serializable, Writable {
 
     // for spill to disk
     public static final String EXTERNAL_SORT_BYTES_THRESHOLD = "external_sort_bytes_threshold";
-    public static final String EXTERNAL_AGG_BYTES_THRESHOLD = "external_agg_bytes_threshold";
     public static final String EXTERNAL_AGG_PARTITION_BITS = "external_agg_partition_bits";
     public static final String SPILL_STREAMING_AGG_MEM_LIMIT = "spill_streaming_agg_mem_limit";
     public static final String MIN_REVOCABLE_MEM = "min_revocable_mem";
@@ -695,9 +693,6 @@ public class SessionVariable implements Serializable, Writable {
     })
     public double scannerScaleUpRatio = 0;
 
-    @VariableMgr.VarAttr(name = ENABLE_SPILLING)
-    public boolean enableSpilling = false;
-
     // By default, the number of Limit items after OrderBy is changed from 65535 items
     // before v1.2.0 (not included), to return all items by default
     @VariableMgr.VarAttr(name = DEFAULT_ORDER_BY_LIMIT)
@@ -2029,12 +2024,6 @@ public void setIgnoreShapePlanNodes(String ignoreShapePlanNodes) {
             checker = "checkExternalSortBytesThreshold", varType = VariableAnnotation.DEPRECATED)
     public long externalSortBytesThreshold = 0;
 
-    // Set to 0 to disable; min: 128M
-    public static final long MIN_EXTERNAL_AGG_BYTES_THRESHOLD = 134217728;
-    @VariableMgr.VarAttr(name = EXTERNAL_AGG_BYTES_THRESHOLD,
-            checker = "checkExternalAggBytesThreshold", fuzzy = true, varType = VariableAnnotation.DEPRECATED)
-    public long externalAggBytesThreshold = 0;
-
     // The memory limit of streaming agg when spilling is enabled
     // NOTE: streaming agg operator will not spill to disk.
     @VariableMgr.VarAttr(name = SPILL_STREAMING_AGG_MEM_LIMIT, fuzzy = true)
@@ -3386,14 +3375,6 @@ public void checkExternalSortBytesThreshold(String externalSortBytesThreshold) {
         }
     }
 
-    public void checkExternalAggBytesThreshold(String externalAggBytesThreshold) {
-        long value = Long.valueOf(externalAggBytesThreshold);
-        if (value > 0 && value < MIN_EXTERNAL_AGG_BYTES_THRESHOLD) {
-            LOG.warn("external agg bytes threshold: {}, min: {}", value, MIN_EXTERNAL_AGG_BYTES_THRESHOLD);
-            throw new UnsupportedOperationException("minimum value is " + MIN_EXTERNAL_AGG_BYTES_THRESHOLD);
-        }
-    }
-
     public void checkExternalAggPartitionBits(String externalAggPartitionBits) {
         int value = Integer.valueOf(externalAggPartitionBits);
         if (value < MIN_EXTERNAL_AGG_PARTITION_BITS || value > MAX_EXTERNAL_AGG_PARTITION_BITS) {
@@ -3571,8 +3552,6 @@ public TQueryOptions toThrift() {
             tResult.setMaxPushdownConditionsPerColumn(maxPushdownConditionsPerColumn);
         }
 
-        tResult.setEnableSpilling(enableSpilling);
-
         tResult.setRuntimeFilterWaitTimeMs(runtimeFilterWaitTimeMs);
         tResult.setRuntimeFilterMaxInNum(runtimeFilterMaxInNum);
         tResult.setRuntimeBloomFilterMinSize(runtimeBloomFilterMinSize);
@@ -3605,8 +3584,6 @@ public TQueryOptions toThrift() {
 
         tResult.setSpillStreamingAggMemLimit(spillStreamingAggMemLimit);
 
-        tResult.setExternalAggPartitionBits(externalAggPartitionBits);
-
         tResult.setEnableFileCache(enableFileCache);
 
         tResult.setEnablePageCache(enablePageCache);

regression-test/data/tpcds_sf1_unique_p1/spill/q01.out

@@ -0,0 +1,103 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !q01 --
+AAAAAAAAAAABBAAA
+AAAAAAAAAAADBAAA
+AAAAAAAAAAADBAAA
+AAAAAAAAAAAKAAAA
+AAAAAAAAAABDAAAA
+AAAAAAAAAABHBAAA
+AAAAAAAAAABLAAAA
+AAAAAAAAAABMAAAA
+AAAAAAAAAACHAAAA
+AAAAAAAAAACMAAAA
+AAAAAAAAAADDAAAA
+AAAAAAAAAADGAAAA
+AAAAAAAAAADGBAAA
+AAAAAAAAAADGBAAA
+AAAAAAAAAADPAAAA
+AAAAAAAAAAEBAAAA
+AAAAAAAAAAEFBAAA
+AAAAAAAAAAEGBAAA
+AAAAAAAAAAEIAAAA
+AAAAAAAAAAEMAAAA
+AAAAAAAAAAFAAAAA
+AAAAAAAAAAFPAAAA
+AAAAAAAAAAGGBAAA
+AAAAAAAAAAGHBAAA
+AAAAAAAAAAGJAAAA
+AAAAAAAAAAGMAAAA
+AAAAAAAAAAHEBAAA
+AAAAAAAAAAHFBAAA
+AAAAAAAAAAIEBAAA
+AAAAAAAAAAJGBAAA
+AAAAAAAAAAJHBAAA
+AAAAAAAAAAKCAAAA
+AAAAAAAAAAKCAAAA
+AAAAAAAAAAKJAAAA
+AAAAAAAAAAKMAAAA
+AAAAAAAAAAKMAAAA
+AAAAAAAAAALAAAAA
+AAAAAAAAAALABAAA
+AAAAAAAAAALGAAAA
+AAAAAAAAAALHBAAA
+AAAAAAAAAALJAAAA
+AAAAAAAAAANHAAAA
+AAAAAAAAAANHBAAA
+AAAAAAAAAANJAAAA
+AAAAAAAAAANMAAAA
+AAAAAAAAAANMAAAA
+AAAAAAAAAANNAAAA
+AAAAAAAAAAOBBAAA
+AAAAAAAAAAODBAAA
+AAAAAAAAAAOLAAAA
+AAAAAAAAAAPGBAAA
+AAAAAAAAABAAAAAA
+AAAAAAAAABAEAAAA
+AAAAAAAAABAEBAAA
+AAAAAAAAABAFBAAA
+AAAAAAAAABAIAAAA
+AAAAAAAAABAOAAAA
+AAAAAAAAABBDBAAA
+AAAAAAAAABCFAAAA
+AAAAAAAAABCHBAAA
+AAAAAAAAABDHAAAA
+AAAAAAAAABENAAAA
+AAAAAAAAABFEBAAA
+AAAAAAAAABFGAAAA
+AAAAAAAAABFMAAAA
+AAAAAAAAABFPAAAA
+AAAAAAAAABGFAAAA
+AAAAAAAAABGFBAAA
+AAAAAAAAABGJAAAA
+AAAAAAAAABIBBAAA
+AAAAAAAAABICBAAA
+AAAAAAAAABIIAAAA
+AAAAAAAAABJNAAAA
+AAAAAAAAABKGBAAA
+AAAAAAAAABLOAAAA
+AAAAAAAAABLPAAAA
+AAAAAAAAABMABAAA
+AAAAAAAAABMPAAAA
+AAAAAAAAABNAAAAA
+AAAAAAAAABNCBAAA
+AAAAAAAAABNEBAAA
+AAAAAAAAABNLAAAA
+AAAAAAAAABNOAAAA
+AAAAAAAAABNPAAAA
+AAAAAAAAABOAAAAA
+AAAAAAAAABOFBAAA
+AAAAAAAAABOOAAAA
+AAAAAAAAABOPAAAA
+AAAAAAAAABPEAAAA
+AAAAAAAAACADAAAA
+AAAAAAAAACAFAAAA
+AAAAAAAAACAFAAAA
+AAAAAAAAACAHBAAA
+AAAAAAAAACAJAAAA
+AAAAAAAAACBDAAAA
+AAAAAAAAACBDAAAA
+AAAAAAAAACBEBAAA
+AAAAAAAAACBNAAAA
+AAAAAAAAACBPAAAA
+AAAAAAAAACCHAAAA
+