Allow loading custom Catalog implementation in Spark and Flink

[jackye1995]

As we are having multiple new Catalog implementations added to Iceberg, we need a way to load those Catalogs in Spark and Flink easily. Currently there is a simple switch branch that chooses between hive and hadoop catalogs. This approach requires the iceberg-spark and iceberg-flink module to take a dependency on the catalog implementation modules. This would potentially bring in many unnecessary dependencies as more and more cloud providers try to add support for Iceberg.

This PR proposes the following way to load custom Catalog implementations:

1. the type of a catalog can be hive or hadoop to keep existing behaviors
2. if catalog-impl is set, type is ignored and we will load catalog based on the class value
3. A catalog must have a no-arg constructor be initialized by calling initialize()
4. A catalog must implement Hadoop Configurable to read Hadoop configuration

For example, a GlueCatalog will be used in Spark like the following:

spark.sql.catalog.glue = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.glue.catalog-impl = org.apache.iceberg.aws.glue.GlueCatalog

[giovannifumarola]

Thanks Jack. I like the following approach.
spark.sql.catalog.glue = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.glue.type = custom
spark.sql.catalog.glue.impl = org.apache.iceberg.aws.glue.GlueCatalog

[rdblue]

My main concern with this is how configuration is currently passed using Hadoop Configuration. We want to avoid using Configuration and only pass it where it is needed for our use of Hadoop APIs, like FileSystem. Passing Configuration as the last argument is fine because many catalogs will currently need it to use HadoopFileIO, but we'll need an alternative as well.

I think we also want to pass options from the catalog config in Flink and Spark, where users can pass properties like uri and warehouse. Could you add a Map to this to pass the catalog config options?

[rymurr]

I think we also want to pass options from the catalog config in Flink and Spark, where users can pass properties like uri and warehouse. Could you add a Map to this to pass the catalog config options?

I like the Map over Configuration suggestion as well. In #1587 I made the constructor take String name, Map props, Configuration conf as it still needs a HadoopFileIO

Has anyone thought of how to do this for the IcebergSource? Currently df.write().format("iceberg") is as far as I understand going to use Hive/HDFS regardless of these settings.

[rdblue]

@rymurr, for your IcebergSource question: the source can implement a trait to return the catalog and identifier to load instead of returning a table itself. There are two reasons we don't do this already:

1. The catalog must be defined in Spark properties, so we either need to have a "default" catalog or have a way to configure a catalog for the source
2. If we use a catalog, then we still need to support path-based tables. We will need to add a way to pass a path as an identifier to the catalog and have it load using HadoopTables

Figuring out how we want to do this shouldn't be too difficult. We just found it easier to keep the existing behavior for the last release since there weren't other catalogs at the time.

[jackye1995]

fix rebase issue, reopen PR

[rdblue]

Instead of using type=custom and impl=com.example.Catalog, why not just combine them into type=com.example.Catalog. We can try to load the type as an implementation class if it isn't a well-known name like "hive".

[rdblue]

@aokolnychyi and @RussellSpitzer, do you have an opinion here?

[jackye1995]

Yeah this sounds like a cleaner way to go, the only disadvantage is that we are overloading the term type. I have updated the code.

[rdblue]

@jackye1995, I'm happy with the implementation and behavior (other than changing IcebergSource later). The main things to fix are names and how we configure it. Thanks!

[jackye1995]

I meant to make a comment before on this. We used Function<String, String> in Nessie for this and it feels like a narrower interface for this purpose. See here: https://github.com/projectnessie/nessie/blob/main/clients/client/src/main/java/com/dremio/nessie/client/NessieClient.java#L161

Interesting, but it should still be compatible by using Map<String, String>::get

[jacques-n]

I was suggesting using the narrower function instead map here as well.

[rdblue]

What's the drawback of using a map? It is a bit narrower to use the function, but it is also limiting. Catalogs can't use getOrDefault or check contains. I also don't have a serious problem if we need to copy to create an Immutable map for this because it isn't in a performance-critical path.

[jacques-n]

Just a lot extra methods exposed. Iteration, put, etc.

[jackye1995]

Just a lot extra methods exposed. Iteration, put, etc.

I think the advantage of using a Map is that:

1. consistent with the TableProperties interface
2. consistent with the Spark and Flink properties interface
3. as Ryan says, there are some useful methods like getOrDefault and contains, and iteration might still be needed for some use cases such as getting all configs of a certain prefix.

[jackye1995]

@rdblue@jacques-n any thoughts?

[jacques-n]

I'm fine with it as is.

Some background: I'm not especially supportive of maps in interfaces since they are so broad. I've had bad experiences in the past where we've had to move maps to external storage (say DynamoDB) and having a broad use of the Map interface hurt us in reworking things.

[rdblue]

I think it is okay to use a map here. Like I said, even if we had to copy values to create a map to pass in here, that happens once per session and is not an unreasonable amount of overhead.