[NSE-317]fix columnar cache

[xuechendi]

This PR is aim to fix current UT issues related with RDD cache

1. for tests wo/ Arrow Serialiazer, we now avoided using ColumarInMemoryTableScan
2. for tests w/ Arrow Serializer, while its cachedPlan doesn't support columnar, now we supported ConvertInternalRowToCachedBatch in ArrowSerializer
3. for tests w/ Arrow Serializer and its cachedPlan also supports columnar, we will use Arrow for fast cache.

TODO:
need to back port spark PR's for manual close cached blocks, https://issues.apache.org/jira/browse/SPARK-35396
need to test with our jupyter test(now only UT tested)

Fixed: #317

[github-actions]

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/oap-project/native-sql-engine/issues

Then could you also rename commit message and pull request title in the following format?

[NSE-${ISSUES_ID}] ${detailed message}

See also:

• Other pull requests

[xuechendi]

@rui-mo , please take a check

[rui-mo]

verified on my env. This pr cleans up my test code and refresh the ut: xuechendi#3

[xuechendi]

@zhouyuan , should be mergable

native_sql_path = "/mnt/nvme2/chendi/intel-bigdata/OAP/native-sql-engine/native-sql-engine/core/target/spark-columnar-core-1.2.0-snapshot-jar-with-dependencies.jar"
native_arrow_datasource_path = "/mnt/nvme2/chendi/intel-bigdata/OAP/native-sql-engine/arrow-data-source/standard/target/spark-arrow-datasource-standard-1.2.0-snapshot-jar-with-dependencies.jar"
spark = SparkSession.builder.master('yarn')\
        .appName("Recsys2021_data_process")\
        .config("spark.executorEnv.LD_LIBRARY_PATH", "/usr/local/lib64/")\
        .config("spark.driver.extraClassPath", 
                f"{native_sql_path}:{native_arrow_datasource_path}")\
        .config("spark.executor.extraClassPath",
                f"{native_sql_path}:{native_arrow_datasource_path}")\
        .config("spark.sql.extensions", "com.intel.oap.ColumnarPlugin")\
        .config("spark.shuffle.manager", "org.apache.spark.shuffle.sort.ColumnarShuffleManager")\
        .config("spark.sql.cache.serializer", "org.apache.spark.sql.execution.ArrowColumnarCachedBatchSerializer")\
        .config("spark.executor.memory", "10g")\
        .config("spark.executor.memoryOverhead", "16g")\
        .config("spark.memory.offHeap.use", "true")\
        .config("spark.memory.offHeap.size", "12G")\
        .config("spark.executor.extraJavaOptions", "-XX:MaxDirectMemorySize=25G")\
        .getOrCreate()

test: cache join result then do aggregate

df = spark.read.format('arrow').load("/recsys2021_0608")
dict_df = spark.read.parquet("/recsys2021_0608_processed/recsys_dicts/language")
df = df.select("tweet_id", "language", "tweet_timestamp", "engaged_with_user_id", "engaging_user_id")
df = df.join(dict_df.withColumnRenamed('dict_col', 'language'), 'language', 'left')
df.cache()
df.groupby('dict_col_id', 'language').count().show()

[rui-mo]

hi @xuechendi , we have changed this package "travis" into "nativesql", could you also put this file into nativesql folder?
This file should also be renamed into NativeCachedTableSuite.scala.

[github-actions]

#317