Extract IdempotencyRecord data-model change and JDBC schema v5

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

@@ -62,10 +62,10 @@ public ReserveResult reserve(
       insertMap.put(ModelIdempotencyRecord.IDEMPOTENCY_KEY, idempotencyKey);
       insertMap.put(ModelIdempotencyRecord.OPERATION_TYPE, operationType);
       insertMap.put(ModelIdempotencyRecord.RESOURCE_ID, normalizedResourceId);
+      insertMap.put(ModelIdempotencyRecord.PRINCIPAL_HASH, "");
       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));
@@ -197,14 +197,12 @@ public boolean finalizeRecord(
       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));
 

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

@@ -8,8 +8,8 @@
  *
  *   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
+ * 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
@@ -42,46 +42,36 @@ public interface ModelIdempotencyRecord extends Converter<IdempotencyRecord> {
 
   String TABLE_NAME = "idempotency_records";
 
-  // Logical tenant / realm identifier.
   String REALM_ID = "realm_id";
-  // Client-provided idempotency key.
   String IDEMPOTENCY_KEY = "idempotency_key";
-  // Logical operation type (e.g. commit-table).
   String OPERATION_TYPE = "operation_type";
-  // Normalized identifier of the affected resource.
   String RESOURCE_ID = "resource_id";
 
-  // Final HTTP status code once the operation is completed (null while in-progress).
+  // Hash of caller principal identity bound to the reservation.
+  // Compared on replay to prevent cross-principal cache hits.
+  String PRINCIPAL_HASH = "principal_hash";
+
   String HTTP_STATUS = "http_status";
-  // Optional error subtype for failures.
   String ERROR_SUBTYPE = "error_subtype";
-  // Short serialized representation of the response body.
+
   String RESPONSE_SUMMARY = "response_summary";
-  // Serialized representation of response headers.
-  String RESPONSE_HEADERS = "response_headers";
-  // Timestamp when the operation was finalized (null while in-progress).
   String FINALIZED_AT = "finalized_at";
 
-  // Timestamp when the record was created.
   String CREATED_AT = "created_at";
-  // Timestamp when the record was last updated.
   String UPDATED_AT = "updated_at";
-  // Timestamp for the last heartbeat update (null if no heartbeat recorded).
   String HEARTBEAT_AT = "heartbeat_at";
-  // Identifier of the executor that owns the in-progress record (null if not owned).
   String EXECUTOR_ID = "executor_id";
-  // Expiration timestamp after which the record can be considered stale/purgeable.
   String EXPIRES_AT = "expires_at";
 
   List<String> ALL_COLUMNS =
       List.of(
           IDEMPOTENCY_KEY,
           OPERATION_TYPE,
           RESOURCE_ID,
+          PRINCIPAL_HASH,
           HTTP_STATUS,
           ERROR_SUBTYPE,
           RESPONSE_SUMMARY,
-          RESPONSE_HEADERS,
           FINALIZED_AT,
           CREATED_AT,
           UPDATED_AT,
@@ -97,6 +87,8 @@ public interface ModelIdempotencyRecord extends Converter<IdempotencyRecord> {
 
   String getResourceId();
 
+  String getPrincipalHash();
+
   @Nullable
   Integer getHttpStatus();
 
@@ -106,9 +98,6 @@ public interface ModelIdempotencyRecord extends Converter<IdempotencyRecord> {
   @Nullable
   String getResponseSummary();
 
-  @Nullable
-  String getResponseHeaders();
-
   @Nullable
   Instant getFinalizedAt();
 
@@ -131,7 +120,6 @@ default IdempotencyRecord fromResultSet(ResultSet rs) throws SQLException {
 
   /** Convert the current ResultSet row into an {@link IdempotencyRecord}. */
   static IdempotencyRecord fromRow(ResultSet rs) throws SQLException {
-    // Requires realm_id to be projected in the ResultSet.
     return fromRow(rs.getString(REALM_ID), rs);
   }
 
@@ -143,11 +131,11 @@ static IdempotencyRecord fromRow(String realmId, ResultSet rs) throws SQLExcepti
     String idempotencyKey = rs.getString(IDEMPOTENCY_KEY);
     String operationType = rs.getString(OPERATION_TYPE);
     String resourceId = rs.getString(RESOURCE_ID);
+    String principalHash = rs.getString(PRINCIPAL_HASH);
 
     Integer httpStatus = (Integer) rs.getObject(HTTP_STATUS);
     String errorSubtype = rs.getString(ERROR_SUBTYPE);
     String responseSummary = rs.getString(RESPONSE_SUMMARY);
-    String responseHeaders = rs.getString(RESPONSE_HEADERS);
 
     Instant createdAt = rs.getTimestamp(CREATED_AT).toInstant();
     Instant updatedAt = rs.getTimestamp(UPDATED_AT).toInstant();
@@ -166,10 +154,10 @@ static IdempotencyRecord fromRow(String realmId, ResultSet rs) throws SQLExcepti
         idempotencyKey,
         operationType,
         resourceId,
+        principalHash,
         httpStatus,
         errorSubtype,
         responseSummary,
-        responseHeaders,
         createdAt,
         updatedAt,
         finalizedAt,
@@ -184,10 +172,10 @@ default Map<String, Object> toMap(DatabaseType databaseType) {
     map.put(IDEMPOTENCY_KEY, getIdempotencyKey());
     map.put(OPERATION_TYPE, getOperationType());
     map.put(RESOURCE_ID, getResourceId());
+    map.put(PRINCIPAL_HASH, getPrincipalHash());
     map.put(HTTP_STATUS, getHttpStatus());
     map.put(ERROR_SUBTYPE, getErrorSubtype());
     map.put(RESPONSE_SUMMARY, getResponseSummary());
-    map.put(RESPONSE_HEADERS, getResponseHeaders());
     map.put(FINALIZED_AT, getFinalizedAt() == null ? null : Timestamp.from(getFinalizedAt()));
     map.put(CREATED_AT, Timestamp.from(getCreatedAt()));
     map.put(UPDATED_AT, Timestamp.from(getUpdatedAt()));