Add Idempotency-Key support for Iceberg REST createTable

Author: huaxingao

CHANGELOG.md

@@ -30,6 +30,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ### Highlights
 
 ### Upgrade notes
+- The relational-JDBC metastore schema is bumped to v5, which reshapes the (previously unused) `idempotency_records` table for the optimistic-commit idempotency model. New installations bootstrap at v5 automatically. Existing v4 installations are not migrated in place; enabling idempotency requires bootstrapping at v5 or applying the equivalent DDL from `schema-v5.sql`. The v4 `idempotency_records` table was never wired to request handling, so no data migration is needed.
 - Event listeners are now executed on a dedicated executor. **This executor does not propagate the original request's CDI context**; listeners that were improperly relying on that should instead manage their own CDI request scope from now on. Furthermore, two new configuration options were introduced to configure the executor: 
   - `polaris.event-listener.executor.pool-size` configures the thread pool size.
   - `polaris.event-listener.executor.queue-size` configures the queue size for pending events when all threads are busy.
@@ -52,6 +53,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - Added support for `register table` overwrite semantics in the Iceberg REST catalog flow (`overwrite=true`) for internal Polaris catalogs. With overwrite enabled, existing table pointers can be updated to a new metadata location while preserving default behavior for `overwrite=false`.
 - Added `REGISTER_TABLE_OVERWRITE` authorization operation mapped to `TABLE_FULL_METADATA` for deterministic overwrite authorization.
 - Added Polaris Spark 4.0 client.
+- Added handler-level support for the Iceberg REST `Idempotency-Key` header on `createTable`, using an optimistic-commit model: the terminal outcome is recorded only after a successful (2xx) response and retries replay an equivalent response rebuilt from current catalog state (no response body is stored). The key is bound to the full canonicalized request (operation, namespace, name, schema, partition spec, sort order, properties, location and access-delegation modes) and the caller principal, so reusing a key for a different request or by a different caller is rejected with HTTP 422. A retry that loses a concurrent create race replays the winning request instead of returning 409, and a replay returns 422 if the table has advanced beyond the originally-created state. Only successful outcomes are recorded — a retry after a failure simply re-runs the operation. Idempotency is disabled by default and configured under `polaris.idempotency`; records are kept in a standalone idempotency store decoupled from the metastore persistence.
 
 ### Changes
 - Added REPL support to Polaris CLI.

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatabaseType.java

@@ -53,9 +53,9 @@ public String getDisplayName() {
    */
   public int getLatestSchemaVersion() {
     return switch (this) {
-      case POSTGRES -> 4; // PostgreSQL has schemas v1, v2, v3, v4
-      case COCKROACHDB -> 4; // CockroachDB schema version kept in sync with PostgreSQL
-      case H2 -> 4; // H2 uses same schemas as PostgreSQL
+      case POSTGRES -> 5; // PostgreSQL has schemas v1, v2, v3, v4, v5
+      case COCKROACHDB -> 5; // CockroachDB schema version kept in sync with PostgreSQL
+      case H2 -> 5; // H2 uses same schemas as PostgreSQL
     };
   }
 

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/idempotency/RelationalJdbcIdempotencyStore.java

@@ -20,74 +20,36 @@
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.time.Instant;
-import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import javax.sql.DataSource;
 import org.apache.polaris.core.entity.IdempotencyRecord;
 import org.apache.polaris.core.persistence.IdempotencyPersistenceException;
 import org.apache.polaris.core.persistence.IdempotencyStore;
+import org.apache.polaris.persistence.relational.jdbc.DatabaseType;
 import org.apache.polaris.persistence.relational.jdbc.DatasourceOperations;
 import org.apache.polaris.persistence.relational.jdbc.QueryGenerator;
-import org.apache.polaris.persistence.relational.jdbc.RelationalJdbcConfiguration;
 import org.apache.polaris.persistence.relational.jdbc.models.Converter;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelIdempotencyRecord;
 import org.jspecify.annotations.NonNull;
 
+/**
+ * JDBC-backed {@link IdempotencyStore}.
+ *
+ * <p>Implements the "optimistic commit" model: a row is inserted only after the originating
+ * operation has finalized. Race conditions between concurrent retries are detected via the table's
+ * {@code (realm_id, idempotency_key)} primary key — a duplicate INSERT surfaces as a constraint
+ * violation, which we translate into a {@link RecordResultType#DUPLICATE} along with the existing
+ * row.
+ */
 public class RelationalJdbcIdempotencyStore implements IdempotencyStore {
 
   private final DatasourceOperations datasourceOperations;
 
-  public RelationalJdbcIdempotencyStore(
-      @NonNull DataSource dataSource, @NonNull RelationalJdbcConfiguration cfg)
-      throws SQLException {
-    this.datasourceOperations = new DatasourceOperations(dataSource, cfg);
-  }
-
-  @Override
-  public ReserveResult reserve(
-      String realmId,
-      String idempotencyKey,
-      String operationType,
-      String normalizedResourceId,
-      Instant expiresAt,
-      String executorId,
-      Instant now) {
-    try {
-      // Build insert values directly to avoid requiring an Immutables-generated model type.
-      Map<String, Object> insertMap = new LinkedHashMap<>();
-      insertMap.put(ModelIdempotencyRecord.IDEMPOTENCY_KEY, idempotencyKey);
-      insertMap.put(ModelIdempotencyRecord.OPERATION_TYPE, operationType);
-      insertMap.put(ModelIdempotencyRecord.RESOURCE_ID, normalizedResourceId);
-      insertMap.put(ModelIdempotencyRecord.HTTP_STATUS, null);
-      insertMap.put(ModelIdempotencyRecord.ERROR_SUBTYPE, null);
-      insertMap.put(ModelIdempotencyRecord.RESPONSE_SUMMARY, null);
-      insertMap.put(ModelIdempotencyRecord.RESPONSE_HEADERS, null);
-      insertMap.put(ModelIdempotencyRecord.FINALIZED_AT, null);
-      insertMap.put(ModelIdempotencyRecord.CREATED_AT, Timestamp.from(now));
-      insertMap.put(ModelIdempotencyRecord.UPDATED_AT, Timestamp.from(now));
-      insertMap.put(ModelIdempotencyRecord.HEARTBEAT_AT, Timestamp.from(now));
-      insertMap.put(ModelIdempotencyRecord.EXECUTOR_ID, executorId);
-      insertMap.put(ModelIdempotencyRecord.EXPIRES_AT, Timestamp.from(expiresAt));
-
-      List<Object> values = insertMap.values().stream().toList();
-      QueryGenerator.PreparedQuery insert =
-          QueryGenerator.generateInsertQuery(
-              ModelIdempotencyRecord.ALL_COLUMNS,
-              ModelIdempotencyRecord.TABLE_NAME,
-              values,
-              realmId);
-      datasourceOperations.executeUpdate(insert);
-      return new ReserveResult(ReserveResultType.OWNED, Optional.empty());
-    } catch (SQLException e) {
-      if (datasourceOperations.isUniquenessConstraintViolation(e)) {
-        return new ReserveResult(ReserveResultType.DUPLICATE, load(realmId, idempotencyKey));
-      }
-      throw new IdempotencyPersistenceException("Failed to reserve idempotency key", e);
-    }
+  public RelationalJdbcIdempotencyStore(@NonNull DatasourceOperations datasourceOperations) {
+    this.datasourceOperations = datasourceOperations;
   }
 
   @Override
@@ -98,10 +60,8 @@ public Optional<IdempotencyRecord> load(String realmId, String idempotencyKey) {
               ModelIdempotencyRecord.ALL_COLUMNS,
               ModelIdempotencyRecord.TABLE_NAME,
               Map.of(
-                  ModelIdempotencyRecord.REALM_ID,
-                  realmId,
-                  ModelIdempotencyRecord.IDEMPOTENCY_KEY,
-                  idempotencyKey));
+                  ModelIdempotencyRecord.REALM_ID, realmId,
+                  ModelIdempotencyRecord.IDEMPOTENCY_KEY, idempotencyKey));
       List<IdempotencyRecord> results =
           datasourceOperations.executeSelect(
               query,
@@ -112,8 +72,7 @@ public IdempotencyRecord fromResultSet(ResultSet rs) throws SQLException {
                 }
 
                 @Override
-                public Map<String, Object> toMap(
-                    org.apache.polaris.persistence.relational.jdbc.DatabaseType databaseType) {
+                public Map<String, Object> toMap(DatabaseType databaseType) {
                   throw new UnsupportedOperationException("Not used for SELECT conversion");
                 }
               });
@@ -134,99 +93,56 @@ public Map<String, Object> toMap(
   }
 
   @Override
-  public HeartbeatResult updateHeartbeat(
-      String realmId, String idempotencyKey, String executorId, Instant now) {
-    Optional<IdempotencyRecord> existing = load(realmId, idempotencyKey);
-    if (existing.isEmpty()) {
-      return HeartbeatResult.NOT_FOUND;
-    }
-
-    IdempotencyRecord record = existing.get();
-    if (record.httpStatus() != null) {
-      return HeartbeatResult.FINALIZED;
-    }
-    if (record.executorId() == null || !record.executorId().equals(executorId)) {
-      return HeartbeatResult.LOST_OWNERSHIP;
-    }
-
-    QueryGenerator.PreparedQuery update =
-        QueryGenerator.generateUpdateQuery(
-            ModelIdempotencyRecord.ALL_COLUMNS,
-            ModelIdempotencyRecord.TABLE_NAME,
-            Map.of(
-                ModelIdempotencyRecord.HEARTBEAT_AT,
-                Timestamp.from(now),
-                ModelIdempotencyRecord.UPDATED_AT,
-                Timestamp.from(now)),
-            Map.of(
-                ModelIdempotencyRecord.REALM_ID,
-                realmId,
-                ModelIdempotencyRecord.IDEMPOTENCY_KEY,
-                idempotencyKey,
-                ModelIdempotencyRecord.EXECUTOR_ID,
-                executorId),
-            Map.of(),
-            Map.of(),
-            Set.of(ModelIdempotencyRecord.HTTP_STATUS),
-            Set.of());
-
-    try {
-      int updated = datasourceOperations.executeUpdate(update);
-      if (updated > 0) {
-        return HeartbeatResult.UPDATED;
-      }
-    } catch (SQLException e) {
-      throw new IdempotencyPersistenceException("Failed to update idempotency heartbeat", e);
-    }
-
-    // Raced with finalize/ownership loss; re-check to return a meaningful result.
-    Optional<IdempotencyRecord> after = load(realmId, idempotencyKey);
-    if (after.isEmpty()) {
-      return HeartbeatResult.NOT_FOUND;
-    }
-    if (after.get().httpStatus() != null) {
-      return HeartbeatResult.FINALIZED;
-    }
-    return HeartbeatResult.LOST_OWNERSHIP;
-  }
-
-  @Override
-  public boolean finalizeRecord(
+  public RecordResult recordIfAbsent(
       String realmId,
       String idempotencyKey,
-      Integer httpStatus,
-      String errorSubtype,
-      String responseSummary,
-      String responseHeaders,
-      Instant finalizedAt) {
-    // Use ordered/set maps so we can include nullable values (Map.of disallows nulls).
-    Map<String, Object> setClause = new LinkedHashMap<>();
-    setClause.put(ModelIdempotencyRecord.HTTP_STATUS, httpStatus);
-    setClause.put(ModelIdempotencyRecord.ERROR_SUBTYPE, errorSubtype);
-    setClause.put(ModelIdempotencyRecord.RESPONSE_SUMMARY, responseSummary);
-    setClause.put(ModelIdempotencyRecord.RESPONSE_HEADERS, responseHeaders);
-    setClause.put(ModelIdempotencyRecord.FINALIZED_AT, Timestamp.from(finalizedAt));
-    setClause.put(ModelIdempotencyRecord.UPDATED_AT, Timestamp.from(finalizedAt));
-
-    Map<String, Object> whereEquals = new HashMap<>();
-    whereEquals.put(ModelIdempotencyRecord.REALM_ID, realmId);
-    whereEquals.put(ModelIdempotencyRecord.IDEMPOTENCY_KEY, idempotencyKey);
-
-    QueryGenerator.PreparedQuery update =
-        QueryGenerator.generateUpdateQuery(
-            ModelIdempotencyRecord.ALL_COLUMNS,
-            ModelIdempotencyRecord.TABLE_NAME,
-            setClause,
-            whereEquals,
-            Map.of(),
-            Map.of(),
-            Set.of(ModelIdempotencyRecord.HTTP_STATUS),
-            Set.of());
-
+      String operationType,
+      String requestHash,
+      String principalHash,
+      int httpStatus,
+      String metadataLocation,
+      Instant createdAt,
+      Instant expiresAt) {
     try {
-      return datasourceOperations.executeUpdate(update) > 0;
+      Map<String, Object> insertMap = new LinkedHashMap<>();
+      insertMap.put(ModelIdempotencyRecord.IDEMPOTENCY_KEY, idempotencyKey);
+      insertMap.put(ModelIdempotencyRecord.OPERATION_TYPE, operationType);
+      insertMap.put(ModelIdempotencyRecord.REQUEST_HASH, requestHash);
+      insertMap.put(ModelIdempotencyRecord.PRINCIPAL_HASH, principalHash);
+      insertMap.put(ModelIdempotencyRecord.HTTP_STATUS, httpStatus);
+      insertMap.put(ModelIdempotencyRecord.METADATA_LOCATION, metadataLocation);
+      insertMap.put(ModelIdempotencyRecord.CREATED_AT, Timestamp.from(createdAt));
+      insertMap.put(ModelIdempotencyRecord.EXPIRES_AT, Timestamp.from(expiresAt));
+
+      List<Object> values = insertMap.values().stream().toList();
+      QueryGenerator.PreparedQuery insert =
+          QueryGenerator.generateInsertQuery(
+              ModelIdempotencyRecord.ALL_COLUMNS,
+              ModelIdempotencyRecord.TABLE_NAME,
+              values,
+              realmId);
+      datasourceOperations.executeUpdate(insert);
+      return new RecordResult(RecordResultType.OWNED, Optional.empty());
     } catch (SQLException e) {
-      throw new IdempotencyPersistenceException("Failed to finalize idempotency record", e);
+      if (datasourceOperations.isUniquenessConstraintViolation(e)) {
+        Optional<IdempotencyRecord> existing = load(realmId, idempotencyKey);
+        if (existing.isEmpty()) {
+          // The insert lost the race on the (realm_id, idempotency_key) constraint, yet the winning
+          // row is no longer visible (e.g. purged or rolled back between the conflict and this
+          // reload). Surface a persistence error rather than a DUPLICATE without a record, which
+          // the
+          // handler layer treats as an invariant violation.
+          throw new IdempotencyPersistenceException(
+              "Insert for realm/key "
+                  + realmId
+                  + "/"
+                  + idempotencyKey
+                  + " conflicted on the unique constraint but the existing record could not be"
+                  + " reloaded");
+        }
+        return new RecordResult(RecordResultType.DUPLICATE, existing);
+      }
+      throw new IdempotencyPersistenceException("Failed to record idempotency entry", e);
     }
   }
 

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/idempotency/RelationalJdbcIdempotencyStoreFactory.java

@@ -0,0 +1,49 @@
+/*
+ * 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.persistence.relational.jdbc.idempotency;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.persistence.IdempotencyStore;
+import org.apache.polaris.core.persistence.IdempotencyStoreFactory;
+import org.apache.polaris.persistence.relational.jdbc.DatasourceOperations;
+
+/**
+ * {@link IdempotencyStoreFactory} backed by the same JDBC {@link DatasourceOperations} used by the
+ * primary metastore.
+ *
+ * <p>The store is stateless and shared across realms — realm scoping is enforced inside SQL via the
+ * {@code realm_id} column.
+ */
+@ApplicationScoped
+@Identifier("relational-jdbc")
+public class RelationalJdbcIdempotencyStoreFactory implements IdempotencyStoreFactory {
+
+  private final RelationalJdbcIdempotencyStore store;
+
+  @Inject
+  public RelationalJdbcIdempotencyStoreFactory(DatasourceOperations datasourceOperations) {
+    this.store = new RelationalJdbcIdempotencyStore(datasourceOperations);
+  }
+
+  @Override
+  public IdempotencyStore getOrCreateIdempotencyStore(RealmContext realmContext) {
+    return store;
+  }
+}