Implement fast min/max accumulator for binary / strings (now it uses the slower path)

[alamb]

Is your feature request related to a problem or challenge?

#6904 introduces some fancy new hashing and ways to implement aggregates

See related blog post: https://datafusion.apache.org/blog/2023/08/05/datafusion_fast_grouping/

min/max for strings (StringArray / LargeStringArray, etc) now uses the slower Accumulator implementation which could be made much faster

Describe the solution you'd like

I would like to implement a fast GroupsAccumulator for Min/Max

Describe alternatives you've considered

here is one potential way to implement it:

We could store the current minimum for all groups in the same Rows 🤔 and track an index into that Rows for the current minimum for each group.

This would require an extra copy of the input values, but it could probably be vectorized pretty well, as shown in the following diagram.

Sorry what I meant was something like the following where the accumulator only stored the current minimum values.

This approach would potentially end up with min_storage being full of "garbage" if many batches had new minumums, but I think we could heuristically "compact" min_storage (if it had 2*num_groups, for example) if it got too large

                                    ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐  
                                                                           
                                    │           Accumulator             │  
                                                state                      
┌─────────┐       ┌─────────┐       │ ┌─────────┐          ┌─────────┐  │  
│ ┌─────┐ │       │ ┌─────┐ │         │ ┌─────┐ │          │ ┌─────┐ │     
│ │  A  │ │       │ │  A  │─┼────┐  │ │ │  D  │ │    ┌─────┼─│  1  │ │  │  
│ ├─────┤ │       │ ├─────┤ │    │    │ ├─────┤ │    │     │ ├─────┤ │     
│ │  B  │ │       │ │  B  │ │    └──┼─┼▶│  A  │◀┼────┘     │ │  0  │ │  │  
│ ├─────┤ │       │ ├─────┤ │         │ └─────┘ │          │ └─────┘ │     
│ │  A  │ │       │ │  A  │ │       │ │         │          │         │  │  
│ ├─────┤ │       │ ├─────┤ │         │         │          │         │     
│ │  A  │ │       │ │  A  │ │       │ │         │          │         │  │  
│ ├─────┤ │       │ ├─────┤ │         │         │          │         │     
│ │  C  │ │       │ │  C  │ │       │ │         │          │         │  │  
│ └─────┘ │       │ └─────┘ │         │         │          │         │     
└─────────┘       └─────────┘       │ └─────────┘          └─────────┘  │  
                                                                           
  input              input          │ min_storage:         min_values   │  
  values             values           Rows                                 
  (Array)            (Rows)         └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘  
          step 1:           step 2: for                                    
          convert           any value                   step 3: min value  
          arguments to      that is a new               (per group) is     
          Row format        group                       tracked as an      
                            minimum, copy               index into         
                            it to a                     min_storage `Rows` 
                            second `Rows`                                  
                                                                           

See #6800 (comment) for more details

Additional context

No response

[alamb]

One observation here is that min and max on strings is not that common of an operation from what it seems -- grouping on strings is more common.

Maybe there is some binary usecase where it is important (e.g. embeddings 🤔 )

[devanbenz]

@alamb is there anyone working on this + is this issue still relevant? I would love to tackle it as it seems like an interesting feature/optimization.

[alamb]

@alamb is there anyone working on this + is this issue still relevant? I would love to tackle it as it seems like an interesting feature/optimization.

I dont know of anyone working on this @devanbenz -- but I also don't know of any benchmarks or actual queries that use min / max on string columns. The place it shows up is when computing statistics when writing parquet, but I think parquet is already pretty good at this (and has its own code to compute min/max)

[alamb]

It actually turns out that Min / Max on string/binary columns are in several ClickBench queries:

datafusion/benchmarks/queries/clickbench/queries.sql

Lines 22 to 23 in a08f923

	SELECT "SearchPhrase", MIN("URL"), COUNT(*) AS c FROM hits WHERE "URL" LIKE '%google%' AND "SearchPhrase" <> '' GROUP BY "SearchPhrase" ORDER BY c DESC LIMIT 10;
	SELECT "SearchPhrase", MIN("URL"), MIN("Title"), COUNT(*) AS c, COUNT(DISTINCT "UserID") FROM hits WHERE "Title" LIKE '%Google%' AND "URL" NOT LIKE '%.google.%' AND "SearchPhrase" <> '' GROUP BY "SearchPhrase" ORDER BY c DESC LIMIT 10;

datafusion/benchmarks/queries/clickbench/queries.sql

Line 29 in a08f923

SELECT REGEXP_REPLACE("Referer", '^https?://(?:www\.)?([^/]+)/.*$', '\1') AS k, AVG(length("Referer")) AS l, COUNT(*) AS c, MIN("Referer") FROM hits WHERE "Referer" <> '' GROUP BY k HAVING COUNT(*) > 100000 ORDER BY l DESC LIMIT 25;

I don't think min/max have appeared as priorities in benchmarking before because the queries in question are doing other string heavy operations that tend to dominate. Thus the use of GroupsAccumulatorAdaptor for Min/Max on strings, while bad, is overshadowed by other things

However, while working on #12092 it turns out that Min / Max on BinaryView and StringView are suuuuper slow. We can likely restore their speed to something similar to apache/arrow-rs#6408 but I also think this is a good time to actually make Min / Max on strings faster. I will write up some ideas on how to do this

[alamb]

Background

(I will make a PR shortly to add this to the actual datafusion docs)

GroupsAccumulator logically does this:

      ┌─────┐                            
      │  0  │───────────▶   "A"          
      ├─────┤                            
      │  1  │───────────▶   "Z"          
      └─────┘                            
        ...                 ...          
      ┌─────┐                            
      │ N-2 │               "A"          
      ├─────┤                            
      │ N-1 │───────────▶   "Q"          
      └─────┘                            
                                         
                                         
    Logical group      Current Min/Max   
       number          value for that    
                       group             
                                         
                                         
                                         
GroupsAccumulator to store N aggregate   
values: logically keepa a mapping from   
each group index to the current value                                        

Today, String / Binary min/max values are implemented using GroupsAccumulatorAdapter which results in

                                                              Individual String
                                                              (separate        
                                                              allocation)      
                                                                               
   ┌─────┐            ┌──────────────────────────┐                             
   │  0  │───────────▶│  ScalarValue::Utf8("A")  ├──────────▶   "A"            
   ├─────┤            ├──────────────────────────┤                             
   │  1  │───────────▶│  ScalarValue::Utf8("Z")  │──────────▶   "Z"            
   └─────┘            └──────────────────────────┘                             
     ...                 ...                                    ...            
   ┌─────┐            ┌──────────────────────────┐                             
   │ N-2 │            │  ScalarValue::Utf8("A")  │──────────▶   "A"            
   ├─────┤            ├──────────────────────────┤                             
   │ N-1 │───────────▶│  ScalarValue::Utf8("Q")  │──────────▶   "Q"            
   └─────┘            └──────────────────────────┘                             
                                                                               
                                                                               
 Logical group         Current Min/Max value for that group stored             
    number             as a ScalarValue which points to an                     
                       indivdually allocated String                            
                                                                               
                                                                               
                                                                               
   How GroupsAccumulatorAdaptor works today:                                   
   stores each current min/max as a                                            
   ScalarValue                                                                 
                                                                               

[alamb]

Potential Design

One high level idea is to build a data structure that uses the same internal format (views/buffers) as StringViewArray in Arrow:

                  ┌───────────────────────────────────────────┐  
                  │                        Stored in Vec<u8>  │  
                  │   Stored in a         ┌─────────────────┐ │  
                  │     Vec<u128>         │ ┌───────────┐   │ │  
                  │ ┌─────────────┐    ─ ─│▶│some value │   │ │  
   ┌─────┐        │ │ ┌─────────┐ │   │   │ └───────────┘   │ │  
   │  0  │────────┼─┼▶│  View   │ │       │                 │ │  
   ├─────┤        │ │ ├─────────┤ │   │   │                 │ │  
   │  1  │────────┼─┼▶│  View   │─│─ ─ ┐  │      ...        │ │  
   └─────┘        │ │ └─────────┘ │   │   │                 │ │  
     ...          │ │    ...      │    │  │                 │ │  
   ┌─────┐        │ │ ┌─────────┐ │   │   │                 │ │  
   │ N-2 │        │ │ │  View   │─│─ ─ │  │   ┌────────────┐│ │  
   ├─────┤        │ │ ├─────────┤ │     ─ ┼ ─▶│other value ││ │  
   │ N-1 │────────┼─┼▶│  View   │ │       │   └────────────┘│ │  
   └─────┘        │ │ └─────────┘ │       └─────────────────┘ │  
                  │ └─────────────┘  String values are stored │  
                  │                  inline or in extra byte  │  
 Logical group    │                  buffer                   │  
    number        └───────────────────────────────────────────┘  
                   New structure: MutableStringViewBuilder       
                                                                 
                   Current Min/Max value for that group stored in
                   same format as StringViewArray                
                                                                 

In this design, the current value for each group is stored in two parts (as described on arrow docs)

1. a fixed size u128 view
2. a variable length part with the string data

As new batches are updated, each View is updated if necessary

Benefits of design:

1. Hopefully use the same code as in arrow-rs
2. Allows Zero copy conversion to StringView / BinaryView at output
3. Use inlined values for quick min/max comparison

I believe (though we will have to verify it) that the conversion from MutableStringViewBuilder to just StringArray (not StringViewArray) should also be better than the current GroupsAccumulatorAdapter. Both conversions need to copy the string bytes again into the packed StringArray format, but the GroupsAccumulatorAdapter also has to allocate/free owned Strings as well

Potential challenges

I think the trickiest part of this code, other the low level code optimizations is that as min/max values are replaced, data in the variable length buffer will become "garbage" (not reachable) thus consuming more memory than necessary:

                        Stored in Vec<u8>   
                     ┌────────────────────┐ 
                     │ ┌────────────────┐ │ 
┌─────────────┐      │ │prev max value 1│ │ 
│ ┌─────────┐ │      │ └────────────────┘ │ 
│ │  View   │─│─ ┐   │        ...         │ 
│ └─────────┘ │      │ ┌────────────────┐ │ 
│     ...     │  │   │ │prev max value m│ │ 
│             │      │ └────────────────┘ │ 
│             │  │   │ ┌────────────────┐ │ 
│             │   ─ ─│▶│prev max value m│ │ 
│             │      │ └────────────────┘ │ 
│             │      │        ...         │ 
│             │      └────────────────────┘ 
│             │                             
└─────────────┘      Previous min/max values
                     are not pointed to     
                     anymore and need to be 
                     cleaned up             

I think this means the code will need something GenericByteViewArray::gc run occasionally

Random Thoughts

Thoughts: maybe this structure (MutableStringViewBuilder??) could be upstreamed eventually

[alamb]

@alamb is there anyone working on this + is this issue still relevant? I would love to tackle it as it seems like an interesting feature/optimization.

@devanbenz I think this would be a fun and interesting project as well as valuable to DataFusion. However, I also think it is pretty advanced -- I would enjoy helping with it, but also maybe @Rachelint or @jayzhan211 are interested in helping out 🤔