Support DayTimeIntervalType in ParquetCachedBatchSerializer[databricks]

[firestarman]

This PR is to add the new DayTimeIntervalType support in ParquetCachedBatchSerializer for Spark v3.3.0+, along with some tests.

closes #4148
closes #4931

Signed-off-by: Firestarman firestarmanllc@gmail.com

[firestarman]

build

[firestarman]

build

[razajafri]

Why is this test method so complicated? What are we doing in this test that we aren't doing in any other test? We have a DF that we cache and then pull it from the cache, compare the result from the CPU and GPU

In other words, why don't we just do the following?

def test_cache_daytimeinterval_input_columnar():
    def func(spark):
        df = two_col_df(spark, DayTimeIntervalGen(), alongside_gen)
        df.cache().count()
        return df.selectExpr("a") // or whatever the column name is

    assert_gpu_and_cpu_are_equal_collect(func, conf={YOUR CONF})

I could be missing something

[firestarman]

I thought since we implement all the paths (both CPU and GPU) in the PCBS, I think the output for all of the paths should be equal to the original data.
Comparing the GPU output to CPU output bases on the CPU output of the PCBS is reliable and equal to the Spark output.
Maybe it is over designed.

[firestarman]

Reading data from parquet and writing it back is to get a columnar input to be cached and then convert the cached batches to columnar batches.
I do the same as your suggestion in the test_cache_daytimeinterval_input_row test, it only checks the paths of conversion between internal rows and cached batches.

[firestarman]

Simplified this test

[razajafri]

Reading/writing to parquet isn't necessary to write/read columnar cache.

If you look at InMemoryTableScanExec it calls the convertColumnarBatchToCachedBatch if the spark plan supports columnar input and if the serializer supports columnar input (which it always does here)

Reading cache columnar in PCBS depends on three variables, whether the conf spark.sql.inMemoryColumnarStorage.enableVectorizedReader is enabled and the plan has 100 or less columns and the plan's output is AtomicType or NullType

[firestarman]

Good to know. Just got home and updated it.
Merged the two tests into one and updated the test function as your suggestion above.

[firestarman]

build

[res-life]

Can use the following to improve the performance:

var rowIndex = 0
while (rowIndex < batch.numRows()) {

    ......
    rowIndex += 1
}

A similar PR: #4770

[firestarman]

Number of rows is always quite large, so this change can improve some performance. However here is for columns, this suggestion will get little benfit for performance, since number of columns is usually small.

[firestarman]

@razajafri Could you review this again ?

[firestarman]

build

[firestarman]

build