Yxang
commented
1 year ago I want to provide more information on the last limitation that time functions support only one input type. I tried but failed to use auto type casting via UnboundFunction.bind(StructType). It is not working when that function it is used in requiredDistribution in RequiresDistributionAndOrdering.
The problem can be reproduced code by
- uncomment the line of xenon/clickhouse/func/clickhouse/Months.scala:38:
case Array(StructField(_, TimestampType, _, _)) | Array(StructField(_, TimestampNTZType, _, _)) => this - Build the dev docker environment. Need to set spark version to 3.4.0, and maybe manually compile
iceberg-spark-runtimeand add the jar into spark package - Run SQL in clickhouse
create database if not exists test on cluster single_replica; create or replace table test.test_month_shard on cluster single_replica ( create_time timestamp, create_date date, value String ) engine = MergeTree() order by create_time; create or replace table test.test_month on cluster single_replica as test.test_month_shard engine =Distributed(single_replica, 'test', 'test_month_shard', toYYYYMM(create_date)); - Run SQL in spark. It will fail with
org.apache.spark.SparkException: [INTERNAL_ERROR] A method named "invoke" is not declared in any enclosing class nor any supertypeinsert into clickhouse_s1r1.test.test_month values (timestamp'2021-01-01 10:10:10', timestamp'2021-01-01 10:10:10', '1'), (timestamp'2022-02-02 10:10:10', timestamp'2022-02-02 10:10:10', '2'), (timestamp'2023-03-03 10:10:10', timestamp'2023-03-03 10:10:10', '3'), (timestamp'2024-04-04 10:10:10', timestamp'2024-04-04 10:10:10', '4') AS tab(create_time, create_date, value) - If we create Clickhouse distribute table as
toYYYYMM(create_time), and inserttimestamptype in Spark, it will fail with the same message as 4 - Directly run UDF will success
use clickhouse_s1r1.test; select clickhouse_months(timestamp'2024-04-04 10:10:10')
So it seems that this functionality is broken with requiredDistribution in RequiresDistributionAndOrdering. I suppose this is a problem on Spark's side?
I believe adding a type cast before calling UDF is always helpful because we do not know the inserted data type in advance. However, this is also not achievable because requiredDistribution does not handle casting, though it can be modified just a little to support this.
pan3793
What does this PR do?
This PR introduces support for writing to ClickHouse tables with functions in distribute/partition expressions after Spark 3.4's merge of SPARK-39607, with the need to implement ClickHouse functions into corresponding Spark ones.
The implemented functions in this PR include:
toYear,toYYYYMM, andtoYYYYMMDD(with only Sparkdateinput), andtoHour(with only Sparktimestamportimestamp_ntz)TransformlikeYearsTransformis not evaluablemurmurHash3_64,murmurHash3_32,murmurHash2_64,murmurHash2_32, andcityHash64with variable number ofstringarguments.Also existing v2 function for
xxHash64is available.Corresponding tests are also included.
Limitations
Main limitation is Spark function cannot fully match Clickhouse function's behavior, including:
Hash functions support for maximum of 5 arguments, limited by Spark's v2 function does not support variable length of inputsstringdata type, due to the complexity in aligning Spark's and ClickHouse's implementation of hash function and type castingorg.apache.spark.sql.AnalysisException: some_function(arg) is not currently supportedis ambiguous and misleading