feat: Add spill for MarkDistinct (#16577)

Summary:
Pull Request resolved: #16577

### Added Spill for MarkDistinct

This diff adds a new feature to MarkDistinct to allow spilling.

Here are the highlights of the change:

### Changed Files include
* GroupingSet.h
* core/QueryConfig.h
*  exec/EnforceDistinct.cpp
* exec/GroupingSet.cpp
* core/PlanNode.h in addition to configured spilling flag in QueryConfig.h
* Added SpillConfig to GroupingSet.h’s constructor to allow spilling in GroupingSet.cpp
* Added a canSpill() function to MarkDistinct’s PlanNode to enforce spilling support
In addition to updating all GroupingSet.h’s function

Added spilling support for operator MarkDistinct

disabled by default

Reviewed By: xiaoxmeng, tanjialiang

Differential Revision: D90342763

fbshipit-source-id: 184841f9af7eca47e44a5394a86238c9dfa1c4ed

Author: duxiao1212

velox/core/PlanNode.h

@@ -5419,6 +5419,10 @@ class MarkDistinctNode : public PlanNode {
     return "MarkDistinct";
   }
 
+  bool canSpill(const QueryConfig& queryConfig) const override {
+    return queryConfig.markDistinctSpillEnabled();
+  }
+
   const std::string& markerName() const {
     return markerName_;
   }

velox/core/QueryConfig.h

@@ -368,6 +368,10 @@ class QueryConfig {
   static constexpr const char* kRowNumberSpillEnabled =
       "row_number_spill_enabled";
 
+  /// MarkDistinct spilling flag, only applies if "spill_enabled" flag is set.
+  static constexpr const char* kMarkDistinctSpillEnabled =
+      "mark_distinct_spill_enabled";
+
   /// TopNRowNumber spilling flag, only applies if "spill_enabled" flag is set.
   static constexpr const char* kTopNRowNumberSpillEnabled =
       "topn_row_number_spill_enabled";
@@ -1185,6 +1189,10 @@ class QueryConfig {
     return get<bool>(kRowNumberSpillEnabled, true);
   }
 
+  bool markDistinctSpillEnabled() const {
+    return get<bool>(kMarkDistinctSpillEnabled, true);
+  }
+
   bool topNRowNumberSpillEnabled() const {
     return get<bool>(kTopNRowNumberSpillEnabled, true);
   }

velox/docs/configs.rst

@@ -413,6 +413,10 @@ Spilling
      - boolean
      - true
      - When `spill_enabled` is true, determines whether TopNRowNumber operator can spill to disk under memory pressure.
+   * - mark_distinct_spill_enabled
+     - boolean
+     - false
+     - When `spill_enabled` is true, determines whether MarkDistinct operator can spill to disk under memory pressure.
    * - writer_spill_enabled
      - boolean
      - true

velox/docs/develop/memory.rst

@@ -667,7 +667,7 @@ Here is the memory reclaim process within a query:
    reclamation through disk spilling and table writer flush. *Operator::reclaim*
    is added to support memory reclamation with the default implementation does
    nothing. Only spillable operators override that method: *OrderBy*, *HashBuild*,
-   *HashAggregation*, *RowNumber*, *TopNRowNumber*, *Window* and *TableWriter*.
+   *HashAggregation*, *RowNumber*, *TopNRowNumber*, *MarkDistinct*, *Window* and *TableWriter*.
    As for now, we simply spill everything from the spillable operator’s row
    container to free up memory. After we add memory compaction support for row
    containers, we could leverage fine-grained disk spilling features in Velox

velox/docs/develop/operators.rst

@@ -1018,6 +1018,11 @@ MarkDistinctNode
 The MarkDistinct operator is used to produce aggregate mask columns for aggregations over distinct values, e.g. agg(DISTINCT a).
 Mask is a boolean column set to true for a subset of input rows that collectively represent a set of unique values of 'distinctKeys'.
 
+This operator supports spilling. The spill mechanism follows the same pattern as RowNumber: when memory pressure
+triggers spilling, the hash table contents and future input are partitioned and written to disk. During restore,
+each partition's hash table is rebuilt from the spilled data, preserving knowledge of which keys were already seen.
+Disabled by default; enable with `mark_distinct_spill_enabled` configuration property.
+
 .. list-table::
   :widths: 10 30
   :align: left

velox/exec/GroupingSet.cpp

@@ -276,7 +276,7 @@ void GroupingSet::addInputForActiveRows(
     const RowVectorPtr& input,
     bool mayPushdown) {
   VELOX_CHECK(!isGlobal_);
-  if (!table_) {
+  if (table_ == nullptr) {
     createHashTable();
   }
   ensureInputFits(input);

velox/exec/GroupingSet.h

@@ -149,6 +149,12 @@ class GroupingSet {
     return table_ ? table_->rows()->numRows() : 0;
   }
 
+  /// Returns the underlying hash table, or nullptr if it has not been created
+  /// yet.
+  BaseHashTable* table() const {
+    return table_.get();
+  }
+
   /// Frees hash tables and other state when giving up partial aggregation as
   /// non-productive. Must be called before toIntermediate() is used.
   void abandonPartialAggregation();