[Epic] Implement `RunArray` (Run Length Encoding (RLE) / Run End Encoding (REE) support)

[alamb]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
Arrow has added REE support apache/arrow#14176, similar to dictionary arrays that allow repeated values to be encoded in a space efficient manner that also allows fast processing.

Describe the solution you'd like
Implement REE in arrow-rs. Some likely candidate:

• Support in DataType
• Support in ArrayData
• New REE array
• Support REE in IPC
• Support REE in cast kernels: Cast support for RunEndEncoded arrays #8589
• Support REE in compute kernels
• Support converting RunEndEncodedType to parquet #8016
• Implements Sum,sum_checked,min,max,is Distict,inverse for REE.  #7933
• Implement String functions for REE #8019

Remaining tasks:

• arrow-row: Add support for REE #7649
• arrow-select: Implement concat for RunArrays #7487
• arrow-data: Add REE support for build_extend and build_extend_nulls #7671
• Implement PartialEq for RunArray #7691
• Reduce repetition in tests for arrow-row/src/run.rs #7692
• Improve performance of RunArray --> Row conversion #7693
• Potential Optimization for interleave/take on RunEndEncoded arrays #7710
• Cast support for RunEndEncoded arrays #8589
• Improve RunArray documentation #9019
• Kernels don't account for slicing of RunArrays #9018
• Implement logical comparison for run encoded array #3747
• Improve performance of RunEndEncoded cast #8707
• PrimitiveRunBuilder does not support augmenting its data_type. #8042
• Support RunEndEncoded arrays in arrow-json #9359
• Support RunEndEncoded in ord comparator #9360

Additional context

Among other things, @brancz is working to improve aggregation performance in DataFusion using Runarrays, see

• Support Aggregating by RunArrays datafusion#16011

[tustvold]

One thing to perhaps give thought to is what kernels we need to facilitate this use-case. For example, a core pattern for dictionaries is to evaluate against the child values array and then take the results based on the keys. This avoids generating code for each combination of key and value type. We possibly need something similar for the REE use-case, i.e. some sort of take_runs kernel.

[askoa]

I would like to start this by writing REE Array.

[alippai]

Would be adding parquet support in the scope? I’m not sure if parquet has RLE directly, but highly compressed (zstd,snappy,zlib) values can be converted to REE arrays instead of inflating.

[tustvold]

Dictionaries are RLE encoded in parquet, so we could theoretically preserve RLE encoding for dictionary arrays, on top of the existing logic to preserve the dictionary encoding. However, the cost and complexity of translating between the two representations is likely to negate much of the potential gains.

zlib

Do you have a link for this, I'm not sure how you would reliably use the codec's internal RLE coding as it isn't guaranteed to be at a meaningful granularity for the encoded data?

[alippai]

@tustvold I agree that the reading performance might not improve but the downstream operations like filter/join/group by (if optimized for REE) would definitely make it worth. Absolutely not a day 0 feature.

The zlib/zstd idea is far fetched, but:
1.) zlib sometimes decides to store data as Z_RLE, might worth checking if this happens with Parquet too: https://optipng.sourceforge.net/pngtech/z_rle.html
2.) I was wondering if LZ77 -> REE is possible without fully decompressing the data, something like posted here (for a different algorithm): https://www.researchgate.net/publication/261072335_From_Run_Length_Encoding_to_LZ78_and_Back_Again

Both of these are super theoretical and needs access to the compression building blocks, the low level api (as you don’t want to ask the lib to decompress the data, but you want to go compressed -> REE to save big)

[askoa]

I am starting to write Iterator / ArrayAccessor for RunEndEncodedArray. I could see two scenarios for accessing the values

1. The caller know the physical index and just wants value in that physical index.
   • This method is useful when implementing iterators. The iterator can decide if the physical index has to be increased in O(1) time. After the decision is made, the iterator needs to access the value in the physical index.
2. The caller does not know the physical index and trying to access a value in logical index. A binary search has to be implemented to determine physical index based on logical index. This will take O(log N) time.
   • This method is useful when the caller wants to access only few random indices (may be in take kernels?)

My questions are below

1. is it okay to implement two ArrayAccessor for RunEndEncodedArray. One based on logical index and one based on physical index?
2. ArrayIter cannot be used as is to implement an Iterator. REE iterator needs to store two values and need a different increment logic. Is it okay to implement RunEndEncodedArrayIter and not use ArrayIter?

cc: @tustvold @viirya @alamb

[tustvold]

I'm not sure you can implement ArrayAccessor for REEArray as it doesn't know its value type? This is fine imo, we don't implement ArrayAccessor for DictionaryArray for much the same reason.

Providing an iterator abstraction, similar to TypedDictionaryArray, that downcasts the values and uses ArrayAccessor to "decode" the runs makes sense to me to help ergonomics

However, most kernels I imagine will need custom logic to handle RunEncodedArrays efficiently, e.g. take will need to parse the runs array and compute a new set of runs along with the take indices to apply to the values array. Filter will need to do something similar.

One thing to be extremely careful of, is to avoid generic code typed on both the key type and the value type in our kernels - this explodes codegen and has caused a lot of pain with dictionaries

[askoa]

Providing an iterator abstraction, similar to TypedDictionaryArray, that downcasts the values and uses ArrayAccessor to "decode" the runs makes sense to me to help ergonomics

Yes, I am planning to model ArrayAccessor after TypedDictionaryArray.

However, most kernels I imagine will need custom logic to handle RunEncodedArrays efficiently, e.g. take will need to parse the runs array and compute a new set of runs along with the take indices to apply to the values array. Filter will need to do something similar.

Does it make sense to add a function to RunEndEncodedArray that'll convert from logical index to physical index? I think it can be used by functions that are trying to decode arbitrary indices.

One thing to be extremely careful of, is to avoid generic code typed on both the key type and the value type in our kernels - this explodes codegen and has caused a lot of pain with dictionaries

I am planning to model after TypedDictionaryArray which has generics for both key and value. Would that be a problem?

[tustvold]

Would that be a problem?

Yes, wherever possible we should handle the indices and values separately, similarly to how we currently handle dictionaries. Otherwise you catastrophically explode code gen and build times, see #2596 and apache/datafusion#4999 for some more context. Non-scalar binary kernels are the only ones where I foresee this not being possible, we should gate those behind feature flags like we do for dictionaries.

TypedDictionaryArray is useful for statically typed codepaths, but using it in kernels that must be generated for every combination of types ends up being problematic