[FEA] memory efficient explode and pos_explode implementations

[revans2]

Is your feature request related to a problem? Please describe.

#2975 asks for pandas explode. We need something similar for Spark. We have already implemented a version of this when the explode is operating on scalar lists (don't need actual list support to implement that), but quickly ran into memory issues with it. This is because when you explode a data frame on a single column all of the other columns end up being replicated.

>> a = pd.Series([[[1, 2], [3, 4]], [[5, 6], None]])
>> b = pd.Series([100, 200])
>> c = pd.DataFrame(data={'a': a, 'b' : b})
>> c.explode('a')
        a    b
0  [1, 2]  100
0  [3, 4]  100
1  [5, 6]  200
1    None  200

>> c.explode('a').explode('a')
      a    b
0     1  100
0     2  100
0     3  100
0     4  100
1     5  200
1     6  200
1  None  200

So the more columns you have and the deeper the arrays the more duplicate data that ends up being produced.

Because Spark is processing a stream of tables instead of a single table we implemented a "fix" for this by essentially keeping all of the columns cached in memory and outputting a new table for each entry in the scalar list. i.e.

for each item in scalar_explode_list:
    val new_column = scalarToColumn(item, cachedTable.numRows())
    yield new_column + cachedTable

Note that we also support a posexplode where we insert a new column that also includes the index of the array item.

Describe the solution you'd like

We would like to have a solution that lets us do an explode and a posexplode but limiting the output size.

I see a few options for this.

1. We could splice the input table into smaller pieces and then run a regular explode on each of them in turn. The main thing we would need form this is some way to come up with the split positions. At a minimum we could just do this based off of a target number of output rows (get the sum of the length of all of the columns in the explode and then assuming they are all equal come up with the number of splits to do). But that would be vulnerable to data skew so a better algorithm possibly looking at data size instead of just row count would be ideal.

2. We could do something similar to our current "fix" and get the length of the longest list entry in the column. Then go through one array index at a time, filter out all all entries that are not long enough, and then return the batch along with the item form the array at the given index.

I like one if we can do it efficiently because it is simple, and just one separate stand alone command. Also if we have skew in the length of the arrays, then 2 would need some special case processing to avoid having too much overhead when we hit the tail.

I am also open to other ideas if someone has a good one.

[harrism]

I think this was erroneously automatically closed by the merge of #7140. Reopening.