Spark - Implement FunctionCatalog and Truncate

[kbendick]

Implements FunctionCatalog for Spark 3.3 and implements all variants of Truncate.

FunctionCatalog

This allows users of SparkCatalog and SparkSessionCatalog to use truncate without having to register it as a UDF.

All Iceberg functions that we register into the function catalog are accessible when used with an Iceberg spark catalog and:

1. No namespace is referenced - the storage partitioned joins implementation requires this.
   e.g. my_catalog.truncate(width, value).
   Note - Using truncate(width, value) typically does not work, as Spark adds the namespace to the call. system.truncate should be preferred.
2. The system namespace is referenced, to match called procedure syntax. Note this only works right now with the SparkCatalog, as the SparkSessionCatalog has logic in Spark to verify the namespace exists.
   e.g. my_catalog.system.truncate(width, value) or system.truncate(6, column)

Truncate

The truncate function also allows for a dynamic width or the width to come from a column - though typically the width will likely be static for one given call as it's mostly intended to be used to match partition transforms (specifically with joins or on non-partition columns to create a new column in the data without needing to partition on it).

This PR refactors the definition of the transform functions into a utility class where needed so that Spark’s magic functions can call them via the static invoke function and not duplicate logic. This allows Spark to include the functions in codegen.

Special Considerations for Using Function Catalog Efficiently via Magic Functions and Code Gen

The requirements for magic functions to be used with codegen include that:

1. invoke is a static function
2. invoke takes in the primitive types / native Spark types corresponding to each of Spark's input DataTypes (e.g. int for IntegerType and UTF8String for StringType).

Further documentation on the magic functions is found here in the ScalarFunction JavaDoc

This partially closes #5349

[kbendick]

After this, I'll add bucket and zorder as well.

This is to facilitate the usage of the various transforms from PySpark as well as SQL.

Additionally, having a zorder function will make it possible for people to pre-sort their data on input, vs having to zorder sort it when running a data compaction job.

[kbendick]

@rdblue PTAL. I was going to tag other people as well.

[kbendick]

@rdblue Do you think I should break this into a 2 PRs?

[kbendick]

Looking at the storage partition joins, it looks like the function bucket will need to be resolvable on the empty namespace. I can update that here or a follow up:

https://github.com/apache/spark/blob/e9eb28e27d10497c8b36774609823f4bbd2c8500/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala#L84-L100

[aokolnychyi]

I'd love to take a look as well. I should have some time in a day. We also had some progress on bucketed joins internally.

[kbendick]

I'd love to take a look as well. I should have some time in a day. We also had some progress on bucketed joins internally.

Thanks Anton. Was going to tag you today now that it's cleaned up. Also cc @huaxingao @flyrain @nastra @Fokko

[kbendick]

Right not we're requiring the call be to the system namespace, but the storage partitioned join implementation looks for a function called bucket in the FunctionCatalog using an empty array for the namespace per this diff in the merged PR in Spark for Storage Partitioned Joins inside V2ExpressionUtils#toCatalystTransform..

So we should probably allow the empty namespace to resolve functions as well.

[kbendick]

Link to the code that resolves our own bucket implementation in case the link above doesn't resolve: https://github.com/apache/spark/blob/47f0303944abb11d3018186bc125113772eff8ef/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala#L84-L100

[kbendick]

Because this PR is so big, I'm going to separate out the FunctionCatalog implementation from Truncate.

I'm going to add a very simple function to be able to test it but keep the code to review a lot smaller. 👍

[kbendick]

I've opened #5377 to cover just the FunctionCatalog.

This PR is too big, and this way we can focus on just the FunctionCatalog business without having to worry about the details of truncate.

I've added an iceberg_version function in the other PR to assist with testing.

[kbendick]

This PR is closed in favor of #5377 and #5411.

I'll open a PR for Truncate and link it shortly.