You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "iting0321 (via GitHub)" <gi...@apache.org> on 2026/02/18 17:13:22 UTC
[PR] [WIP] Add integration tests for Paimon [polaris]
iting0321 opened a new pull request, #3820:
URL: https://github.com/apache/polaris/pull/3820
<!--
๐ Describe what changes you're proposing, especially breaking or user-facing changes.
๐ See https://github.com/apache/polaris/blob/main/CONTRIBUTING.md for more.
-->
## Checklist
- [ ] ๐ก๏ธ Don't disclose security issues! (contact security@apache.org)
- [ ] ๐ Clearly explained why the changes are needed, or linked related issues: Fixes #
- [ ] ๐งช Added/updated tests with good coverage, or manually tested (and explained how)
- [ ] ๐ก Added comments for complex logic
- [ ] ๐งพ Updated `CHANGELOG.md` (if needed)
- [ ] ๐ Updated documentation in `site/content/in-dev/unreleased` (if needed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "flyingImer (via GitHub)" <gi...@apache.org>.
flyingImer commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2886897201
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -143,7 +143,29 @@ public Table loadTable(Identifier ident) throws NoSuchTableException {
try {
return this.icebergsSparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
- return this.polarisSparkCatalog.loadTable(ident);
+ // For generic tables, first try to load from Polaris to check the format
Review Comment:
yea, lets start a ML thread on this.
IMHO l if this is the pattern Polaris is going forward, we should have the spec or a rule deterministically suggests the resolving order. And technically, making them pluggable and consistent ordering across all operations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "flyingImer (via GitHub)" <gi...@apache.org>.
flyingImer commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3855985665
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +191,68 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
Review Comment:
IIUC, this adds the Paimon-specific lifecycle for createTable, but the staged paths still unconditionally delegate to Iceberg.
Since this catalog implements StagingTableCatalog, Spark routes Create Table As Select(CTAS)/Replace Table As Select(RTAS) through stageCreate/stageReplace rather than this method, so USING paimon can bypass both Paimon creation and Polaris registration.
Is this the desired behavior? Could we either route Paimon staged operations through the corresponding Paimon + Polaris lifecycle, or explicitly reject them before any side effect? It would also be good to cover CTAS/RTAS in regression tests.
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +276,104 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.useDelta(provider)) {
+ // For delta table, most of the alter operations is a delta log manipulation,
+ // we load the delta catalog to help handling the alter table operation.
+ // NOTE: This currently doesn't work for changing file location and file format
+ // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
+ TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
+ return deltaCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.useHudi(provider)) {
+ TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
+ return hudiCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
return paimonCatalog.alterTable(ident, changes);
- } else {
- return this.polarisSparkCatalog.alterTable(ident);
+ } catch (NoSuchTableException paimonException) {
+ throw paimonException;
+ } catch (Exception paimonException) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to alter Paimon table %s: %s", ident, paimonException.getMessage()),
+ paimonException);
}
}
+
+ // Default: use Polaris catalog for alter operation
+ return this.polarisSparkCatalog.alterTable(ident);
}
@Override
public boolean dropTable(Identifier ident) {
- return this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+ // First, try to drop as an Iceberg table
+ if (this.icebergsSparkCatalog.dropTable(ident)) {
+ return true;
+ }
+
+ // For generic tables, check the provider to delegate to the appropriate catalog
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ try {
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ if (PolarisCatalogUtils.usePaimon(provider)) {
Review Comment:
now that Paimon dropTable has a two-catalog lifecycle, purgeTable needs the same provider-aware treatment. Today a Paimon purge falls through to PolarisSparkCatalog.purgeTable(), which only removes the Generic Table record, leaving the physical Paimon table behind.
Should we either define the Paimon โ Polaris purge lifecycle here as well, or reject Paimon purge before modifying either side? A regression test that checks both catalog states would help prevent this from drifting again IMO.
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +191,68 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, use Paimon's SparkCatalog to create the table first,
+ // then register it in Polaris so it appears in the unified catalog view (SHOW TABLES).
+ // Paimon manages its own table paths at warehouse/database.db/table_name
+ // so we don't require a location to be specified.
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ // Ensure namespace exists in Paimon before creating table
+ paimonHelper.ensureNamespaceExists(ident.namespace());
+ Table paimonTable = paimonCatalog.createTable(ident, schema, transforms, properties);
+
+ // Register the table in Polaris as a generic table so it's visible via SHOW TABLES.
+ // Paimon's SparkTable.properties() populates "location" from CoreOptions.PATH,
+ String tableLocation = paimonTable.properties().get(TableCatalog.PROP_LOCATION);
+ if (tableLocation == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Paimon table %s was created but its properties do not contain a location. "
+ + "This is unexpected โ Paimon should always populate the location from its CoreOptions.PATH.",
+ ident));
+ }
+ Map<String, String> registrationProps = new java.util.HashMap<>(properties);
+ registrationProps.put(TableCatalog.PROP_LOCATION, tableLocation);
+
+ try {
+ Tasks.foreach(registrationProps)
+ .retry(MAX_POLARIS_REGISTRATION_RETRIES)
+ .exponentialBackoff(RETRY_MIN_WAIT_MS, RETRY_MAX_WAIT_MS, RETRY_TOTAL_TIMEOUT_MS, 2.0)
+ .shouldRetryTest(
+ e ->
+ !(e instanceof TableAlreadyExistsException)
+ && !(e instanceof NoSuchNamespaceException))
Review Comment:
I don't think this retry classifier can currently see the exception types it is testing for. TableAlreadyExistsException and NoSuchNamespaceException are wrapped in RuntimeException inside the task, so shouldRetryTest receives the wrapper and treats both as retryable.
Also, Tasks.retry(3) means the initial attempt plus three retries. Could we preserve/unwrap the original cause for classification and add a regression test for the attempt count and single rollback behavior?
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +191,68 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, use Paimon's SparkCatalog to create the table first,
+ // then register it in Polaris so it appears in the unified catalog view (SHOW TABLES).
+ // Paimon manages its own table paths at warehouse/database.db/table_name
+ // so we don't require a location to be specified.
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ // Ensure namespace exists in Paimon before creating table
+ paimonHelper.ensureNamespaceExists(ident.namespace());
+ Table paimonTable = paimonCatalog.createTable(ident, schema, transforms, properties);
+
+ // Register the table in Polaris as a generic table so it's visible via SHOW TABLES.
+ // Paimon's SparkTable.properties() populates "location" from CoreOptions.PATH,
+ String tableLocation = paimonTable.properties().get(TableCatalog.PROP_LOCATION);
+ if (tableLocation == null) {
Review Comment:
The physical Paimon table has already been created before this validation, but this failure is outside the rollback-protected scope. Even if location is expected to be an invariant for the default Paimon implementation, violating that invariant should not leave a successfully created physical table behind while the overall create reports failure.
I feel we make the rollback guard cover every failure after paimonCatalog.createTable(...) succeeds, including this validation, WDYT?
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PaimonHelper.java:
##########
@@ -81,13 +113,33 @@ public TableCatalog loadPaimonCatalog(PolarisSparkCatalog polarisSparkCatalog) {
e);
}
- // Set the Polaris Spark Catalog as the delegate catalog of Paimon Catalog.
- // This allows Paimon to use Polaris for metadata management while handling
- // Paimon-specific operations like manifest and snapshot management.
- if (this.paimonCatalog instanceof DelegatingCatalogExtension) {
- ((DelegatingCatalogExtension) this.paimonCatalog).setDelegateCatalog(polarisSparkCatalog);
- }
+ // Build options for Paimon initialization
+ Map<String, String> paimonOptions = new HashMap<>();
+ paimonOptions.put("warehouse", this.paimonWarehouse);
+
+ // Initialize Paimon catalog with the catalog name and options
+ ((CatalogPlugin) catalog)
+ .initialize(this.catalogName + "_paimon", new CaseInsensitiveStringMap(paimonOptions));
+ this.paimonCatalog = catalog;
return this.paimonCatalog;
}
+
+ /**
+ * Ensure the namespace exists in Paimon catalog before creating a table.
+ *
+ * @param namespace the namespace to ensure exists
+ */
+ public void ensureNamespaceExists(String[] namespace) {
+ if (this.paimonCatalog instanceof SupportsNamespaces) {
+ SupportsNamespaces nsSupport = (SupportsNamespaces) this.paimonCatalog;
+ if (!nsSupport.namespaceExists(namespace)) {
+ try {
+ nsSupport.createNamespace(namespace, new HashMap<>());
+ } catch (Exception e) {
Review Comment:
This currently treats every namespace-creation failure as an already-exists race. That can hide authorization, storage/I/O, or unsupported-operation failures and let table creation continue with the original cause lost.
I feel we should only ignore NamespaceAlreadyExistsException (or re-check existence for that race) and propagate other failures?
##########
plugins/spark/v3.5/spark/src/test/java/org/apache/polaris/spark/SparkCatalogTest.java:
##########
@@ -442,21 +443,28 @@ void testCreateAndLoadGenericTable(String format) throws Exception {
Identifier identifier = Identifier.of(defaultNS, "generic-test-table");
createAndValidateGenericTableWithLoad(catalog, identifier, defaultSchema, format);
- Identifier[] icebergTables = catalog.listTables(defaultNS);
- assertThat(icebergTables.length).isEqualTo(1);
- assertThat(icebergTables[0]).isEqualTo(Identifier.of(defaultNS, "generic-test-table"));
+ // Paimon tables are managed separately by PaimonCatalog and not listed in Polaris
+ if (!PolarisCatalogUtils.usePaimon(format)) {
Review Comment:
This skip seems to contradict the contract introduced by this PR. Paimon owns the physical table, but the explicit Polaris registration is added specifically so the table participates in the unified catalog view/SHOW TABLES.
Could we include Paimon in the listing, duplicate-name, and post-drop assertions? As written, a regression that stops registering the Paimon table in Polaris could still pass this test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3864083987
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PaimonHelper.java:
##########
@@ -81,13 +113,33 @@ public TableCatalog loadPaimonCatalog(PolarisSparkCatalog polarisSparkCatalog) {
e);
}
- // Set the Polaris Spark Catalog as the delegate catalog of Paimon Catalog.
- // This allows Paimon to use Polaris for metadata management while handling
- // Paimon-specific operations like manifest and snapshot management.
- if (this.paimonCatalog instanceof DelegatingCatalogExtension) {
- ((DelegatingCatalogExtension) this.paimonCatalog).setDelegateCatalog(polarisSparkCatalog);
- }
+ // Build options for Paimon initialization
+ Map<String, String> paimonOptions = new HashMap<>();
+ paimonOptions.put("warehouse", this.paimonWarehouse);
+
+ // Initialize Paimon catalog with the catalog name and options
+ ((CatalogPlugin) catalog)
+ .initialize(this.catalogName + "_paimon", new CaseInsensitiveStringMap(paimonOptions));
+ this.paimonCatalog = catalog;
return this.paimonCatalog;
}
+
+ /**
+ * Ensure the namespace exists in Paimon catalog before creating a table.
+ *
+ * @param namespace the namespace to ensure exists
+ */
+ public void ensureNamespaceExists(String[] namespace) {
+ if (this.paimonCatalog instanceof SupportsNamespaces) {
+ SupportsNamespaces nsSupport = (SupportsNamespaces) this.paimonCatalog;
+ if (!nsSupport.namespaceExists(namespace)) {
+ try {
+ nsSupport.createNamespace(namespace, new HashMap<>());
+ } catch (Exception e) {
Review Comment:
Agreed. I narrowed the catch to NamespaceAlreadyExistsException and propagate all other failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2877353899
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -143,7 +143,29 @@ public Table loadTable(Identifier ident) throws NoSuchTableException {
try {
return this.icebergsSparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
- return this.polarisSparkCatalog.loadTable(ident);
+ // For generic tables, first try to load from Polaris to check the format
Review Comment:
The design resolves the provider via `polarisSparkCatalog.loadGenericTable(ident)` first, then delegates based on the result, no speculative multi-catalog probing. Delegation is deterministic, it only happens once the provider is known, and keeping Polaris as the single entry point.
Happy to discuss the broader routing strategy on the dev mailing list if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2876737618
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PaimonHelper.java:
##########
@@ -19,75 +19,125 @@
package org.apache.polaris.spark.utils;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.iceberg.common.DynConstructors;
-import org.apache.polaris.spark.PolarisSparkCatalog;
-import org.apache.spark.sql.connector.catalog.DelegatingCatalogExtension;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
/**
* Helper class for integrating Apache Paimon table functionality with Polaris Spark Catalog.
*
- * <p>This class is responsible for dynamically loading and configuring a Paimon Catalog
- * implementation to work with Polaris. It sets up the Paimon Catalog as a delegating catalog
- * extension with Polaris Spark Catalog as the delegate, enabling Paimon table operations through
- * Polaris.
+ * <p>This class is responsible for dynamically loading and configuring a Paimon SparkCatalog
+ * implementation to work with Polaris. Paimon SparkCatalog manages its own table metadata at the
+ * warehouse location following the pattern: warehouse/database.db/table_name.
+ *
+ * <p>Unlike Delta and Hudi which use DelegatingCatalogExtension pattern, Paimon's SparkCatalog is a
+ * standalone catalog that requires proper initialization with a warehouse path. This helper ensures
+ * Paimon is correctly configured using the Polaris warehouse location.
*
* <p>Apache Paimon is a streaming data lake platform with high-speed data ingestion, changelog
* tracking and efficient real-time analytics. This helper enables Polaris to manage Paimon tables
* alongside Iceberg, Delta, and Hudi tables in a unified catalog.
+ *
+ * <p>Configuration options:
+ *
+ * <ul>
+ * <li>{@code paimon-warehouse}: The warehouse path for Paimon tables. This is required for Paimon
+ * to manage table metadata at the location: warehouse/database.db/table_name
+ * <li>{@code paimon-catalog-impl}: Optional custom Paimon SparkCatalog implementation class.
+ * Defaults to org.apache.paimon.spark.SparkCatalog
+ * </ul>
*/
public class PaimonHelper {
public static final String PAIMON_CATALOG_IMPL_KEY = "paimon-catalog-impl";
+ public static final String PAIMON_WAREHOUSE_KEY = "paimon-warehouse";
private static final String DEFAULT_PAIMON_CATALOG_CLASS = "org.apache.paimon.spark.SparkCatalog";
- private TableCatalog paimonCatalog = null;
- private String paimonCatalogImpl = DEFAULT_PAIMON_CATALOG_CLASS;
+ private CatalogPlugin paimonCatalog = null;
+ private final String paimonCatalogImpl;
+ private final String paimonWarehouse;
public PaimonHelper(CaseInsensitiveStringMap options) {
if (options.get(PAIMON_CATALOG_IMPL_KEY) != null) {
this.paimonCatalogImpl = options.get(PAIMON_CATALOG_IMPL_KEY);
+ } else {
+ this.paimonCatalogImpl = DEFAULT_PAIMON_CATALOG_CLASS;
}
+ // Get the Paimon-specific warehouse path from options
+ this.paimonWarehouse = options.get(PAIMON_WAREHOUSE_KEY);
}
/**
- * Load and configure the Paimon catalog with Polaris Spark Catalog as the delegate.
+ * Load and configure the Paimon SparkCatalog with the configured warehouse path.
+ *
+ * <p>Paimon SparkCatalog requires a warehouse path to manage table locations. This method
+ * initializes Paimon with the warehouse path configured via the paimon-warehouse option.
*
- * @param polarisSparkCatalog the Polaris Spark Catalog to set as delegate
+ * @param catalogName the name of the catalog
* @return the configured Paimon TableCatalog
+ * @throws IllegalArgumentException if paimon-warehouse is not configured
*/
- public TableCatalog loadPaimonCatalog(PolarisSparkCatalog polarisSparkCatalog) {
+ public TableCatalog loadPaimonCatalog(String catalogName) {
if (this.paimonCatalog != null) {
Review Comment:
We didnโt have this pattern. `HudiHelper` and `DeltaHelper` also have this issue.
I can help with the follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3864080864
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +191,68 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, use Paimon's SparkCatalog to create the table first,
+ // then register it in Polaris so it appears in the unified catalog view (SHOW TABLES).
+ // Paimon manages its own table paths at warehouse/database.db/table_name
+ // so we don't require a location to be specified.
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ // Ensure namespace exists in Paimon before creating table
+ paimonHelper.ensureNamespaceExists(ident.namespace());
+ Table paimonTable = paimonCatalog.createTable(ident, schema, transforms, properties);
+
+ // Register the table in Polaris as a generic table so it's visible via SHOW TABLES.
+ // Paimon's SparkTable.properties() populates "location" from CoreOptions.PATH,
+ String tableLocation = paimonTable.properties().get(TableCatalog.PROP_LOCATION);
+ if (tableLocation == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Paimon table %s was created but its properties do not contain a location. "
+ + "This is unexpected โ Paimon should always populate the location from its CoreOptions.PATH.",
+ ident));
+ }
+ Map<String, String> registrationProps = new java.util.HashMap<>(properties);
+ registrationProps.put(TableCatalog.PROP_LOCATION, tableLocation);
+
+ try {
+ Tasks.foreach(registrationProps)
+ .retry(MAX_POLARIS_REGISTRATION_RETRIES)
+ .exponentialBackoff(RETRY_MIN_WAIT_MS, RETRY_MAX_WAIT_MS, RETRY_TOTAL_TIMEOUT_MS, 2.0)
+ .shouldRetryTest(
+ e ->
+ !(e instanceof TableAlreadyExistsException)
+ && !(e instanceof NoSuchNamespaceException))
Review Comment:
I preserved the original exception types so the retry classifier now handles them correctly. I also added tests verifying one attempt for non retryable failures, four attempts for retryable failures,and a single rollback.
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +191,68 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, use Paimon's SparkCatalog to create the table first,
+ // then register it in Polaris so it appears in the unified catalog view (SHOW TABLES).
+ // Paimon manages its own table paths at warehouse/database.db/table_name
+ // so we don't require a location to be specified.
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ // Ensure namespace exists in Paimon before creating table
+ paimonHelper.ensureNamespaceExists(ident.namespace());
+ Table paimonTable = paimonCatalog.createTable(ident, schema, transforms, properties);
+
+ // Register the table in Polaris as a generic table so it's visible via SHOW TABLES.
+ // Paimon's SparkTable.properties() populates "location" from CoreOptions.PATH,
+ String tableLocation = paimonTable.properties().get(TableCatalog.PROP_LOCATION);
+ if (tableLocation == null) {
Review Comment:
I agree with your points. I widened the rollback scope to cover every failure after physical Paimon creation. And added the regression test for the missing location case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2876782001
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PaimonHelper.java:
##########
@@ -19,75 +19,125 @@
package org.apache.polaris.spark.utils;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.iceberg.common.DynConstructors;
-import org.apache.polaris.spark.PolarisSparkCatalog;
-import org.apache.spark.sql.connector.catalog.DelegatingCatalogExtension;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
/**
* Helper class for integrating Apache Paimon table functionality with Polaris Spark Catalog.
*
- * <p>This class is responsible for dynamically loading and configuring a Paimon Catalog
- * implementation to work with Polaris. It sets up the Paimon Catalog as a delegating catalog
- * extension with Polaris Spark Catalog as the delegate, enabling Paimon table operations through
- * Polaris.
+ * <p>This class is responsible for dynamically loading and configuring a Paimon SparkCatalog
+ * implementation to work with Polaris. Paimon SparkCatalog manages its own table metadata at the
+ * warehouse location following the pattern: warehouse/database.db/table_name.
+ *
+ * <p>Unlike Delta and Hudi which use DelegatingCatalogExtension pattern, Paimon's SparkCatalog is a
+ * standalone catalog that requires proper initialization with a warehouse path. This helper ensures
+ * Paimon is correctly configured using the Polaris warehouse location.
*
* <p>Apache Paimon is a streaming data lake platform with high-speed data ingestion, changelog
* tracking and efficient real-time analytics. This helper enables Polaris to manage Paimon tables
* alongside Iceberg, Delta, and Hudi tables in a unified catalog.
+ *
+ * <p>Configuration options:
+ *
+ * <ul>
+ * <li>{@code paimon-warehouse}: The warehouse path for Paimon tables. This is required for Paimon
+ * to manage table metadata at the location: warehouse/database.db/table_name
+ * <li>{@code paimon-catalog-impl}: Optional custom Paimon SparkCatalog implementation class.
+ * Defaults to org.apache.paimon.spark.SparkCatalog
+ * </ul>
*/
public class PaimonHelper {
public static final String PAIMON_CATALOG_IMPL_KEY = "paimon-catalog-impl";
+ public static final String PAIMON_WAREHOUSE_KEY = "paimon-warehouse";
private static final String DEFAULT_PAIMON_CATALOG_CLASS = "org.apache.paimon.spark.SparkCatalog";
- private TableCatalog paimonCatalog = null;
- private String paimonCatalogImpl = DEFAULT_PAIMON_CATALOG_CLASS;
+ private CatalogPlugin paimonCatalog = null;
+ private final String paimonCatalogImpl;
+ private final String paimonWarehouse;
public PaimonHelper(CaseInsensitiveStringMap options) {
if (options.get(PAIMON_CATALOG_IMPL_KEY) != null) {
this.paimonCatalogImpl = options.get(PAIMON_CATALOG_IMPL_KEY);
+ } else {
+ this.paimonCatalogImpl = DEFAULT_PAIMON_CATALOG_CLASS;
}
+ // Get the Paimon-specific warehouse path from options
+ this.paimonWarehouse = options.get(PAIMON_WAREHOUSE_KEY);
}
/**
- * Load and configure the Paimon catalog with Polaris Spark Catalog as the delegate.
+ * Load and configure the Paimon SparkCatalog with the configured warehouse path.
+ *
+ * <p>Paimon SparkCatalog requires a warehouse path to manage table locations. This method
+ * initializes Paimon with the warehouse path configured via the paimon-warehouse option.
*
- * @param polarisSparkCatalog the Polaris Spark Catalog to set as delegate
+ * @param catalogName the name of the catalog
* @return the configured Paimon TableCatalog
+ * @throws IllegalArgumentException if paimon-warehouse is not configured
*/
- public TableCatalog loadPaimonCatalog(PolarisSparkCatalog polarisSparkCatalog) {
+ public TableCatalog loadPaimonCatalog(String catalogName) {
Review Comment:
updated!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2876790363
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -186,30 +211,56 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
- return paimonCatalog.alterTable(ident, changes);
- } else {
- return this.polarisSparkCatalog.alterTable(ident);
+ // Try to load from Polaris first
+ try {
+ Table table = this.polarisSparkCatalog.loadTable(ident);
+ String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
+ if (PolarisCatalogUtils.useDelta(provider)) {
+ // For delta table, most of the alter operations is a delta log manipulation,
+ // we load the delta catalog to help handling the alter table operation.
+ // NOTE: This currently doesn't work for changing file location and file format
+ // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
+ TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
+ return deltaCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.useHudi(provider)) {
+ TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
+ return hudiCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ return paimonCatalog.alterTable(ident, changes);
+ } else {
+ return this.polarisSparkCatalog.alterTable(ident);
+ }
+ } catch (NoSuchTableException polarisException) {
+ // Table not found in Polaris, try Paimon directly
+ // Paimon tables created via Paimon's SparkCatalog may not be registered in Polaris
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ return paimonCatalog.alterTable(ident, changes);
+ } catch (Exception paimonException) {
+ // Paimon catalog not configured or table not found there either
+ throw polarisException;
+ }
}
}
}
@Override
public boolean dropTable(Identifier ident) {
- return this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+ boolean dropped =
+ this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+
+ // Also try to drop from Paimon catalog if it exists
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ if (paimonCatalog.dropTable(ident)) {
+ dropped = true;
+ }
+ } catch (Exception e) {
Review Comment:
Fix!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3256552015
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -135,16 +145,42 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
initRESTCatalog(name, options);
this.deltaHelper = new DeltaHelper(options);
this.hudiHelper = new HudiHelper(options);
- this.paimonHelper = new PaimonHelper(options);
+ this.paimonHelper = new PaimonHelper(name, options);
}
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
+ // First, try to load as an Iceberg table
try {
return this.icebergsSparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
- return this.polarisSparkCatalog.loadTable(ident);
+ // Not an Iceberg table, fall through to handle as generic table
}
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ // For Paimon tables, use Paimon's SparkCatalog to load the table
+ // This ensures proper handling of Paimon's metadata and schema files
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ return paimonCatalog.loadTable(ident);
Review Comment:
Hi @dimas-b, I just deleted the integration test and update the PR title and description. Looking forward to your feedback!
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -135,16 +145,42 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
initRESTCatalog(name, options);
this.deltaHelper = new DeltaHelper(options);
this.hudiHelper = new HudiHelper(options);
- this.paimonHelper = new PaimonHelper(options);
+ this.paimonHelper = new PaimonHelper(name, options);
}
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
+ // First, try to load as an Iceberg table
try {
return this.icebergsSparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
- return this.polarisSparkCatalog.loadTable(ident);
+ // Not an Iceberg table, fall through to handle as generic table
}
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ // For Paimon tables, use Paimon's SparkCatalog to load the table
+ // This ensures proper handling of Paimon's metadata and schema files
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ return paimonCatalog.loadTable(ident);
Review Comment:
Hi @dimas-b, I just deleted the integration test and updated the PR title and description. Looking forward to your feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3210234234
##########
plugins/spark/v3.5/integration/build.gradle.kts:
##########
@@ -93,6 +93,25 @@ dependencies {
exclude("org.apache.logging.log4j", "log4j-slf4j-impl")
}
+ // Paimon dependency for Paimon table support
+ // Note: Paimon only publishes paimon-spark-3.5 for Scala 2.12, not 2.13
+ if (scalaVersion == "2.12") {
Review Comment:
Would it be possible to factor Paimon tests out into a separate Gradle module to avoid `if/else` in build scritps?
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -135,16 +145,42 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
initRESTCatalog(name, options);
this.deltaHelper = new DeltaHelper(options);
this.hudiHelper = new HudiHelper(options);
- this.paimonHelper = new PaimonHelper(options);
+ this.paimonHelper = new PaimonHelper(name, options);
}
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
+ // First, try to load as an Iceberg table
try {
return this.icebergsSparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
- return this.polarisSparkCatalog.loadTable(ident);
+ // Not an Iceberg table, fall through to handle as generic table
}
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ // For Paimon tables, use Paimon's SparkCatalog to load the table
+ // This ensures proper handling of Paimon's metadata and schema files
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ return paimonCatalog.loadTable(ident);
Review Comment:
These look like substantial implementation changes, but the PR title mentions only "Add tests" ๐ค Could you update the PR title (and description) to be more aligned with actual code changes?
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PaimonHelper.java:
##########
@@ -19,75 +19,125 @@
package org.apache.polaris.spark.utils;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.iceberg.common.DynConstructors;
-import org.apache.polaris.spark.PolarisSparkCatalog;
-import org.apache.spark.sql.connector.catalog.DelegatingCatalogExtension;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
/**
* Helper class for integrating Apache Paimon table functionality with Polaris Spark Catalog.
*
- * <p>This class is responsible for dynamically loading and configuring a Paimon Catalog
- * implementation to work with Polaris. It sets up the Paimon Catalog as a delegating catalog
- * extension with Polaris Spark Catalog as the delegate, enabling Paimon table operations through
- * Polaris.
+ * <p>This class is responsible for dynamically loading and configuring a Paimon SparkCatalog
+ * implementation to work with Polaris. Paimon SparkCatalog manages its own table metadata at the
+ * warehouse location following the pattern: warehouse/database.db/table_name.
+ *
+ * <p>Unlike Delta and Hudi which use DelegatingCatalogExtension pattern, Paimon's SparkCatalog is a
+ * standalone catalog that requires proper initialization with a warehouse path. This helper ensures
+ * Paimon is correctly configured using the Polaris warehouse location.
*
* <p>Apache Paimon is a streaming data lake platform with high-speed data ingestion, changelog
* tracking and efficient real-time analytics. This helper enables Polaris to manage Paimon tables
* alongside Iceberg, Delta, and Hudi tables in a unified catalog.
+ *
+ * <p>Configuration options:
+ *
+ * <ul>
+ * <li>{@code paimon-warehouse}: The warehouse path for Paimon tables. This is required for Paimon
+ * to manage table metadata at the location: warehouse/database.db/table_name
+ * <li>{@code paimon-catalog-impl}: Optional custom Paimon SparkCatalog implementation class.
+ * Defaults to org.apache.paimon.spark.SparkCatalog
+ * </ul>
*/
public class PaimonHelper {
public static final String PAIMON_CATALOG_IMPL_KEY = "paimon-catalog-impl";
+ public static final String PAIMON_WAREHOUSE_KEY = "paimon-warehouse";
private static final String DEFAULT_PAIMON_CATALOG_CLASS = "org.apache.paimon.spark.SparkCatalog";
- private TableCatalog paimonCatalog = null;
- private String paimonCatalogImpl = DEFAULT_PAIMON_CATALOG_CLASS;
+ private CatalogPlugin paimonCatalog = null;
+ private final String paimonCatalogImpl;
+ private final String paimonWarehouse;
public PaimonHelper(CaseInsensitiveStringMap options) {
if (options.get(PAIMON_CATALOG_IMPL_KEY) != null) {
this.paimonCatalogImpl = options.get(PAIMON_CATALOG_IMPL_KEY);
+ } else {
+ this.paimonCatalogImpl = DEFAULT_PAIMON_CATALOG_CLASS;
}
+ // Get the Paimon-specific warehouse path from options
+ this.paimonWarehouse = options.get(PAIMON_WAREHOUSE_KEY);
}
/**
- * Load and configure the Paimon catalog with Polaris Spark Catalog as the delegate.
+ * Load and configure the Paimon SparkCatalog with the configured warehouse path.
+ *
+ * <p>Paimon SparkCatalog requires a warehouse path to manage table locations. This method
+ * initializes Paimon with the warehouse path configured via the paimon-warehouse option.
*
- * @param polarisSparkCatalog the Polaris Spark Catalog to set as delegate
+ * @param catalogName the name of the catalog
* @return the configured Paimon TableCatalog
+ * @throws IllegalArgumentException if paimon-warehouse is not configured
*/
- public TableCatalog loadPaimonCatalog(PolarisSparkCatalog polarisSparkCatalog) {
+ public TableCatalog loadPaimonCatalog(String catalogName) {
if (this.paimonCatalog != null) {
Review Comment:
If these changes make sense without the new test code, it might be preferable to move them into a separate (smaller) PR to simplify review. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "flyingImer (via GitHub)" <gi...@apache.org>.
flyingImer commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2886867909
##########
plugins/spark/v3.5/integration/src/intTest/java/org/apache/polaris/spark/quarkus/it/SparkPaimonIT.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.spark.quarkus.it;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import java.io.File;
+import java.nio.file.Path;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.polaris.service.it.ext.SparkSessionBuilder;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+@QuarkusIntegrationTest
+public class SparkPaimonIT extends SparkIntegrationBase {
+ private String defaultNs;
+ private String paimonWarehouseDir;
+
+ private String getTableNameWithRandomSuffix() {
+ return generateName("paimontb");
+ }
+
+ @Override
+ protected SparkSession buildSparkSession() {
+ // Get Paimon warehouse path - use parent of Spark warehouse to create a sibling directory
+ paimonWarehouseDir = warehouseDir.resolve("../paimon_warehouse").normalize().toString();
+
+ return SparkSessionBuilder.buildWithTestDefaults()
+ .withExtensions(
+ "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,"
+ + "io.delta.sql.DeltaSparkSessionExtension,"
+ + "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
+ .withConfig(
+ "spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
+ .withWarehouse(warehouseDir)
+ .addCatalog(catalogName, "org.apache.polaris.spark.SparkCatalog", endpoints, sparkToken)
+ // Configure Paimon warehouse for the catalog
+ .withConfig("spark.sql.catalog." + catalogName + ".paimon-warehouse", paimonWarehouseDir)
+ .getOrCreate();
+ }
+
+ @BeforeEach
+ public void createDefaultResources(@TempDir Path tempDir) {
+ spark.sparkContext().setLogLevel("WARN");
+ defaultNs = generateName("paimon");
+ // create a default namespace
+ sql("CREATE NAMESPACE %s", defaultNs);
+ sql("USE NAMESPACE %s", defaultNs);
+ }
+
+ @AfterEach
+ public void cleanupPaimonData() {
+ // clean up paimon data
+ if (paimonWarehouseDir != null) {
+ File dirToDelete = new File(paimonWarehouseDir);
+ FileUtils.deleteQuietly(dirToDelete);
+ }
+ sql("DROP NAMESPACE %s", defaultNs);
+ }
+
+ @Test
+ public void testBasicTableOperations() {
+ // create a regular paimon table
+ // Note: Paimon manages table location internally at warehouse/database.db/table_name
+ String paimontb1 = "paimontb1";
+ sql("CREATE TABLE %s (id INT, name STRING) USING PAIMON", paimontb1);
+ sql("INSERT INTO %s VALUES (1, 'anna'), (2, 'bob')", paimontb1);
+ List<Object[]> results =
+ sql("SELECT id, name FROM %s WHERE id > 1 ORDER BY id DESC", paimontb1);
+ assertThat(results.size()).isEqualTo(1);
+ assertThat(results.get(0)).isEqualTo(new Object[] {2, "bob"});
+
+ // create a paimon table with partition
+ String paimontb2 = "paimontb2";
+ sql(
+ "CREATE TABLE %s (name STRING, age INT, country STRING) USING PAIMON PARTITIONED BY (country)",
+ paimontb2);
+ sql(
+ "INSERT INTO %s VALUES ('anna', 10, 'US'), ('james', 32, 'US'), ('yan', 16, 'CHINA')",
+ paimontb2);
+ results = sql("SELECT name, country FROM %s ORDER BY age", paimontb2);
+ assertThat(results.size()).isEqualTo(3);
+ assertThat(results.get(0)).isEqualTo(new Object[] {"anna", "US"});
+ assertThat(results.get(1)).isEqualTo(new Object[] {"yan", "CHINA"});
+ assertThat(results.get(2)).isEqualTo(new Object[] {"james", "US"});
+
+ // drop tables
+ sql("DROP TABLE %s", paimontb1);
+ sql("DROP TABLE %s", paimontb2);
+ }
+
+ @Test
+ public void testTableUpdate() {
+ // Note: UPDATE with primary key table has compatibility issues between
+ // Paimon 1.0.0 and Spark 3.5's RewriteOperationForRowLineage.
+ // This test demonstrates the basic flow: we test INSERT overwrite instead
+ // which achieves similar functionality.
+ String tableName = getTableNameWithRandomSuffix();
+ sql(
+ "CREATE TABLE %s (id INT, name STRING) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ tableName);
+
+ // insert initial data
+ sql("INSERT INTO %s VALUES (1, 'alice'), (2, 'bob')", tableName);
+
+ // For primary key tables, inserting with same key will update the row
+ sql("INSERT INTO %s VALUES (1, 'charlie')", tableName);
+
+ List<Object[]> results = sql("SELECT * FROM %s ORDER BY id", tableName);
+ assertThat(results.size()).isEqualTo(2);
+ assertThat(results.get(0)).isEqualTo(new Object[] {1, "charlie"});
+ assertThat(results.get(1)).isEqualTo(new Object[] {2, "bob"});
+
+ sql("DROP TABLE %s", tableName);
+ }
+
+ @Test
+ public void testTableDelete() {
+ String tableName = getTableNameWithRandomSuffix();
+ // Paimon requires primary key table for DELETE operations
+ sql(
+ "CREATE TABLE %s (id INT, name STRING) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ tableName);
+
+ sql("INSERT INTO %s VALUES (1, 'alice'), (2, 'bob'), (3, 'charlie')", tableName);
+ sql("DELETE FROM %s WHERE id = 2", tableName);
+
+ List<Object[]> results = sql("SELECT * FROM %s ORDER BY id", tableName);
+ assertThat(results.size()).isEqualTo(2);
+ assertThat(results.get(0)).isEqualTo(new Object[] {1, "alice"});
+ assertThat(results.get(1)).isEqualTo(new Object[] {3, "charlie"});
+
+ sql("DROP TABLE %s", tableName);
+ }
+
+ @Test
+ public void testTableMerge() {
+ String targetTable = getTableNameWithRandomSuffix();
+ String sourceTable = getTableNameWithRandomSuffix();
+
+ // Paimon requires primary key table for MERGE operations
+ sql(
+ "CREATE TABLE %s (id INT, name STRING, age INT) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ targetTable);
+ sql(
+ "CREATE TABLE %s (id INT, name STRING, age INT) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ sourceTable);
+
+ sql("INSERT INTO %s VALUES (1, 'alice', 20), (2, 'bob', 30)", targetTable);
+ sql("INSERT INTO %s VALUES (2, 'bobby', 31), (3, 'charlie', 25)", sourceTable);
+
+ // Merge source into target - use column names directly without table alias prefix in SET
+ sql(
+ "MERGE INTO %s AS t USING %s AS s ON t.id = s.id "
+ + "WHEN MATCHED THEN UPDATE SET name = s.name, age = s.age "
+ + "WHEN NOT MATCHED THEN INSERT *",
+ targetTable, sourceTable);
+
+ List<Object[]> results = sql("SELECT * FROM %s ORDER BY id", targetTable);
+ assertThat(results.size()).isEqualTo(3);
+ assertThat(results.get(0)).isEqualTo(new Object[] {1, "alice", 20});
+ assertThat(results.get(1)).isEqualTo(new Object[] {2, "bobby", 31});
+ assertThat(results.get(2)).isEqualTo(new Object[] {3, "charlie", 25});
+
+ sql("DROP TABLE %s", targetTable);
+ sql("DROP TABLE %s", sourceTable);
+ }
+
+ @Test
+ public void testShowTables() {
+ // Note: Paimon tables managed by Paimon SparkCatalog are not visible via SHOW TABLES
Review Comment:
Hmm, have such pattern been discussed? I'm concerned about this behavior discrepancy due to per provider implementation nuance. I wonder if possible to "enable" Paimon tables managed by Paimon SparkCatalog via SHOW TABLES? what is the drawback?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3820:
URL: https://github.com/apache/polaris/pull/3820#issuecomment-4274992539
This PR is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3214481626
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -135,16 +145,42 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
initRESTCatalog(name, options);
this.deltaHelper = new DeltaHelper(options);
this.hudiHelper = new HudiHelper(options);
- this.paimonHelper = new PaimonHelper(options);
+ this.paimonHelper = new PaimonHelper(name, options);
}
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
+ // First, try to load as an Iceberg table
try {
return this.icebergsSparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
- return this.polarisSparkCatalog.loadTable(ident);
+ // Not an Iceberg table, fall through to handle as generic table
}
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ // For Paimon tables, use Paimon's SparkCatalog to load the table
+ // This ensures proper handling of Paimon's metadata and schema files
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ return paimonCatalog.loadTable(ident);
Review Comment:
I agree this PR has become a mix of Paimon implementation changes and integration test work, so the current title and scope are misleading. Iโm planning to remove the integration-test-specific changes from this PR and narrow it to aligning the Paimon implementation with its non-delegation behavior. Once that implementation is settled, Iโll open a follow-up PR for the Paimon integration tests. If you think that sounds like the better approach, Iโm happy to proceed that way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2876958625
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -186,30 +211,56 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
- return paimonCatalog.alterTable(ident, changes);
- } else {
- return this.polarisSparkCatalog.alterTable(ident);
+ // Try to load from Polaris first
+ try {
+ Table table = this.polarisSparkCatalog.loadTable(ident);
+ String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
+ if (PolarisCatalogUtils.useDelta(provider)) {
+ // For delta table, most of the alter operations is a delta log manipulation,
+ // we load the delta catalog to help handling the alter table operation.
+ // NOTE: This currently doesn't work for changing file location and file format
+ // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
+ TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
+ return deltaCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.useHudi(provider)) {
+ TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
+ return hudiCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ return paimonCatalog.alterTable(ident, changes);
+ } else {
+ return this.polarisSparkCatalog.alterTable(ident);
+ }
+ } catch (NoSuchTableException polarisException) {
+ // Table not found in Polaris, try Paimon directly
+ // Paimon tables created via Paimon's SparkCatalog may not be registered in Polaris
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ return paimonCatalog.alterTable(ident, changes);
+ } catch (Exception paimonException) {
+ // Paimon catalog not configured or table not found there either
+ throw polarisException;
+ }
}
}
}
@Override
public boolean dropTable(Identifier ident) {
- return this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+ boolean dropped =
+ this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+
+ // Also try to drop from Paimon catalog if it exists
Review Comment:
1. Iโve refactored `dropTable` to resolve the Paimon catalog via the provider first, consistent with how load works.
2. Added a validation to ensure the table belongs to the Paimon catalog before deleting, to avoid accidentally dropping unrelated tables with the same identifier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2895903449
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +234,78 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
Review Comment:
In this scenario, we just need to get the format (e.g. "paimon").
For the previous one will triggers full Spark DataSource resolution which need lots of times.
But `getTableFormat` gives the exact same routing information at a fraction of the cost.
We can extract format in
```java
public record GenericTable(
String name,
String format,
String baseLocation,
String doc,
Map<String, String> properties) { ... }
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on PR #3820:
URL: https://github.com/apache/polaris/pull/3820#issuecomment-4400323108
Thanks for your contribution, @iting0321 ! Do you have time / interest in continuing this PR? I hope we can find consensus in the comment threads :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2876744323
##########
plugins/spark/v3.5/integration/src/intTest/java/org/apache/polaris/spark/quarkus/it/SparkPaimonIT.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.spark.quarkus.it;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import java.io.File;
+import java.nio.file.Path;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.polaris.service.it.ext.SparkSessionBuilder;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+@QuarkusIntegrationTest
+public class SparkPaimonIT extends SparkIntegrationBase {
+ private String defaultNs;
+ private String paimonWarehouseDir;
+
+ private String getTableNameWithRandomSuffix() {
+ return generateName("paimontb");
+ }
+
+ @Override
+ protected SparkSession buildSparkSession() {
+ // Get Paimon warehouse path - use parent of Spark warehouse to create a sibling directory
+ paimonWarehouseDir = warehouseDir.resolve("../paimon_warehouse").normalize().toString();
+
+ return SparkSessionBuilder.buildWithTestDefaults()
+ .withExtensions(
+ "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,"
+ + "io.delta.sql.DeltaSparkSessionExtension,"
Review Comment:
I just removed Delta dependency. Thanks for point it out.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2877383499
##########
plugins/spark/v3.5/integration/src/intTest/java/org/apache/polaris/spark/quarkus/it/SparkPaimonIT.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.spark.quarkus.it;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import java.io.File;
+import java.nio.file.Path;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.polaris.service.it.ext.SparkSessionBuilder;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+@QuarkusIntegrationTest
+public class SparkPaimonIT extends SparkIntegrationBase {
+ private String defaultNs;
+ private String paimonWarehouseDir;
+
+ private String getTableNameWithRandomSuffix() {
+ return generateName("paimontb");
+ }
+
+ @Override
+ protected SparkSession buildSparkSession() {
+ // Get Paimon warehouse path - use parent of Spark warehouse to create a sibling directory
+ paimonWarehouseDir = warehouseDir.resolve("../paimon_warehouse").normalize().toString();
+
+ return SparkSessionBuilder.buildWithTestDefaults()
+ .withExtensions(
+ "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,"
+ + "io.delta.sql.DeltaSparkSessionExtension,"
+ + "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
+ .withConfig(
+ "spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
+ .withWarehouse(warehouseDir)
+ .addCatalog(catalogName, "org.apache.polaris.spark.SparkCatalog", endpoints, sparkToken)
+ // Configure Paimon warehouse for the catalog
+ .withConfig("spark.sql.catalog." + catalogName + ".paimon-warehouse", paimonWarehouseDir)
+ .getOrCreate();
+ }
+
+ @BeforeEach
+ public void createDefaultResources(@TempDir Path tempDir) {
+ spark.sparkContext().setLogLevel("WARN");
+ defaultNs = generateName("paimon");
+ // create a default namespace
+ sql("CREATE NAMESPACE %s", defaultNs);
+ sql("USE NAMESPACE %s", defaultNs);
+ }
+
+ @AfterEach
+ public void cleanupPaimonData() {
+ // clean up paimon data
+ if (paimonWarehouseDir != null) {
+ File dirToDelete = new File(paimonWarehouseDir);
+ FileUtils.deleteQuietly(dirToDelete);
+ }
+ sql("DROP NAMESPACE %s", defaultNs);
+ }
+
+ @Test
+ public void testBasicTableOperations() {
+ // create a regular paimon table
+ // Note: Paimon manages table location internally at warehouse/database.db/table_name
+ String paimontb1 = "paimontb1";
+ sql("CREATE TABLE %s (id INT, name STRING) USING PAIMON", paimontb1);
+ sql("INSERT INTO %s VALUES (1, 'anna'), (2, 'bob')", paimontb1);
+ List<Object[]> results =
+ sql("SELECT id, name FROM %s WHERE id > 1 ORDER BY id DESC", paimontb1);
+ assertThat(results.size()).isEqualTo(1);
+ assertThat(results.get(0)).isEqualTo(new Object[] {2, "bob"});
+
+ // create a paimon table with partition
+ String paimontb2 = "paimontb2";
+ sql(
+ "CREATE TABLE %s (name STRING, age INT, country STRING) USING PAIMON PARTITIONED BY (country)",
+ paimontb2);
+ sql(
+ "INSERT INTO %s VALUES ('anna', 10, 'US'), ('james', 32, 'US'), ('yan', 16, 'CHINA')",
+ paimontb2);
+ results = sql("SELECT name, country FROM %s ORDER BY age", paimontb2);
+ assertThat(results.size()).isEqualTo(3);
+ assertThat(results.get(0)).isEqualTo(new Object[] {"anna", "US"});
+ assertThat(results.get(1)).isEqualTo(new Object[] {"yan", "CHINA"});
+ assertThat(results.get(2)).isEqualTo(new Object[] {"james", "US"});
+
+ // drop tables
+ sql("DROP TABLE %s", paimontb1);
+ sql("DROP TABLE %s", paimontb2);
+ }
+
+ @Test
+ public void testTableUpdate() {
+ // Note: UPDATE with primary key table has compatibility issues between
+ // Paimon 1.0.0 and Spark 3.5's RewriteOperationForRowLineage.
+ // This test demonstrates the basic flow: we test INSERT overwrite instead
+ // which achieves similar functionality.
+ String tableName = getTableNameWithRandomSuffix();
+ sql(
+ "CREATE TABLE %s (id INT, name STRING) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ tableName);
+
+ // insert initial data
+ sql("INSERT INTO %s VALUES (1, 'alice'), (2, 'bob')", tableName);
+
+ // For primary key tables, inserting with same key will update the row
+ sql("INSERT INTO %s VALUES (1, 'charlie')", tableName);
+
+ List<Object[]> results = sql("SELECT * FROM %s ORDER BY id", tableName);
+ assertThat(results.size()).isEqualTo(2);
+ assertThat(results.get(0)).isEqualTo(new Object[] {1, "charlie"});
+ assertThat(results.get(1)).isEqualTo(new Object[] {2, "bob"});
+
+ sql("DROP TABLE %s", tableName);
+ }
+
+ @Test
+ public void testTableDelete() {
+ String tableName = getTableNameWithRandomSuffix();
+ // Paimon requires primary key table for DELETE operations
+ sql(
+ "CREATE TABLE %s (id INT, name STRING) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ tableName);
+
+ sql("INSERT INTO %s VALUES (1, 'alice'), (2, 'bob'), (3, 'charlie')", tableName);
+ sql("DELETE FROM %s WHERE id = 2", tableName);
+
+ List<Object[]> results = sql("SELECT * FROM %s ORDER BY id", tableName);
+ assertThat(results.size()).isEqualTo(2);
+ assertThat(results.get(0)).isEqualTo(new Object[] {1, "alice"});
+ assertThat(results.get(1)).isEqualTo(new Object[] {3, "charlie"});
+
+ sql("DROP TABLE %s", tableName);
+ }
+
+ @Test
+ public void testTableMerge() {
+ String targetTable = getTableNameWithRandomSuffix();
+ String sourceTable = getTableNameWithRandomSuffix();
+
+ // Paimon requires primary key table for MERGE operations
+ sql(
+ "CREATE TABLE %s (id INT, name STRING, age INT) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ targetTable);
+ sql(
+ "CREATE TABLE %s (id INT, name STRING, age INT) USING PAIMON TBLPROPERTIES ('primary-key' = 'id')",
+ sourceTable);
+
+ sql("INSERT INTO %s VALUES (1, 'alice', 20), (2, 'bob', 30)", targetTable);
+ sql("INSERT INTO %s VALUES (2, 'bobby', 31), (3, 'charlie', 25)", sourceTable);
+
+ // Merge source into target - use column names directly without table alias prefix in SET
+ sql(
+ "MERGE INTO %s AS t USING %s AS s ON t.id = s.id "
+ + "WHEN MATCHED THEN UPDATE SET name = s.name, age = s.age "
+ + "WHEN NOT MATCHED THEN INSERT *",
+ targetTable, sourceTable);
+
+ List<Object[]> results = sql("SELECT * FROM %s ORDER BY id", targetTable);
+ assertThat(results.size()).isEqualTo(3);
+ assertThat(results.get(0)).isEqualTo(new Object[] {1, "alice", 20});
+ assertThat(results.get(1)).isEqualTo(new Object[] {2, "bobby", 31});
+ assertThat(results.get(2)).isEqualTo(new Object[] {3, "charlie", 25});
+
+ sql("DROP TABLE %s", targetTable);
+ sql("DROP TABLE %s", sourceTable);
+ }
+
+ @Test
+ public void testShowTables() {
+ // Note: Paimon tables managed by Paimon SparkCatalog are not visible via SHOW TABLES
Review Comment:
The original implementation is contradicted the unified catalog goal, now this change introduces explicit dual registration that Paimon remains the physical owner of metadata and data files, while Polaris maintains the logical catalog entry so that listing operations are unified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3864078118
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +276,104 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.useDelta(provider)) {
+ // For delta table, most of the alter operations is a delta log manipulation,
+ // we load the delta catalog to help handling the alter table operation.
+ // NOTE: This currently doesn't work for changing file location and file format
+ // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
+ TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
+ return deltaCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.useHudi(provider)) {
+ TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
+ return hudiCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
return paimonCatalog.alterTable(ident, changes);
- } else {
- return this.polarisSparkCatalog.alterTable(ident);
+ } catch (NoSuchTableException paimonException) {
+ throw paimonException;
+ } catch (Exception paimonException) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to alter Paimon table %s: %s", ident, paimonException.getMessage()),
+ paimonException);
}
}
+
+ // Default: use Polaris catalog for alter operation
+ return this.polarisSparkCatalog.alterTable(ident);
}
@Override
public boolean dropTable(Identifier ident) {
- return this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+ // First, try to drop as an Iceberg table
+ if (this.icebergsSparkCatalog.dropTable(ident)) {
+ return true;
+ }
+
+ // For generic tables, check the provider to delegate to the appropriate catalog
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ try {
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ if (PolarisCatalogUtils.usePaimon(provider)) {
Review Comment:
I added provider-aware Paimon purge routing : purge the physical Paimon table first, then remove the Polaris registration.
Failures preserve or clearly report inconsistent state. Regression tests cover both catalog states and failure paths.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2899403287
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PaimonHelper.java:
##########
@@ -81,13 +112,48 @@ public TableCatalog loadPaimonCatalog(PolarisSparkCatalog polarisSparkCatalog) {
e);
}
- // Set the Polaris Spark Catalog as the delegate catalog of Paimon Catalog.
- // This allows Paimon to use Polaris for metadata management while handling
- // Paimon-specific operations like manifest and snapshot management.
- if (this.paimonCatalog instanceof DelegatingCatalogExtension) {
- ((DelegatingCatalogExtension) this.paimonCatalog).setDelegateCatalog(polarisSparkCatalog);
- }
+ // Build options for Paimon initialization
+ Map<String, String> paimonOptions = new HashMap<>();
+ paimonOptions.put("warehouse", this.paimonWarehouse);
+
+ // Initialize Paimon catalog with the catalog name and options
+ ((CatalogPlugin) catalog)
+ .initialize(catalogName + "_paimon", new CaseInsensitiveStringMap(paimonOptions));
+ this.paimonCatalog = catalog;
return this.paimonCatalog;
}
+
+ /**
+ * Resolve the table location based on Paimon's warehouse convention.
+ *
+ * <p>Paimon stores tables at: warehouse/database.db/table_name
+ *
+ * @param namespace the namespace (database) of the table
+ * @param tableName the name of the table
+ * @return the resolved table location path
+ */
+ public String resolveTableLocation(String[] namespace, String tableName) {
+ // Paimon convention: warehouse/database.db/table_name
+ String dbPath = String.join(".", namespace) + ".db";
+ return this.paimonWarehouse + "/" + dbPath + "/" + tableName;
+ }
Review Comment:
This function was used in `SparkCatalog.java` to reconstruct the path manually(failback). Before this function we will call `String tableLocation = paimonTable.properties().get(PROP_LOCATION);` to take the location.
Now I remove `resolveTableLocation` and throw the error directly.
```
throw new IllegalStateException(
String.format(
"Paimon table %s was created but its properties do not contain a location. "
+ "This is unexpected โ Paimon should always populate the location from its CoreOptions.PATH.",
ident));
}
```
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +234,78 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.useDelta(provider)) {
+ // For delta table, most of the alter operations is a delta log manipulation,
+ // we load the delta catalog to help handling the alter table operation.
+ // NOTE: This currently doesn't work for changing file location and file format
+ // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
+ TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
+ return deltaCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.useHudi(provider)) {
+ TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
+ return hudiCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
return paimonCatalog.alterTable(ident, changes);
- } else {
- return this.polarisSparkCatalog.alterTable(ident);
+ } catch (NoSuchTableException paimonException) {
+ throw paimonException;
+ } catch (Exception paimonException) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to alter Paimon table %s: %s", ident, paimonException.getMessage()),
+ paimonException);
}
}
+
+ // Default: use Polaris catalog for alter operation
+ return this.polarisSparkCatalog.alterTable(ident);
}
@Override
public boolean dropTable(Identifier ident) {
- return this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+ // First, try to drop as an Iceberg table
+ if (this.icebergsSparkCatalog.dropTable(ident)) {
+ return true;
+ }
+
+ // For generic tables, check the provider to delegate to the appropriate catalog
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ try {
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, drop from both Polaris and Paimon catalog
+ boolean droppedFromPolaris = this.polarisSparkCatalog.dropTable(ident);
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ paimonCatalog.dropTable(ident);
+ } catch (Exception e) {
+ LOG.warn("Failed to drop table {} from Paimon catalog: {}", ident, e.getMessage());
Review Comment:
Fix for drop Paimon first then Polaris.
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +185,32 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, use Paimon's SparkCatalog to create the table first,
+ // then register it in Polaris so it appears in the unified catalog view (SHOW TABLES).
+ // Paimon manages its own table paths at warehouse/database.db/table_name
+ // so we don't require a location to be specified.
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ // Ensure namespace exists in Paimon before creating table
+ paimonHelper.ensureNamespaceExists(ident.namespace());
+ Table paimonTable = paimonCatalog.createTable(ident, schema, transforms, properties);
+
+ // Register the table in Polaris as a generic table so it's visible via SHOW TABLES.
+ // Paimon table location follows the convention: warehouse/database.db/table_name
+ String tableLocation = paimonTable.properties().get(TableCatalog.PROP_LOCATION);
+ if (tableLocation == null) {
+ // Fallback: construct location from Paimon warehouse convention
+ tableLocation = paimonHelper.resolveTableLocation(ident.namespace(), ident.name());
+ }
+ Map<String, String> registrationProps = new java.util.HashMap<>(properties);
+ registrationProps.put(TableCatalog.PROP_LOCATION, tableLocation);
+ try {
+ this.polarisSparkCatalog.createTable(ident, schema, transforms, registrationProps);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to register Paimon table {} in Polaris catalog: {}", ident, e.getMessage());
+ }
Review Comment:
Fixed!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on PR #3820:
URL: https://github.com/apache/polaris/pull/3820#issuecomment-4403267364
> Thanks for your contribution, @iting0321 ! Do you have time / interest in continuing this PR? I hope we can find consensus in the comment threads :)
Yes, I'm definitely interested in continuing this PR!
Actually, I opened an issue in Paimon to ask whether they could support [DelegatingCatalogExtension](https://github.com/apache/paimon/issues/7629), but no one has replied yet. Alternatively, Iโm considering implementing DelegatingCatalogExtension for Paimon myself. This would enable end-to-end functionality while avoiding the need for explicit dual registration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #3820: Add integration tests for Paimon
URL: https://github.com/apache/polaris/pull/3820
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3210298515
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -135,16 +145,42 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
initRESTCatalog(name, options);
this.deltaHelper = new DeltaHelper(options);
this.hudiHelper = new HudiHelper(options);
- this.paimonHelper = new PaimonHelper(options);
+ this.paimonHelper = new PaimonHelper(name, options);
}
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
+ // First, try to load as an Iceberg table
try {
return this.icebergsSparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
- return this.polarisSparkCatalog.loadTable(ident);
+ // Not an Iceberg table, fall through to handle as generic table
}
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ // For Paimon tables, use Paimon's SparkCatalog to load the table
+ // This ensures proper handling of Paimon's metadata and schema files
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog();
+ return paimonCatalog.loadTable(ident);
Review Comment:
These look like substantial implementation changes, but the PR title mentions only "Add tests" ๐ค Could you update the PR title to be more aligned with actual code changes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3864085983
##########
plugins/spark/v3.5/spark/src/test/java/org/apache/polaris/spark/SparkCatalogTest.java:
##########
@@ -442,21 +443,28 @@ void testCreateAndLoadGenericTable(String format) throws Exception {
Identifier identifier = Identifier.of(defaultNS, "generic-test-table");
createAndValidateGenericTableWithLoad(catalog, identifier, defaultSchema, format);
- Identifier[] icebergTables = catalog.listTables(defaultNS);
- assertThat(icebergTables.length).isEqualTo(1);
- assertThat(icebergTables[0]).isEqualTo(Identifier.of(defaultNS, "generic-test-table"));
+ // Paimon tables are managed separately by PaimonCatalog and not listed in Polaris
+ if (!PolarisCatalogUtils.usePaimon(format)) {
Review Comment:
I removed the Paimon exclusions so the test now verifies listing, duplicate name rejection, and post drop cleanup for Paimon .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2895903449
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +234,78 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
Review Comment:
In this scenario, we just need to get the format (e.g. Paimon).
For the previous one will triggers full Spark DataSource resolution which need lots of times.
But `getTableFormat` gives the exact same routing information at a fraction of the cost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 closed pull request #3820: Integrate Paimon with Polaris without delegation mode
URL: https://github.com/apache/polaris/pull/3820
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "flyingImer (via GitHub)" <gi...@apache.org>.
flyingImer commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r4067440741
##########
plugins/spark/common/src/main/java/org/apache/polaris/spark/BasePolarisSparkCatalog.java:
##########
@@ -222,11 +351,47 @@ public void invalidateTable(Identifier ident) {
@Override
public boolean purgeTable(Identifier ident) {
- if (this.icebergsSparkCatalog.purgeTable(ident)) {
- return true;
- } else {
- return this.polarisSparkCatalog.purgeTable(ident);
+ String provider;
+ try {
+ provider = this.polarisSparkCatalog.getTableFormat(ident);
Review Comment:
IIUC, this makes Iceberg purge depend on Generic Table support. With `ENABLE_GENERIC_TABLES=false`, getTableFormat calls PolarisRESTCatalog.loadGenericTable, whose endpoint check throws UnsupportedOperationException. Since this catch only handles NoSuchTableException, an existing Iceberg table can no longer be purged, although the previous Iceberg-first path supported it. Could we preserve Iceberg purge when Generic Tables are disabled, keep the Paimon cleanup ordering, and add a regression for that configuration?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r4151140774
##########
plugins/spark/common/src/main/java/org/apache/polaris/spark/BasePolarisSparkCatalog.java:
##########
@@ -222,11 +351,51 @@ public void invalidateTable(Identifier ident) {
@Override
public boolean purgeTable(Identifier ident) {
- if (this.icebergsSparkCatalog.purgeTable(ident)) {
- return true;
- } else {
- return this.polarisSparkCatalog.purgeTable(ident);
+ if (this.icebergsSparkCatalog.tableExists(ident)) {
+ return this.icebergsSparkCatalog.purgeTable(ident);
+ }
+
+ String provider;
+ try {
+ provider = this.polarisSparkCatalog.getTableFormat(ident);
+ } catch (NoSuchTableException e) {
+ return this.icebergsSparkCatalog.purgeTable(ident);
Review Comment:
How can this work, given lines 354-355? ๐ค
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b merged PR #3820:
URL: https://github.com/apache/polaris/pull/3820
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2895903449
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +234,78 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
Review Comment:
In this scenario, we just need to get the format (e.g. "paimon").
For the previous one will triggers full Spark DataSource resolution which need lots of times.
But `getTableFormat` gives the exact same routing information at a fraction of the cost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integration tests for Paimon [polaris]
Posted by "flyingImer (via GitHub)" <gi...@apache.org>.
flyingImer commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r2887022157
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/utils/PaimonHelper.java:
##########
@@ -81,13 +112,48 @@ public TableCatalog loadPaimonCatalog(PolarisSparkCatalog polarisSparkCatalog) {
e);
}
- // Set the Polaris Spark Catalog as the delegate catalog of Paimon Catalog.
- // This allows Paimon to use Polaris for metadata management while handling
- // Paimon-specific operations like manifest and snapshot management.
- if (this.paimonCatalog instanceof DelegatingCatalogExtension) {
- ((DelegatingCatalogExtension) this.paimonCatalog).setDelegateCatalog(polarisSparkCatalog);
- }
+ // Build options for Paimon initialization
+ Map<String, String> paimonOptions = new HashMap<>();
+ paimonOptions.put("warehouse", this.paimonWarehouse);
+
+ // Initialize Paimon catalog with the catalog name and options
+ ((CatalogPlugin) catalog)
+ .initialize(catalogName + "_paimon", new CaseInsensitiveStringMap(paimonOptions));
+ this.paimonCatalog = catalog;
return this.paimonCatalog;
}
+
+ /**
+ * Resolve the table location based on Paimon's warehouse convention.
+ *
+ * <p>Paimon stores tables at: warehouse/database.db/table_name
+ *
+ * @param namespace the namespace (database) of the table
+ * @param tableName the name of the table
+ * @return the resolved table location path
+ */
+ public String resolveTableLocation(String[] namespace, String tableName) {
+ // Paimon convention: warehouse/database.db/table_name
+ String dbPath = String.join(".", namespace) + ".db";
+ return this.paimonWarehouse + "/" + dbPath + "/" + tableName;
+ }
Review Comment:
It seems to be us poking around Paimon internals, is there any Paimon offering alternatives? like `paimonTable.properties().get(PROP_LOCATION)`?
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +234,78 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ // Delegate to the appropriate catalog based on the provider
+ if (PolarisCatalogUtils.useDelta(provider)) {
+ // For delta table, most of the alter operations is a delta log manipulation,
+ // we load the delta catalog to help handling the alter table operation.
+ // NOTE: This currently doesn't work for changing file location and file format
+ // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
+ TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
+ return deltaCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.useHudi(provider)) {
+ TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
+ return hudiCatalog.alterTable(ident, changes);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
return paimonCatalog.alterTable(ident, changes);
- } else {
- return this.polarisSparkCatalog.alterTable(ident);
+ } catch (NoSuchTableException paimonException) {
+ throw paimonException;
+ } catch (Exception paimonException) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to alter Paimon table %s: %s", ident, paimonException.getMessage()),
+ paimonException);
}
}
+
+ // Default: use Polaris catalog for alter operation
+ return this.polarisSparkCatalog.alterTable(ident);
}
@Override
public boolean dropTable(Identifier ident) {
- return this.icebergsSparkCatalog.dropTable(ident) || this.polarisSparkCatalog.dropTable(ident);
+ // First, try to drop as an Iceberg table
+ if (this.icebergsSparkCatalog.dropTable(ident)) {
+ return true;
+ }
+
+ // For generic tables, check the provider to delegate to the appropriate catalog
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ try {
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
+
+ if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, drop from both Polaris and Paimon catalog
+ boolean droppedFromPolaris = this.polarisSparkCatalog.dropTable(ident);
+ try {
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ paimonCatalog.dropTable(ident);
+ } catch (Exception e) {
+ LOG.warn("Failed to drop table {} from Paimon catalog: {}", ident, e.getMessage());
Review Comment:
Current impl would make the table becomes an orphan if Polaris successfully drops but Paimon fails. We should do better on the failure mode here: 1) drop Paimon first then Polaris; 2) or drop Polaris first and roll back Polaris if Paimon drop fails
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +185,32 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
+ // For Paimon tables, use Paimon's SparkCatalog to create the table first,
+ // then register it in Polaris so it appears in the unified catalog view (SHOW TABLES).
+ // Paimon manages its own table paths at warehouse/database.db/table_name
+ // so we don't require a location to be specified.
+ TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.catalogName);
+ // Ensure namespace exists in Paimon before creating table
+ paimonHelper.ensureNamespaceExists(ident.namespace());
+ Table paimonTable = paimonCatalog.createTable(ident, schema, transforms, properties);
+
+ // Register the table in Polaris as a generic table so it's visible via SHOW TABLES.
+ // Paimon table location follows the convention: warehouse/database.db/table_name
+ String tableLocation = paimonTable.properties().get(TableCatalog.PROP_LOCATION);
+ if (tableLocation == null) {
+ // Fallback: construct location from Paimon warehouse convention
+ tableLocation = paimonHelper.resolveTableLocation(ident.namespace(), ident.name());
+ }
+ Map<String, String> registrationProps = new java.util.HashMap<>(properties);
+ registrationProps.put(TableCatalog.PROP_LOCATION, tableLocation);
+ try {
+ this.polarisSparkCatalog.createTable(ident, schema, transforms, registrationProps);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to register Paimon table {} in Polaris catalog: {}", ident, e.getMessage());
+ }
Review Comment:
on second thought, I would highly recommend to 1) retry limited times if retryable; 2) roll back paimon table creation and rethrow if not auto recoverable
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -183,33 +234,78 @@ public Table createTable(
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ // First, try to alter as an Iceberg table
try {
return this.icebergsSparkCatalog.alterTable(ident, changes);
} catch (NoSuchTableException e) {
- Table table = this.polarisSparkCatalog.loadTable(ident);
- String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
- if (PolarisCatalogUtils.useDelta(provider)) {
- // For delta table, most of the alter operations is a delta log manipulation,
- // we load the delta catalog to help handling the alter table operation.
- // NOTE: This currently doesn't work for changing file location and file format
- // using ALTER TABLE ...SET LOCATION, and ALTER TABLE ... SET FILEFORMAT.
- TableCatalog deltaCatalog = deltaHelper.loadDeltaCatalog(this.polarisSparkCatalog);
- return deltaCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.useHudi(provider)) {
- TableCatalog hudiCatalog = hudiHelper.loadHudiCatalog(this.polarisSparkCatalog);
- return hudiCatalog.alterTable(ident, changes);
- } else if (PolarisCatalogUtils.usePaimon(provider)) {
- TableCatalog paimonCatalog = paimonHelper.loadPaimonCatalog(this.polarisSparkCatalog);
+ // Not an Iceberg table, fall through to handle as generic table
+ }
+
+ // For generic tables, check the format/provider to decide delegation
+ // Use getTableFormat to avoid triggering Spark DataSource resolution for routing decisions
+ String provider = this.polarisSparkCatalog.getTableFormat(ident);
Review Comment:
info: it is not quite clear to me on why the pattern change from `String provider = table.properties().get(PolarisCatalogUtils.TABLE_PROVIDER_KEY)` to `String provider = this.polarisSparkCatalog.getTableFormat(ident)`, mind elaborating?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r4072192981
##########
plugins/spark/common/src/main/java/org/apache/polaris/spark/BasePolarisSparkCatalog.java:
##########
@@ -222,11 +351,47 @@ public void invalidateTable(Identifier ident) {
@Override
public boolean purgeTable(Identifier ident) {
- if (this.icebergsSparkCatalog.purgeTable(ident)) {
- return true;
- } else {
- return this.polarisSparkCatalog.purgeTable(ident);
+ String provider;
+ try {
+ provider = this.polarisSparkCatalog.getTableFormat(ident);
Review Comment:
Hi @flyingImer, I just let Iceberg purge when Generic Tables are disabled and add corresponding tests. Thanks!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r3864074611
##########
plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java:
##########
@@ -155,6 +191,68 @@ public Table createTable(
String provider = properties.get(PolarisCatalogUtils.TABLE_PROVIDER_KEY);
if (PolarisCatalogUtils.useIceberg(provider)) {
return this.icebergsSparkCatalog.createTable(ident, schema, transforms, properties);
+ } else if (PolarisCatalogUtils.usePaimon(provider)) {
Review Comment:
Thanks for your feedbacks!
Now the Paimon stageCreate, stageReplace, and stageCreateOrReplace now fail before Iceberg/Paimon side effects.
And I also add the corresponding regression test for SQL CTAS/RTAS.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on PR #3820:
URL: https://github.com/apache/polaris/pull/3820#issuecomment-5427623510
> Thanks for carrying the earlier review feedback through. The current version is much closer to the direction discussed on the dev list, and Iโm fine treating a generalized provider router and broader idempotency improvements as follow-ups rather than blockers here.
>
> I still donโt think this is ready to merge yet. I left a few inline comments on lifecycle correctness and failure handling that I think need to be addressed first.
>
> One additional merge requirement is that this branch predates the Spark common-module refactor, so the implementation should be rebased and ported to the current shared Spark catalog structure, with coverage for both Spark 3.5 and 4.0.
>
> From my side, the merge bar is:
>
> * address the lifecycle/failure-handling issues called out inline;
> * add regression coverage for the Paimon unified-catalog behavior introduced by this PR;
> * rebase/port the implementation to the current shared Spark catalog structure and verify both supported Spark versions;
> * document the required Paimon configuration and any intentionally unsupported operations.
>
> The more general routing abstraction can remain follow-up work.
Hi @flyingImer, thanks for the review!
I already ported the implementation to the shared Spark catalog module used by both Spark 3.5 and 4.0. And also documented the required Paimon warehouse configuration and unsupported staged/rename operations in README.md.
For the regression test, I added unified-catalog, rollback, retry, purge, and CTAS/RTAS regression coverage for both Spark versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Integrate Paimon with Polaris without delegation mode [polaris]
Posted by "iting0321 (via GitHub)" <gi...@apache.org>.
iting0321 commented on code in PR #3820:
URL: https://github.com/apache/polaris/pull/3820#discussion_r4241269363
##########
plugins/spark/common/src/main/java/org/apache/polaris/spark/BasePolarisSparkCatalog.java:
##########
@@ -222,11 +351,51 @@ public void invalidateTable(Identifier ident) {
@Override
public boolean purgeTable(Identifier ident) {
- if (this.icebergsSparkCatalog.purgeTable(ident)) {
- return true;
- } else {
- return this.polarisSparkCatalog.purgeTable(ident);
+ if (this.icebergsSparkCatalog.tableExists(ident)) {
+ return this.icebergsSparkCatalog.purgeTable(ident);
+ }
+
+ String provider;
+ try {
+ provider = this.polarisSparkCatalog.getTableFormat(ident);
+ } catch (NoSuchTableException e) {
+ return this.icebergsSparkCatalog.purgeTable(ident);
Review Comment:
Sorry for the miss, I think it can just return false.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org