[VL] Spill related issues tracker

[zhztheplayer]

Description

Mirror issue in facebookincubator/velox facebookincubator/velox#6414

This is to list the large memory occupations that are not spillable so far, which means, that are not able to be spilled to disk, in Velox backend's query execution.

Technically the listed items should be all finally fixed ("fix" means to make then spillable), to ensure the memory stability of Gluten. Otherwise there would be chance that OOM error raises during execution that would fail the user query.

The list of non-spillable large occupations (attach PR following each item once fixing):

• Buffered inputs from Velox's window operator
  • Streaming window
    • Streaming window build
    • Streaming window functions without build, not planned in Velox yet
  • Spillable sort window
• Buffered inputs from Velox's hash-aggregate operator, when aggregate is distinct aggregate
• Buffered inputs from Velox's hash-aggregate operator, when aggregate is partial aggregate (needs confirmation)
• Buffered input in Velox's hash-aggregate/hash-join(build)/sort operator, after all input is added
  • Hash-aggregate
  • Hash-join(build) Velox community is working on this now.
  • Sort
• Pre-allocate split buffers from Gluten's Velox shuffle writer
• A task can take use executor's memory if no other task running in the executor, TPCDS Q67. Vanilla spark does this
• External sort in fallbacked partition write ( can’t be triggered by gluten)

[winningsix]

Before we got spill support from Velox, what's our current plan?

From initial PR, it seems we want to introduce a mem cap for each task attempt?

[Yohahaha]

I have same question with @winningsix , we introduce over-acquire concept before to hold more memory reservations from Spark as buffer to try avoid OOM, and #3101 seems introduce memory limit for each Spark task to try avoid OOM, does these two feature are exclusive or not?

If these Velox operators are still non-spillable, and all available bytes are fixed, operator's used bytes are fixed in specific query, I doubt how much benefits could gains from above features, is there a case that OOM before but success after enable these features?

[Yohahaha]

CC @liujiayi771

[FelixYBW]

update:

1. Buffered inputs from Velox's window operator
2. Buffered inputs from Velox's hash-aggregate operator, when aggregate is distinct aggregate
   Velox PR created: Add spilling support for distinct hash aggregation facebookincubator/velox#3263
3. Buffered inputs from Velox's hash-aggregate operator, when aggregate is partial aggregate (needs confirmation)
   Currently Velox flush the partial agg once OOM
4. Buffered input in Velox's hash-aggregate/hash-join(build)/sort operator, after all input is added
   hashagg support in Velox is done: Add spill support during aggregation output stage facebookincubator/velox#6903
5. Pre-allocate split buffers from Gluten's Velox shuffle writer
   couple of shuffle writer modifications listed here:
   • Merged:
   [VL] Realloc shuffle split buffer if beyond avg threshold #2982 Dynamically adjust split buffer size.
   [VL] Get avaliable off-heap memory for split buffer calculation everytime split() is called #3036 Get avaliable off-heap memory for split buffer calculation everytime split() is called
   [GLUTEN-3189] Fix shuffle spill failure and wrong result #3199 Continuation of [VL] Realloc shuffle split buffer if beyond avg threshold #2982. Bug fix & add UT.
   [VL] Add memory pool to track shuffle cached memory usage #3159 Track memory allocation of split buffer and cached payload separately.
   [VL] Remove preferSpill=true for VeloxShuffleWriter #3091 Remove preferSpill=True.
   [VL] Shrink min sized partition buffers and spill #3265 Shrink minimum partition buffer size and add spill support for partition buffers.
   [VL] Refactor split buffer allocation #3177 Refactor split buffer allocation. (Only code refactor, no functional change)
   • WIP:
   [VL] Shrink min sized partition buffers and spill #3265 Shrink min sized partition buffers and spill

[FelixYBW]

Another fix related:

6. release all previous operator's memory when shufflewriter's stop is called. Because when shuffle write's stop is called all batches are processed. So the shuffle writer can compress the cached batches and write to page cache.
   PRS:
   [VL] Iterator's and its payloads' lifecycle improvements #3526

[FelixYBW]

update:

1. Buffered inputs from Velox's window operator
2. Buffered inputs from Velox's hash-aggregate operator, when aggregate is distinct aggregate
   Supported
3. Buffered inputs from Velox's hash-aggregate operator, when aggregate is partial aggregate (needs confirmation)
   Currently Velox flush the partial agg once OOM
4. Buffered input in Velox's hash-aggregate/hash-join(build)/sort operator, after all input is added
   hashagg supported
5. Pre-allocate split buffers from Gluten's Velox shuffle writer
6. release all previous operator's memory when shufflewriter's stop is called. Because when shuffle write's stop is called all batches are processed. So the shuffle writer can compress the cached batches and write to page cache.
   Supported

[XinShuoWang]

@FelixYBW @zhztheplayer @zhouyuan
Hi, can you give me more details about 5. Pre-allocate split buffers from Gluten's Velox shuffle writer? Like minimal reproducible example or related documents?

[FelixYBW]

Added a feature that a task can take use whole executor's memory, or it use other tasks' memory if other tasks are idle. It's what Vanilla spark does today. In Gluten we disabled the feature since OOM issue isn't solved. @zhztheplayer

[FelixYBW]

@zhztheplayer buffered input for sort is already supported, right?

Buffered input in Velox's hash-aggregate/hash-join(build)/sort operator, after all input is added
Sort

[FelixYBW]

we removed Velox's partial Agg in Velox, use final agg instead because Vanilla spark's partial agg have the same behavior of Velox's final agg. In Vanilla Spark it doesn't support early eviction, instead it spill the data and does full agg in partial agg, it can make sure the agg key is unique in partial agg. On the otherside Velox's partial agg use earl eviction once memory is not enough, it doesn't make sure the agg key is unique in partial agg.

Since we use both final agg, spill is fully supported now