You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by yz...@apache.org on 2026/06/09 01:47:54 UTC
(polaris) 09/12: Refactor PolarisCatalogUtils.java into abstract base and version subclasses
This is an automated email from the ASF dual-hosted git repository.
MonkeyCanCode pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
commit 967544b4803184398d4fd774a7499351e059cdb8
Author: Yong <yo...@gmail.com>
AuthorDate: Thu Jun 4 22:16:08 2026 -0500
Refactor PolarisCatalogUtils.java into abstract base and version subclasses
---
.../polaris/spark/utils/PolarisCatalogUtils.java | 67 ++---------
.../spark/utils/PolarisCatalogSpark3Utils.java | 128 +--------------------
.../spark/utils/PolarisCatalogSpark4Utils.java | 128 +--------------------
3 files changed, 15 insertions(+), 308 deletions(-)
diff --git a/plugins/spark/common/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogUtils.java b/plugins/spark/common/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogUtils.java
index 0eade78658..cfc9cf0608 100644
--- a/plugins/spark/common/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogUtils.java
+++ b/plugins/spark/common/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogUtils.java
@@ -29,9 +29,6 @@ import org.apache.iceberg.rest.auth.OAuth2Util;
import org.apache.iceberg.spark.SparkCatalog;
import org.apache.polaris.spark.rest.GenericTable;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.catalyst.TableIdentifier;
-import org.apache.spark.sql.catalyst.catalog.CatalogTable;
-import org.apache.spark.sql.catalyst.catalog.CatalogTableType;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
@@ -39,12 +36,11 @@ import org.apache.spark.sql.connector.catalog.TableProvider;
import org.apache.spark.sql.execution.datasources.DataSource;
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-import scala.Option;
-import scala.Tuple2;
-import scala.collection.immutable.Map$;
-import scala.collection.mutable.Builder;
-public class PolarisCatalogUtils {
+/**
+ * Base class with helpers used by the Polaris Spark client.
+ */
+public abstract class PolarisCatalogUtils {
public static final String TABLE_PROVIDER_KEY = "provider";
public static final String TABLE_PATH_KEY = "path";
@@ -82,7 +78,7 @@ public class PolarisCatalogUtils {
* Normalize table properties for loading Spark tables by ensuring the TABLE_PATH_KEY is properly
* set. DataSourceV2 requires the path property on table loading.
*/
- private static Map<String, String> normalizeTablePropertiesForLoadSparkTable(
+ protected static Map<String, String> normalizeTablePropertiesForLoadSparkTable(
GenericTable genericTable) {
Map<String, String> properties = genericTable.properties();
boolean hasLocationClause = properties.get(TableCatalog.PROP_LOCATION) != null;
@@ -123,57 +119,8 @@ public class PolarisCatalogUtils {
* Return a Spark V1Table for formats that do not use DataSourceV2. Currently, this is being used
* for Hudi tables
*/
- public static Table loadV1SparkTable(
- GenericTable genericTable, Identifier identifier, String catalogName) {
- Map<String, String> tableProperties = normalizeTablePropertiesForLoadSparkTable(genericTable);
-
- // Need full identifier in order to construct CatalogTable
- String namespacePath = String.join(".", identifier.namespace());
- TableIdentifier tableIdentifier =
- new TableIdentifier(
- identifier.name(), Option.apply(namespacePath), Option.apply(catalogName));
-
- Builder<Tuple2<String, String>, scala.collection.immutable.Map<String, String>> mb =
- Map$.MODULE$.newBuilder();
- tableProperties.forEach((k, v) -> mb.$plus$eq(Tuple2.apply(k, v)));
- scala.collection.immutable.Map<String, String> scalaOptions = mb.result();
-
- org.apache.spark.sql.catalyst.catalog.CatalogStorageFormat storage =
- DataSource.buildStorageFormatFromOptions(scalaOptions);
-
- // Currently Polaris generic table does not contain any schema information, partition columns,
- // stats, etc
- // for now we will fill the parameters we have from polaris catalog, and let underlying client
- // resolve the rest within its catalog implementation
- org.apache.spark.sql.types.StructType emptySchema = new org.apache.spark.sql.types.StructType();
- @SuppressWarnings("deprecation")
- scala.collection.immutable.Seq<String> emptyStringSeq =
- scala.collection.JavaConverters.asScalaBuffer(new java.util.ArrayList<String>()).toList();
- CatalogTable catalogTable =
- new CatalogTable(
- tableIdentifier,
- CatalogTableType.EXTERNAL(),
- storage,
- emptySchema,
- Option.apply(genericTable.format()),
- emptyStringSeq,
- scala.Option.empty(),
- genericTable.properties().get(TableCatalog.PROP_OWNER),
- System.currentTimeMillis(),
- -1L,
- "",
- scalaOptions,
- scala.Option.empty(),
- scala.Option.empty(),
- scala.Option.empty(),
- emptyStringSeq,
- false,
- true,
- scala.collection.immutable.Map$.MODULE$.empty(),
- scala.Option.empty());
-
- return new org.apache.spark.sql.connector.catalog.V1Table(catalogTable);
- }
+ public abstract Table loadV1SparkTable(
+ GenericTable genericTable, Identifier identifier, String catalogName);
/**
* Get the catalogAuth field inside the RESTSessionCatalog used by Iceberg Spark Catalog use
diff --git a/plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark3Utils.java b/plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark3Utils.java
index 0eade78658..8f3d0139fd 100644
--- a/plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark3Utils.java
+++ b/plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark3Utils.java
@@ -18,112 +18,25 @@
*/
package org.apache.polaris.spark.utils;
-import com.google.common.collect.Maps;
-import java.lang.reflect.Field;
import java.util.Map;
-import org.apache.iceberg.CachingCatalog;
-import org.apache.iceberg.catalog.Catalog;
-import org.apache.iceberg.rest.RESTCatalog;
-import org.apache.iceberg.rest.RESTSessionCatalog;
-import org.apache.iceberg.rest.auth.OAuth2Util;
-import org.apache.iceberg.spark.SparkCatalog;
import org.apache.polaris.spark.rest.GenericTable;
-import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.TableIdentifier;
import org.apache.spark.sql.catalyst.catalog.CatalogTable;
import org.apache.spark.sql.catalyst.catalog.CatalogTableType;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
-import org.apache.spark.sql.connector.catalog.TableProvider;
import org.apache.spark.sql.execution.datasources.DataSource;
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils;
-import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import scala.Option;
import scala.Tuple2;
import scala.collection.immutable.Map$;
import scala.collection.mutable.Builder;
-public class PolarisCatalogUtils {
+/** Spark 3.5 implementation of {@link PolarisCatalogUtils}. */
+public class PolarisCatalogSpark3Utils extends PolarisCatalogUtils {
- public static final String TABLE_PROVIDER_KEY = "provider";
- public static final String TABLE_PATH_KEY = "path";
-
- /** Check whether the table provider is iceberg. */
- public static boolean useIceberg(String provider) {
- return provider == null || "iceberg".equalsIgnoreCase(provider);
- }
-
- /** Check whether the table provider is delta. */
- public static boolean useDelta(String provider) {
- return "delta".equalsIgnoreCase(provider);
- }
-
- public static boolean useHudi(String provider) {
- return "hudi".equalsIgnoreCase(provider);
- }
-
- /** Check whether the table provider is paimon. */
- public static boolean usePaimon(String provider) {
- return "paimon".equalsIgnoreCase(provider);
- }
-
- /**
- * For tables whose location is managed by Spark Session Catalog, there will be no location or
- * path in the properties.
- */
- public static boolean isTableWithSparkManagedLocation(Map<String, String> properties) {
- boolean hasLocationClause = properties.containsKey(TableCatalog.PROP_LOCATION);
- boolean hasPathClause = properties.containsKey(TABLE_PATH_KEY);
- return !hasLocationClause && !hasPathClause;
- }
-
- /**
- * Normalize table properties for loading Spark tables by ensuring the TABLE_PATH_KEY is properly
- * set. DataSourceV2 requires the path property on table loading.
- */
- private static Map<String, String> normalizeTablePropertiesForLoadSparkTable(
- GenericTable genericTable) {
- Map<String, String> properties = genericTable.properties();
- boolean hasLocationClause = properties.get(TableCatalog.PROP_LOCATION) != null;
- boolean hasPathClause = properties.get(TABLE_PATH_KEY) != null;
- Map<String, String> tableProperties = Maps.newHashMap();
- tableProperties.putAll(properties);
- if (!hasPathClause) {
- // DataSourceV2 requires the path property on table loading. However, spark today
- // doesn't create the corresponding path property if the path keyword is not
- // provided by user when location is provided. Here, we duplicate the location
- // property as path to make sure the table can be loaded.
- if (genericTable.baseLocation() != null && !genericTable.baseLocation().isEmpty()) {
- tableProperties.put(TABLE_PATH_KEY, genericTable.baseLocation());
- } else if (hasLocationClause) {
- tableProperties.put(TABLE_PATH_KEY, properties.get(TableCatalog.PROP_LOCATION));
- }
- }
- return tableProperties;
- }
-
- /**
- * Load spark table using DataSourceV2.
- *
- * @return V2Table if DataSourceV2 is available for the table format. For delta table, it returns
- * DeltaTableV2.
- */
- public static Table loadV2SparkTable(GenericTable genericTable) {
- SparkSession sparkSession = SparkSession.active();
- TableProvider provider =
- DataSource.lookupDataSourceV2(genericTable.format(), sparkSession.sessionState().conf())
- .get();
- Map<String, String> tableProperties = normalizeTablePropertiesForLoadSparkTable(genericTable);
- return DataSourceV2Utils.getTableFromProvider(
- provider, new CaseInsensitiveStringMap(tableProperties), scala.Option.empty());
- }
-
- /**
- * Return a Spark V1Table for formats that do not use DataSourceV2. Currently, this is being used
- * for Hudi tables
- */
- public static Table loadV1SparkTable(
+ @Override
+ public Table loadV1SparkTable(
GenericTable genericTable, Identifier identifier, String catalogName) {
Map<String, String> tableProperties = normalizeTablePropertiesForLoadSparkTable(genericTable);
@@ -174,37 +87,4 @@ public class PolarisCatalogUtils {
return new org.apache.spark.sql.connector.catalog.V1Table(catalogTable);
}
-
- /**
- * Get the catalogAuth field inside the RESTSessionCatalog used by Iceberg Spark Catalog use
- * reflection. TODO: Deprecate this function once the iceberg client is updated to 1.9.0 to use
- * AuthManager and the capability of injecting an AuthManger is available. Related iceberg PR:
- * https://github.com/apache/iceberg/pull/12655
- */
- public static OAuth2Util.AuthSession getAuthSession(SparkCatalog sparkCatalog) {
- try {
- Field icebergCatalogField = sparkCatalog.getClass().getDeclaredField("icebergCatalog");
- icebergCatalogField.setAccessible(true);
- Catalog icebergCatalog = (Catalog) icebergCatalogField.get(sparkCatalog);
- RESTCatalog icebergRestCatalog;
- if (icebergCatalog instanceof CachingCatalog) {
- Field catalogField = icebergCatalog.getClass().getDeclaredField("catalog");
- catalogField.setAccessible(true);
- icebergRestCatalog = (RESTCatalog) catalogField.get(icebergCatalog);
- } else {
- icebergRestCatalog = (RESTCatalog) icebergCatalog;
- }
-
- Field sessionCatalogField = icebergRestCatalog.getClass().getDeclaredField("sessionCatalog");
- sessionCatalogField.setAccessible(true);
- RESTSessionCatalog sessionCatalog =
- (RESTSessionCatalog) sessionCatalogField.get(icebergRestCatalog);
-
- Field authField = sessionCatalog.getClass().getDeclaredField("catalogAuth");
- authField.setAccessible(true);
- return (OAuth2Util.AuthSession) authField.get(sessionCatalog);
- } catch (Exception e) {
- throw new RuntimeException("Failed to get the catalogAuth from the Iceberg SparkCatalog", e);
- }
- }
}
diff --git a/plugins/spark/v4.0/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark4Utils.java b/plugins/spark/v4.0/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark4Utils.java
index b431916c1b..555e55346c 100644
--- a/plugins/spark/v4.0/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark4Utils.java
+++ b/plugins/spark/v4.0/spark/src/main/java/org/apache/polaris/spark/utils/PolarisCatalogSpark4Utils.java
@@ -18,112 +18,25 @@
*/
package org.apache.polaris.spark.utils;
-import com.google.common.collect.Maps;
-import java.lang.reflect.Field;
import java.util.Map;
-import org.apache.iceberg.CachingCatalog;
-import org.apache.iceberg.catalog.Catalog;
-import org.apache.iceberg.rest.RESTCatalog;
-import org.apache.iceberg.rest.RESTSessionCatalog;
-import org.apache.iceberg.rest.auth.OAuth2Util;
-import org.apache.iceberg.spark.SparkCatalog;
import org.apache.polaris.spark.rest.GenericTable;
-import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.TableIdentifier;
import org.apache.spark.sql.catalyst.catalog.CatalogTable;
import org.apache.spark.sql.catalyst.catalog.CatalogTableType;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
-import org.apache.spark.sql.connector.catalog.TableProvider;
import org.apache.spark.sql.execution.datasources.DataSource;
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils;
-import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import scala.Option;
import scala.Tuple2;
import scala.collection.immutable.Map$;
import scala.collection.mutable.Builder;
-public class PolarisCatalogUtils {
+/** Spark 4.0 implementation of {@link PolarisCatalogUtils}. */
+public class PolarisCatalogSpark4Utils extends PolarisCatalogUtils {
- public static final String TABLE_PROVIDER_KEY = "provider";
- public static final String TABLE_PATH_KEY = "path";
-
- /** Check whether the table provider is iceberg. */
- public static boolean useIceberg(String provider) {
- return provider == null || "iceberg".equalsIgnoreCase(provider);
- }
-
- /** Check whether the table provider is delta. */
- public static boolean useDelta(String provider) {
- return "delta".equalsIgnoreCase(provider);
- }
-
- public static boolean useHudi(String provider) {
- return "hudi".equalsIgnoreCase(provider);
- }
-
- /** Check whether the table provider is paimon. */
- public static boolean usePaimon(String provider) {
- return "paimon".equalsIgnoreCase(provider);
- }
-
- /**
- * For tables whose location is managed by Spark Session Catalog, there will be no location or
- * path in the properties.
- */
- public static boolean isTableWithSparkManagedLocation(Map<String, String> properties) {
- boolean hasLocationClause = properties.containsKey(TableCatalog.PROP_LOCATION);
- boolean hasPathClause = properties.containsKey(TABLE_PATH_KEY);
- return !hasLocationClause && !hasPathClause;
- }
-
- /**
- * Normalize table properties for loading Spark tables by ensuring the TABLE_PATH_KEY is properly
- * set. DataSourceV2 requires the path property on table loading.
- */
- private static Map<String, String> normalizeTablePropertiesForLoadSparkTable(
- GenericTable genericTable) {
- Map<String, String> properties = genericTable.properties();
- boolean hasLocationClause = properties.get(TableCatalog.PROP_LOCATION) != null;
- boolean hasPathClause = properties.get(TABLE_PATH_KEY) != null;
- Map<String, String> tableProperties = Maps.newHashMap();
- tableProperties.putAll(properties);
- if (!hasPathClause) {
- // DataSourceV2 requires the path property on table loading. However, spark today
- // doesn't create the corresponding path property if the path keyword is not
- // provided by user when location is provided. Here, we duplicate the location
- // property as path to make sure the table can be loaded.
- if (genericTable.baseLocation() != null && !genericTable.baseLocation().isEmpty()) {
- tableProperties.put(TABLE_PATH_KEY, genericTable.baseLocation());
- } else if (hasLocationClause) {
- tableProperties.put(TABLE_PATH_KEY, properties.get(TableCatalog.PROP_LOCATION));
- }
- }
- return tableProperties;
- }
-
- /**
- * Load spark table using DataSourceV2.
- *
- * @return V2Table if DataSourceV2 is available for the table format. For delta table, it returns
- * DeltaTableV2.
- */
- public static Table loadV2SparkTable(GenericTable genericTable) {
- SparkSession sparkSession = SparkSession.active();
- TableProvider provider =
- DataSource.lookupDataSourceV2(genericTable.format(), sparkSession.sessionState().conf())
- .get();
- Map<String, String> tableProperties = normalizeTablePropertiesForLoadSparkTable(genericTable);
- return DataSourceV2Utils.getTableFromProvider(
- provider, new CaseInsensitiveStringMap(tableProperties), scala.Option.empty());
- }
-
- /**
- * Return a Spark V1Table for formats that do not use DataSourceV2. Currently, this is being used
- * for Hudi tables
- */
- public static Table loadV1SparkTable(
+ @Override
+ public Table loadV1SparkTable(
GenericTable genericTable, Identifier identifier, String catalogName) {
Map<String, String> tableProperties = normalizeTablePropertiesForLoadSparkTable(genericTable);
@@ -175,37 +88,4 @@ public class PolarisCatalogUtils {
return new org.apache.spark.sql.connector.catalog.V1Table(catalogTable);
}
-
- /**
- * Get the catalogAuth field inside the RESTSessionCatalog used by Iceberg Spark Catalog use
- * reflection. TODO: Deprecate this function once the iceberg client is updated to 1.9.0 to use
- * AuthManager and the capability of injecting an AuthManger is available. Related iceberg PR:
- * https://github.com/apache/iceberg/pull/12655
- */
- public static OAuth2Util.AuthSession getAuthSession(SparkCatalog sparkCatalog) {
- try {
- Field icebergCatalogField = sparkCatalog.getClass().getDeclaredField("icebergCatalog");
- icebergCatalogField.setAccessible(true);
- Catalog icebergCatalog = (Catalog) icebergCatalogField.get(sparkCatalog);
- RESTCatalog icebergRestCatalog;
- if (icebergCatalog instanceof CachingCatalog) {
- Field catalogField = icebergCatalog.getClass().getDeclaredField("catalog");
- catalogField.setAccessible(true);
- icebergRestCatalog = (RESTCatalog) catalogField.get(icebergCatalog);
- } else {
- icebergRestCatalog = (RESTCatalog) icebergCatalog;
- }
-
- Field sessionCatalogField = icebergRestCatalog.getClass().getDeclaredField("sessionCatalog");
- sessionCatalogField.setAccessible(true);
- RESTSessionCatalog sessionCatalog =
- (RESTSessionCatalog) sessionCatalogField.get(icebergRestCatalog);
-
- Field authField = sessionCatalog.getClass().getDeclaredField("catalogAuth");
- authField.setAccessible(true);
- return (OAuth2Util.AuthSession) authField.get(sessionCatalog);
- } catch (Exception e) {
- throw new RuntimeException("Failed to get the catalogAuth from the Iceberg SparkCatalog", e);
- }
- }
}