...

velox/core/PlanNode.h

@@ -2260,6 +2260,10 @@ class TopNRowNumberNode : public PlanNode {
     return outputType_;
   }
 
+  bool canSpill(const QueryConfig& queryConfig) const override {
+    return !partitionKeys_.empty() && queryConfig.topNRowNumberSpillEnabled();
+  }
+
   const RowTypePtr& inputType() const {
     return sources_[0]->outputType();
   }

velox/core/QueryConfig.h

@@ -195,6 +195,10 @@ class QueryConfig {
   static constexpr const char* kRowNumberSpillEnabled =
       "row_number_spill_enabled";
 
+  /// TopNRowNumber spilling flag, only applies if "spill_enabled" flag is set.
+  static constexpr const char* kTopNRowNumberSpillEnabled =
+      "topn_row_number_spill_enabled";
+
   /// The max memory that a final aggregation can use before spilling. If it 0,
   /// then there is no limit.
   static constexpr const char* kAggregationSpillMemoryThreshold =
@@ -469,12 +473,18 @@ class QueryConfig {
     return get<bool>(kOrderBySpillEnabled, true);
   }
 
-  /// Returns 'is row_number spilling enabled' flag. Must also check the
-  /// spillEnabled()!
+  /// Returns true if spilling is enabled for RowNumber operator. Must also
+  /// check the spillEnabled()!
   bool rowNumberSpillEnabled() const {
     return get<bool>(kRowNumberSpillEnabled, true);
   }
 
+  /// Returns true if spilling is enabled for TopNRowNumber operator. Must also
+  /// check the spillEnabled()!
+  bool topNRowNumberSpillEnabled() const {
+    return get<bool>(kTopNRowNumberSpillEnabled, true);
+  }
+
   // Returns a percentage of aggregation or join input batches that
   // will be forced to spill for testing. 0 means no extra spilling.
   int32_t testingSpillPct() const {

velox/docs/configs.rst

@@ -204,6 +204,10 @@ Spilling
      - boolean
      - true
      - When `spill_enabled` is true, determines whether RowNumber operator can spill to disk under memory pressure.
+   * - topn_row_number_spill_enabled
+     - boolean
+     - true
+     - When `spill_enabled` is true, determines whether TopNRowNumber operator can spill to disk under memory pressure.
    * - aggregation_spill_memory_threshold
      - integer
      - 0

velox/exec/PlanNodeStats.cpp

@@ -102,6 +102,12 @@ std::string PlanNodeStats::toString(bool includeInputStats) const {
   if (numSplits > 0) {
     out << ", Splits: " << numSplits;
   }
+
+  if (spilledRows > 0) {
+    out << ", Spilled: " << spilledRows << " rows ("
+        << succinctBytes(spilledBytes) << ", " << spilledFiles << " files)";
+  }
+
   return out.str();
 }
 

velox/exec/RowNumber.cpp

@@ -176,7 +176,7 @@ void RowNumber::restoreNextSpillPartition() {
 }
 
 void RowNumber::ensureInputFits(const RowVectorPtr& input) {
-  if (!spillConfig_.has_value()) {
+  if (!spillEnabled()) {
     // Spilling is disabled.
     return;
   }