Skip to content

Add a generic bounded post commit operations framework - #729

Merged
abhisheknath2011 merged 6 commits into
linkedin:mainfrom
abhisheknath2011:optimizer-stats0
Sep 16, 2026
Merged

abhisheknath2011 merged 6 commits into
linkedin:mainfrom
abhisheknath2011:optimizer-stats0

Conversation

@abhisheknath2011

@abhisheknath2011 abhisheknath2011 commented Sep 14, 2026 •

Copy link
Copy Markdown
Member

Summary

Introduce a generic, best-effort post-commit operations framework in the internal catalog. This is pure infrastructure that lets arbitrary logic run after a successful table commit without the catalog knowing what that logic does, and without any such operation being able to destabilize the commit path.

Motivation (per review on #712): rather than a single-purpose "stats publisher" seam, model post-commit work as a collection of N operations with bounded guarantees. The business logic (commit-stats publishing) is added in follow-up PRs — this PR contains only the framework and its wiring.

Details

New abstractions (iceberg/openhouse/internalcatalog/.../internal/catalog/):

  • PostCommitOperation — interface (getName(), execute(PostCommitContext));
    best-effort, interruptible contract that must never affect commit correctness.
  • PostCommitContext — immutable value object carrying the committed TableIdentifier
    and TableMetadata; operations extract whatever they need.
  • PostCommitOperationRunner (@component) owns the execution safety envelope:
    • Bounded concurrency & memory: fixed-size ThreadPoolExecutor with a bounded queue;
      daemon, named threads that time out when idle (an unused runner holds zero threads).
    • Drop-on-saturation: AbortPolicy sheds work (drop + metric) instead of blocking the
      committer or growing memory unbounded.
    • Hard per-operation timeout: a single-thread timeoutScheduler cancels/interrupts any
      op exceeding the configured timeout, so a slow/hung op cannot pin a worker.
    • Failure isolation: each op runs independently; runAll() never throws.
    • Uniform observability: submitted / success / failed / timeout / rejected counters,
      tagged by operation name.
    • Configurable via cluster.tables.postcommit.{max-threads,queue-capacity,operation-timeout-ms}
      (defaults 4 / 1000 / 10000ms).

Wiring:

  • OpenHouseInternalCatalog autowires the runner and passes it into newTableOps.
  • OpenHouseInternalTableOperations invokes runner.runAll(...) only in the doCommit
    finally SUCCESS branch, so operations fire only after a durably successful commit.
    A backward-compatible 7-arg constructor keeps the runner optional (null-guarded), so
    existing callers/tests are unaffected.

No behavioral change for OSS/dev: those deployments register zero PostCommitOperation
beans, so runAll is a no-op.

Example operation (illustrative)

Adding post-commit behavior is just a Spring bean — no changes to the catalog or
OpenHouseInternalTableOperations. The runner auto-collects every PostCommitOperation
into its List, so presence of the bean is the only switch.

@Component
@Profile("!dev") // production only; OSS/dev registers no ops, so runAll(...) is a no-op
public class CommitStatsPublishOperation implements PostCommitOperation {

  private final CommitStatsPublisher publisher; // transport to the stats/optimizer service

  @Autowired public CommitStatsPublishOperation(CommitStatsPublisher publisher) {
    this.publisher = publisher; }

  @Override public String getName() {
    return "commit-stats-publish"; // low-cardinality; tags all metrics/logs }

  @Override public void execute(PostCommitContext context) {
    Snapshot snapshot = context.getCommittedMetadata().currentSnapshot();
    if (snapshot == null) {
      return;
    }
    Map<String, String> summary = snapshot.summary();
    CommitStats stats =
        CommitStats.builder()
            .tableId(context.getTableIdentifier().toString())
            .snapshotId(snapshot.snapshotId())
            .totalDataFiles(Long.parseLong(summary.getOrDefault("total-data-files", "0")))
            .totalFileSizeBytes(Long.parseLong(summary.getOrDefault("total-files-size", "0")))
            .build();
    publisher.publish(stats); // best-effort; runner enforces timeout + isolation }
}

The runner handles the hard parts around this bean: bounded async execution, the per-op
timeout (interrupts a stuck  publish ), failure isolation, and success/failed/timeout/
rejected metrics tagged  name="commit-stats-publish" . The real implementation lands in a
follow-up PR stacked on this one.

Issue] Briefly discuss the summary of the changes made in this
pull request in 2-3 lines.

Changes

  • Client-facing API Changes
  • Internal API Changes
  • Bug Fixes
  • New Features
  • Performance Improvements
  • Code Style
  • Refactoring
  • Documentation
  • Tests

For all the boxes checked, please include additional details of the changes made in this pull request.

Testing Done

  • New PostCommitOperationRunnerTest: runs all registered ops, isolates a throwing op,
    enforces the per-op timeout (interrupts a slow op), drops overflow work when the pool is
    saturated (rejected metric), and is a safe no-op for null context / empty op list.
  • New OpenHouseInternalTableOperationsTest cases: post-commit ops run exactly once on a
    successful commit (context carries identity + committed metadata) and do NOT run on a
    failed commit.
  • Full :iceberg:openhouse:internalcatalog:test suite passes (Java 17).
  • Manually Tested on local docker setup. Please include commands ran, and their output.
  • Added new tests for the changes made.
  • Updated existing tests to reflect the changes made.
  • No tests added or updated. Please explain why. If unsure, please feel free to ask for help.
  • Some other form of testing like staging or soak time in production. Please explain.

For all the boxes checked, include a detailed description of the testing done for the changes made in this pull request.

Additional Information

  • Breaking Changes
  • Deprecations
  • Large PR broken into smaller PRs, and PR plan linked in the description.

For all the boxes checked, include additional details of the changes made in this pull request.

@mkuchenbecker mkuchenbecker left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we see an example stacked on top?

@abhisheknath2011

abhisheknath2011 commented Sep 15, 2026 •

Copy link
Copy Markdown
Member Author

Can we see an example stacked on top?

Yes added example usage in the PR description.

mkuchenbecker
mkuchenbecker previously approved these changes Sep 16, 2026
@abhisheknath2011
abhisheknath2011 merged commit 1e00f8b into linkedin:main Sep 16, 2026
1 check passed
abhisheknath2011 added a commit that referenced this pull request Sep 17, 2026
…gate (#713)

## Summary
Rework of the Table Optimizer "Stats Bridge" commit-stats step on top of
the merged generic post-commit framework (#729). This PR adds **only**
the stats-specific pieces; all the seam/threading/timeout/wiring is
provided by `PostCommitOperationRunner` from #729.

Previously this PR was stacked on #712 and gated collection at the
**database level**. That
approach is dropped: #712 is superseded by #729, and gating moves to a
**per-table property**.

**New / kept**
- `CommitStats` — neutral, immutable value object, free of
optimizer-client types. Separates
**current-state** totals (`numCurrentFiles`, `tableSizeBytes`) from a
nested **`CommitStats.Delta`**
(`numFilesAdded`/`numFilesDeleted`,
`addedSizeBytes`/`deletedSizeBytes`), plus identity, location,
  version, and table properties.
- `CommitStatsFactory` — extracts `CommitStats` from the committed
`TableMetadata` (`SnapshotSummary`
totals/deltas + OpenHouse canonical properties). Returns empty without a
table UUID; for a commit
that produced **no new snapshot** (metadata-only), current totals and
`delta` are left **null**
(a null delta means "unknown", not zero); tolerates malformed summary
values.
- `AbstractCommitStatsPublishOperation` — a `PostCommitOperation` (from
#729) that extracts and
publishes stats, gated per table. Abstract `publish(CommitStats)` is
implemented by a `@Component`
subclass in the LinkedIn repo; OSS registers no bean, so this is inert
in OSS/dev.

**Gating: database-level → table-property level**
- Removed `TableStatsPublisher`, `NoOpTableStatsPublisher`,
`CommitStatsCollectionGate`,
`ConfigurableCommitStatsCollectionGate`, the
`doCommit()`/`OpenHouseInternalCatalog` wiring, and the
`cluster.tables.optimizer.commit-stats-collection.database-filter`
config — all superseded by #729.
- Collection is enabled per table via the user-settable property
**`optimizer.commitStatsCollectionEnabled=true`**. The key is
intentionally **not**
`openhouse.`-prefixed: `BasePreservedKeyChecker` treats `openhouse.*` as
reserved keys that table
  owners cannot set, which would make the gate un-settable.

**Two-level enablement** (both required to publish):
1. Global runner switch `cluster.tables.postcommit.enabled=true`
(default false, from #729).
2. Per-table opt-in `optimizer.commitStatsCollectionEnabled=true`.


<!--- HINT: Replace #nnn with corresponding Issue number, if you are
fixing an existing issue -->

[Issue](https://github.com/linkedin/openhouse/issues/#nnn)] Briefly
discuss the summary of the changes made in this
pull request in 2-3 lines.

## Changes

- [ ] Client-facing API Changes
- [x] Internal API Changes
- [ ] Bug Fixes
- [x] New Features
- [ ] Performance Improvements
- [ ] Code Style
- [x] Refactoring
- [ ] Documentation
- [x] Tests

For all the boxes checked, please include additional details of the
changes made in this pull request.

## Testing Done
`./gradlew build` passed
- `CommitStatsFactoryTest` — identity/properties-only, snapshot-summary
mapping into nested delta,
empty-without-UUID, malformed-value tolerance, null delta for
snapshot-less commits.
- `AbstractCommitStatsPublishOperationTest` — publishes only when the
table property is `true` **and**
a UUID is present; no publish when the property is absent/false or the
table has no UUID.
- `./gradlew :iceberg:openhouse:internalcatalog:test` (Java 17) passes.
<!--- Check any relevant boxes with "x" -->

- [ ] Manually Tested on local docker setup. Please include commands
ran, and their output.
- [x] Added new tests for the changes made.
- [ ] Updated existing tests to reflect the changes made.
- [ ] No tests added or updated. Please explain why. If unsure, please
feel free to ask for help.
- [ ] Some other form of testing like staging or soak time in
production. Please explain.

For all the boxes checked, include a detailed description of the testing
done for the changes made in this pull request.

# Additional Information

- [ ] Breaking Changes
- [ ] Deprecations
- [ ] Large PR broken into smaller PRs, and PR plan linked in the
description.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants