Introduce idempotency records schema and Postgres-backed IdempotencyStore

[huaxingao]

This PR adds the persistence foundation for REST idempotency in Polaris:
Defines an idempotency_records table (Postgres) with key, binding, liveness, and finalization fields.
Introduces a storage‑agnostic IdempotencyRecord model and IdempotencyStore SPI in polaris-core.
Implements PostgresIdempotencyStore in polaris-relational-jdbc

proposal

discussion thread

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)

[dimas-b]

Please add javadoc. This is a key interface, so describing API contracts is important, especially concurrency and consistency expectations.

[huaxingao]

Java doc added. Thanks!

[dimas-b]

Since the obvious intention in this interface is to provide persistent storage for idempotency records, I suppose putting the new SPI interface into org.apache.polaris.core.persistence might be preferable.

[huaxingao]

Thanks for the suggestion! I have moved new SPI interface to org.apache.polaris.core.persistence

[dimas-b]

@huaxingao : Thanks for starting the implementation for idempoten retries in Polaris! Since this is a major new core feature, please mention this PR on the dev ML for visibility. I know it is part of the previous proposal, but some technical aspects of the new java code might need to be discussed in more depth than the current idempotency proposal doc.

[huaxingao]

Thanks @dimas-b for the comments! I’ll send out a discussion email to the dev mailing list shortly.

[dennishuo]

Looks like a good start! Could you please also re-link your proposal document and the mailing list thread in the PR description for easy navigation and historical posterity?

Two main aspects of the design I'd like to understand better which affect the design decisions in this PR:

1. "Reconciliation flow" - how we distinguish from different IN_PROGRESS error states - where the request-handling crashed before doing any mutations, if it crashed in the middle of attempting mutations, if after successfully mutating but before idempotency-key finalization
2. How we intend to actually use response_summary to ensure reconstructible responses for duplicate requests. Based on your comments about "minimal response" I'm assuming it won't be simply the full response body (good - presumably because LoadTableResponses might be very large and not fit in the database), but I guess we'll need a well-defined convention for defining how different kinds of idempotent requests map to "minimal" response reconstructors. I'm guessing we'd just store a filename to the metadata JSON file for the response to updateTable being able to reconstruct TableMetadata in the response?

I guess the details of those design elements aren't anything that needs to be done on this PR specifically, but we can make sure those are clear in the design document and link to it.

[dennishuo]

Do we have any precedent yet for cross-realm mutations in any of the existing persistence functionality? I see how this could be appealing to purge across all realms in the typical case where the single Postgres database serves all realms, but IIRC it was still open-ended for the persistence factory to be allowed to route to different Postgres databases or even fully different instances for different realms, in which case we'd expect the async job that purges these things to be able to selectively purge for different realms independently.

[huaxingao]

I don’t think we really have any precedent yet for cross‑realm mutations in the persistence APIs – everything else is either scoped by PolarisCallContext or takes an explicit realmId. Given that, I agree it’s safer and more flexible to keep idempotency purging realm‑scoped instead of “all realms in this physical DB.”
I’m going to change the SPI to:

int purgeExpired(String realmId, Instant before);

[dennishuo]

Same as my comment in IdempotencyStore, keeping realm_id as a prefix to this might better reflect the control flow we'd need in the purge driver to support more sophisticated realm routing use cases (e.g. can't assume sophisticated use cases are satisfied with a global cross-realm purge call, but rather might need to be structured as having some realm iterator interface and calling purge individually on each realm).

That said, I guess the realm-splitting design is also something we need to consider for all other async maintenance types of operations, and it's possible I'm forgetting other precedent on cross-realm maintenance. If so, happy to align with that precedent.

[huaxingao]

Good point, i will change to

CREATE INDEX IF NOT EXISTS idx_idemp_realm_expires
  ON idempotency_records (realm_id, expires_at);

[dennishuo]

These field descriptors should also be copied as javadoc comments into the accessors in IdempotencyRecord, since that's where the persistence-agnostic source of truth description of Polaris types/interfaces resides.

[huaxingao]

Good point, I’ll copy those field descriptions into Javadoc on the corresponding getters in IdempotencyRecord

[dennishuo]

My cursory reading of the query used in the postgres impl indicates there might be distinct causes of failing to update the heartbeat -- specifically:

1. If executorId mismatches another in-progress one
2. If already finalized with an HTTP status
3. If the idempotencyKey doesn't already exist

We should probably include these details in the javadoc if these are indeed the intended scenarios. Is it intended for returning false to be a code error case (as in, callers should've already guaranteed none of these situations applies due to calling reserve before), or is it an expected occurrence? If it's only expected in cases of unexpected code bugs, an exception is probably better than a boolean return value.

[huaxingao]

Good catch, you’re right: updateHeartbeat can return false in exactly the three cases you described.

The intent is for these to be normal outcomes, not “exception” cases. From the caller’s point of view, updateHeartbeat is basically “refresh the heartbeat if I still own this in‑progress reservation.” Returning true means “yes, you still own it and we updated the row,” and returning false means “there is no matching in‑progress row you own anymore, so stop heartbeating and treat this reservation as lost.”

Actual failures (SQL errors, connectivity issues, etc.) will still surface as exceptions from the implementation. I’ll update the Javadoc to spell out the three false scenarios and clarify this contract.

[dennishuo]

It seems the three cases are different in terms of whether they should be expected under normal behavior. (1) and (3) in particular seem like code bug cases and (2) seems like the most common "normal" case.

Is the caller supposed to call load after getting a false here to figure out what happened? Maybe the return value should be an IdempotencyRecord already to allow for atomic fetch on error? Or even some kind of HeartbeatResult type of object that could hold more info about the failure.

[huaxingao]

Thanks for the thoughtful suggestions here! I’ve updated the design to make this clearer.

Instead of returning a bare boolean, updateHeartbeat now returns a small HeartbeatResult enum:
UPDATED, FINALIZED, NOT_FOUND, and LOST_OWNERSHIP.

• If the UPDATE matches a row, we return UPDATED.

• If it doesn’t, we do a single load(realmId, key) to distinguish the other cases:

  • no row → NOT_FOUND
  • row with http_status != null → FINALIZED
  • row still IN_PROGRESS but owned by a different executor → LOST_OWNERSHIP.

This way callers don’t have to guess from false or always issue a separate load.

[dennishuo]

Sounds good to me!

[dennishuo]

What's the expected access pattern for this index? Do we expect this one to participate in the updateHeartbeat call? We probably want this to be realm_id prefixed if we really intend to use it outside of a single realm_id, idempotency_key lookup. Like if we're going to add some kind of listRecentIdempotencyKeys call based on heartbeat_at. But then it's not clear whether we really want http_status as the prefix first. Maybe best to add the index in the same PR that adds a use case for it, so I'd just remove this index for now unless I misunderstood the existing query patterns.

If we're just thinking of serving the additional WHERE clauses in update/finalize where we already have realm_id, idempotency_key as predicates we probably don't actually want a whole explicit index on this.

[huaxingao]

Good question. Today the only concrete patterns we have are:

• single‑row lookups by (realm_id, idempotency_key) for load / updateHeartbeat / finalize, which are already handled by the PK, and
• the purge query, which now uses realm_id + expires_at and has its own index.

We don’t yet have an API that actually scans by heartbeat_at. I will remove this for now.

[huaxingao]

@dennishuo Thanks for your comment! I have updated the PR description to include the links to the proposal and the mailing list discussion thread.

"Reconciliation flow" - how we distinguish from different IN_PROGRESS error states

At a high level, we don’t try to infer everything from the idempotency row alone. We use it together with the catalog state: heartbeats/expirations tell us when a row looks stuck, and the catalog tells us whether the underlying mutation actually happened (never ran, ambiguous, or completed).

How we detect rows that need reconciliation

• While a handler is running it periodically calls updateHeartbeat, which keeps heartbeat_at fresh for the owning executor_id.
• A background job (or on‑demand path) scans for rows where http_status IS NULL and either expires_at < now() or heartbeat_at is stale / null for that executor_id. Those rows are “stuck IN_PROGRESS” and get handed to the reconciler, keyed by (operation_type, resource_id).

Case 1 – crashed before doing any durable mutations

• For the first operations we’re targeting (e.g. commit-table), all catalog changes are wrapped in a single metastore transaction, so from the catalog’s point of view there are only two possibilities: the commit exists, or it doesn’t.
• If the reconciler can’t find any evidence of the mutation (no commit / no new metadata, invariants all look unchanged), it treats this as “no durable mutations happened” and simply releases the reservation (delete/expire the row). A future request with the same idempotency key can then reserve again and actually run the operation.

Case 2 – crash mid‑mutation (truly ambiguous)

• With the current single‑transaction model we don’t expect a partially‑committed state: the DB either commits or rolls back.
• If in the future we introduce an operation with multiple independently durable steps, its reconciler would encode invariants over its state and, if it ever saw a partially‑updated / inconsistent view, it would treat that as truly ambiguous and finalize the idempotency row as a terminal error (e.g. idempotency_reconcile_failed). Duplicates would then see a stable failure rather than risk double‑applying side effects.

Case 3 – crashed after successfully mutating but before idempotency‑key finalization

• If the reconciler does see that the mutation clearly completed (e.g. the expected commit/metadata for (operation_type, resource_id) is present and consistent), it treats the operation as logically successful.
• It then reconstructs the “minimal” response from the canonical state plus the stored response_summary, calls finalizeRecord with a 2xx http_status, and any future duplicate requests with the same key just replay that success instead of re‑running the mutation.

So the “never started vs mid‑flight vs finished but not finalized” distinction really comes from the per‑operation reconciliation logic against the catalog; heartbeat_at / executor_id / expires_at only decide when a row is suspicious enough to hand over to that reconciler.

How we intend to actually use response_summary to ensure reconstructible responses for duplicate requests.

For response_summary, the intent is exactly what you’re describing: it’s not meant to store the full HTTP response body, but a small, operation‑specific “replay token” that lets us reconstruct an equivalent response for duplicates.
Concretely:

• Each idempotent endpoint will have a small adapter that defines:

  1. how to take the “real” response and distill it into a compact JSON response_summary plus a whitelisted set of headers serialized into response_headers, and

  2. how to take that summary and reconstruct an HTTP response that is semantically equivalent for the client.

• For “small” responses, that summary can just be the full body (e.g., a tiny JSON result), but for large responses we’ll only store stable identifiers / pointers instead of the entire payload.

• For example, for commit-table / updateTable we would not store the full TableMetadata blob in the idempotency row. Instead, the response_summary would likely contain:

  1. the fully‑qualified table identifier,

  2. a pointer to the metadata location (e.g., the metadata JSON file path / version / snapshot id), and

  3. any other minimal fields needed to reproduce the wire‑level response.

When replaying a duplicate, the reconstructor can follow that pointer back to the metastore / object storage, load the canonical metadata, and build the same logical response that the original call would have produced.

I agree this needs to be spelled out per operation_type, so in the design doc I plan to add a small subsection for each idempotent operation (e.g., commit-table, drop-table, etc.) that defines the shape of its response_summary and how its reconstructor turns that back into an HTTP response.

[dennishuo]

Added some questions and potential follow-ups, but I'm okay with merging and iterating on it from here as an initial pass

[dennishuo]

Since the benefit of putting returning HeartbeatResult directly from the persistence impl instead of making the caller call load afterwards is the conflicting state can be read "atomically" with whatever caused the update to fail.

We probably don't have to over-optimize for those types of edge cases initially though so no action needed. Just leaving this here for posterity.

Optionally, you could add a // TODO to do some kind of atomic or transactional read with the update.

WIth postgres, it might be possible to just use RETURNING to accomplish this, though I'm not a postgres expert so it's just an idea :)

[huaxingao]

Good idea! For now I kept the simple UPDATE + load pattern, but I added a TODO noting a potential future optimization using UPDATE ... RETURNING or a transactional read so we can make this fully atomic if needed later.

[dennishuo]

The most future-proof approach would be to return a struct that could include such an enum in addition to potentially an inline conflicting IdempotencyRecord, error string, etc.

But I guess we'll know better when we have the callsites so I'm okay with iterating on it once we see the entire thing working end-to-end. We should still make sure to minimize needing to change the SPI too significantly in the future though.

[huaxingao]

Makes sense, thanks. For now I’ll keep it as just the HeartbeatResult enum so the SPI stays small, and once we have real call sites we can see if it makes sense to grow this into a richer struct (e.g. result + conflicting IdempotencyRecord).

[dennishuo]

I may have glossed over this case -- I assumed mismatch executorId is normally an unexpected collision on an idempotencyKey. How are executors supposed to force leasing of the idempotency key? Will we actually support different executors/clients "correctly" sharing one idempotency key? And if so why does it differ from a single executor sending (maybe accidental) async retries?

[huaxingao]

Good question. I agree that a mismatched executorId shouldn’t be part of the normal path.
In the intended model, a single executor reserves the key and keeps using the same executorId for its retries, so other executors should see DUPLICATE at reserve time and never call updateHeartbeat for that key. In that sense, LOST_OWNERSHIP is mostly a defensive / diagnostic state.

[huaxingao]

@dennishuo Thanks a lot for reviewing and merging the PR! Really appreciate your help!

I think i also need to change the schema for h2. I will submit a follow-up PR

[singhpk234]

Thank you for the change @huaxingao !
Have a following feedbacks to the change, it would be really nice if we can address / fix forward.

1. we didn't define models for the table being introduced like we do existing tables is it intentional ?
2. implementation glued to PG only and relational DB agnostic, but we have most of the code not opinionated on db specific stuff, this become problematic when extending #3352
3. we should not have v3 schema modified, i believe once we freeze v3 we should only add things which doesn't fail bootstrap
4. we would want to use seperate datasource pools from the one used for managing entities ? also does the idempotency table need to part of same DB. I am a bit concerned of the perf implication this brings in general, i would recommend if we can do some bechmark runs agains polaris to see how the perf is now

[singhpk234]

V3 is already release, we would need v4 now

[singhpk234]

don;t we need realm_id in these index ?

[singhpk234]

can we define model for this ? we have convertors functions for this

[singhpk234]

i wonder if we can leverage query generator for this ?

[singhpk234]

can we use QueryGenerator method to infer table name from model instead of hardcoding it ?

[singhpk234]

can IdempotencyRecord not be a model ? like we have existing model and we just use Convertor abstraction

[singhpk234]

why does this has to be Postgres specific ?

[singhpk234]

This will cause already bootstrapped version with V3 to fail, we should keep this in v4

[singhpk234]

I am not sure if reusing the same datasource where entities are stored is a good idea, can we move idempotency to a different datasource, do we have estimates on how the writes performance regress with this additional db call ?

[singhpk234]

THis should be database agnostic and the profile can be database specific.

[huaxingao]

@singhpk234 Thanks for your comments! I have actually just created a PR to change to V4 schema and also add H2 schema-v4. Will have one more followup PR to address your other questions.