[NSE-394]Support ColumnarArrowEvalPython operator

[xuechendi]

Main functionality is working, UT is added, jupyter script as below

import findspark
findspark.init()

import os
from pyspark.sql import *
from pyspark import *
import pyspark.sql.functions as F
from pyspark.sql.types import *

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("udf_test")\
        .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.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()

from pyspark.sql.functions import pandas_udf, PandasUDFType
@pandas_udf('long')
def pandas_plus_one(v):
    return (v + 1)

df = spark.read.format("arrow").load("/recsys2021_0608")
dict_df = spark.read.parquet("/recsys2021_0608_processed/recsys_dicts/language")
df = df.select("language", "tweet_timestamp")
df = df.join(dict_df.withColumnRenamed('dict_col', 'language'), 'language', 'left')
df = df.withColumn("processed_tweet_timestamp", pandas_plus_one(F.col("tweet_timestamp")))
#df = df.withColumn("processed_tweet_timestamp", pandas_plus_one(F.col("tweet_timestamp")+F.lit(1)))
%time df.show()

[github-actions]

#394

[xuechendi]

string test

from pyspark.sql.functions import pandas_udf, PandasUDFType
@pandas_udf('string')
def pd_get_first(v):
    return v.str.split('\t').str[0]

df = spark.read.format("arrow").load("/recsys2021_0608")
dict_df = spark.read.parquet("/recsys2021_0608_processed/recsys_dicts/language")
df = df.select("language", "present_domains")
df = df.join(dict_df.withColumnRenamed('dict_col', 'language'), 'language', 'left')
df = df.withColumn("processed_present_domains", pd_get_first(F.col("present_domains")))
%time df.select('processed_present_domains').distinct().show()