Add data-availability triggering for TableTriggers by ryannedolan · Pull Request #237 · linkedin/Hoptimator

ryannedolan · 2026-07-01T14:26:05Z

Summary

InputWatermarkSource capability for Databases.
FIRE TRIGGER ... FROM ... TO
ad hoc backfill support
Automatic late data repair via automatic backfills
Demo Kafka event source

Details

Introduces a uniform way for TableTriggers to fire on arbitrary data change events. This is accomplished via a new interface InputWatermarkSource, which enables adapters to provide change events for specific storage systems. The Kafka adapter is extended as a basic example.

Supports on-demand backfills and automatic repair of late/out-of-order data. On-demand backfills can be triggered with FIRE TRIGGER...FROM...TO. Automatic backfills are scheduled by the reconciler when late data is detected (via InputWatermarkSource). This means that TableTriggers will fire under 3 situations:

New data arrives. This existed prior to this PR, and represents the normal operation of TableTriggers.
Ad hoc backfills via FIRE TRIGGER...FROM...TO. These are just one-off Jobs.
Late data arrives -- data that represents changes before the watermark.

All 3 flows are managed by a new source-agnostic reconciler. (Previously, this was done with bespoke reconcilers for specific sources in our internal deployments.)

Testing Done

Creating a trigger:

0: Hoptimator> create trigger foo on ads.page_views as 'retl-job-template' with ('job.properties.online.table.name' 'foo');
INFO Validating statement: CREATE TRIGGER `FOO` ON `ADS`.`PAGE_VIEWS` AS 'retl-job-template' WITH ('job.properties.online.table.name', 'foo')
INFO Validating trigger FOO with deployers
INFO Validated trigger FOO
INFO Creating trigger FOO
INFO Successfully snapshot K8s obj: TableTrigger:foo
INFO Created K8s obj: TableTrigger:foo
INFO Deployed trigger FOO
INFO CREATE TRIGGER FOO completed
No rows affected (0.1 seconds)

Ad hoc firing a trigger:

0: Hoptimator> fire trigger foo;
INFO Validating statement: FIRE TRIGGER `FOO`
INFO Firing trigger FOO
INFO Updated K8s obj status: TableTrigger:foo
INFO FIRE TRIGGER FOO completed
No rows affected (0.082 seconds)

$ k get jobs -w
NAME                     STATUS     COMPLETIONS   DURATION   AGE
foo-retl-job-template-job   Running    0/1                      0s
foo-retl-job-template-job   Running    0/1           0s         0s
foo-retl-job-template-job   Running    0/1           3s         3s
foo-retl-job-template-job   Complete   1/1           3s         3s

Backfill:

0: Hoptimator> fire trigger foo from 2 days ago to now;
INFO Validating statement: FIRE TRIGGER `FOO` FROM '-2d' TO 'now'
INFO Firing trigger FOO
INFO Capping backfill end for trigger FOO from 2026-07-15T03:32:26Z to the watermark 2026-07-15T03:27:39.141872Z (a backfill cannot cover data the cursor has not yet processed).
INFO Created K8s obj: Job:foo-retl-job-template-job-bf-fc39090c
INFO FIRE TRIGGER FOO completed
No rows affected (0.096 seconds)

$ k logs foo-retl-job-template-job-bf-f59ce552-gnhjf
foo-retl-job-template-trigger processing window [2026-07-13T04:21:08Z, 2026-07-15T04:19:36.962060Z] at Wed Jul 15 04:21:09 UTC 2026

Introduce a uniform way for a TableTrigger to fire when its input is *complete* through a data-time frontier, rather than on wall-clock cron ticks alone, plus on-demand backfills and automatic repair of late/out-of-order data. This is source-agnostic: each external system implements one SPI method instead of a bespoke controller. - InputWatermarkProvider SPI (+ InputWatermarkService, TriggerInput, DataChange): a provider reports completeThrough(input) -> data-time completeness watermark, discovered via ServiceLoader. The generic TableTriggerReconciler advances the trigger's cursor to the frontier and launches the Job over the new window [watermark, timestamp]; sources without a provider keep firing on cron/manual. - FIRE TRIGGER ... FROM ... TO ...: one-off backfills over an explicit output window, run as a separate Job that never moves the incremental cursor. Bounds accept absolute instants/dates, "<n> UNIT AGO", and NOW; the end is capped at the watermark. - Late-change repair: when a provider reports a change behind the watermark, the reconciler enqueues a bounded backfill for that window while keeping the user-facing watermark a monotone forward frontier (status.lateWatermark tracks arrival order internally). - Kafka event source: KafkaInputWatermark reports a topic's max record timestamp as its frontier (AdminClient.listOffsets/maxTimestamp), with a sample trigger. - No lookahead/lookback knobs: a completeness watermark makes "look ahead" unsound and late-repair subsumes "look back", so jobs own any wider trailing read in their own SQL and are handed {{watermark}}/{{timestamp}}. The DDL parser is regenerated from the codegen sources (config.fmpp, includes/parserImpls.ftl) via the Apache Calcite process in hoptimator-jdbc/src/main/codegen/README.md; the FIRE grammar is added and no LOOK BACK/AHEAD grammar is present. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

github-actions · 2026-07-01T14:34:12Z

Code Coverage

Overall Project	84.16% `-0.75%`	🟢
Files changed	76.54%	🟢

File	Coverage
Trigger.java	100%	🟢
InputFrontierSource.java	100%	🟢
Template.java	96.52% `-1.27%`	🟢
K8sTriggerJobs.java	96.31% `-3.69%`	🟢
DeployerUtils.java	93.95% `-6.05%`	🟢
TableTriggerReconciler.java	93.19% `-6.34%`	🟢
K8sTriggerDeployer.java	91.97% `-2.46%`	🟢
ClusterSchema.java	90.61% `-9.39%`	🟢
HoptimatorDdlExecutor.java	84.61% `-4.14%`	❌
DataChange.java	73.17% `-26.83%`	🟢
HoptimatorJdbcSchema.java	65.79% `-31.29%`	❌

…timator into tabletrigger-input-watermarks

@timeout

…vider SPI Replace the ServiceLoader-based InputWatermarkProvider SPI (and its global, per-source config) with an InputWatermarkSource capability implemented by the driver's own Calcite schema -- the object that already carries each Database's connection config. This fixes multi-cluster sources (where each cluster is its own Database) and removes the config-provenance leak, mirroring the existing LogicalSchemaMarker pattern. - New com.linkedin.hoptimator.InputWatermarkSource (api); DataChange moved to api. - HoptimatorJdbcSchema.inputWatermarkSource() resolves the capability by unwrapping the driver's inner schema (parallels isLogical()/detectLogical()); DeployerUtils.jdbcSchema() factored out for the shared resolution step. - ClusterSchema implements InputWatermarkSource, reading a topic's event-time frontier via its own bootstrap.servers -- per-Database, no global property. - TableTriggerReconciler resolves the input's Database via K8sContext.connection() and asks it for the watermark; deletes InputWatermarkProvider/Service/TriggerInput and KafkaInputWatermark. - Capability method is named watermark(String table). - Tests: ClusterSchema watermark unit test + live-Kafka integration test; reworked reconciler watermark test. Docs updated (extending/index, crd-reference). Also on-branch: class-level @timeout(3m) on QuidemTestBase and additional FIRE TRIGGER FROM/TO assertions in k8s-trigger-fire.id. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

srnand

niceee!

srnand · 2026-07-14T16:47:51Z

+        query.put(new TopicPartition(topic, partition.partition()), OffsetSpec.maxTimestamp());
+      }
+      long frontier = Long.MIN_VALUE;
+      for (ListOffsetsResultInfo info : adminClient.listOffsets(query).all().get().values()) {


Shouldn’t the completeness watermark be conservative and use the minimum of all partition's latest timestamps?

Hmm. I think Claude is trying to fire the trigger anytime any data arrives in a Kafka topic, but then why is it called "watermark" and not "frontier"?

confirmed; should be named frontier. This gets confusing. I think Input*Watermark*Source confused us.

Gotcha but I am still worried that the frontier can cause missed records.

If partition-0 sets the watermark to T0 (the highest timestamp) when the job finishes, and later partition-1 produces a record with timestamp T1 where T1 < T0, that record won’t be processed because the watermark has already passed it. Only a manual backfill would include it.

Yeah. I think Kafka ends up being a very poor source for this sort of thing. We need a monotonically increasing T, but with Kafka we cannot make that guarantee across partitions. This PR would treat T1 as late-arriving data, which makes sense for batch workflows but not really for Kafka.

Lemme try to add a lagging lookback window.

Added a lagging window, which I think is as good as we can get here. Basically: a Kafka partition that is idle is ignored until new data arrives, rather than hold back the entire stream. Switched to use min, so we conservatively track all (non-idle) partitions.

yeaa there could still be problems with the lagging window. if one of the partition is a hot partition, all the others would be marked as idle. But agreed, kafka is not a great source for this use case.

I thought the same, but it's a bit better than that. If we have one hot partition and N-1 idle partitions, that's actually totally fine, because it means no records have been sent on N-1 partitions. As soon as even one record is sent to an idle partition, the partition will be considered part of the non-idle set and affect the frontier (assuming that the timestamp is close to now, which ofc is usually the case for new records).

The key here is that we aren't actually consuming these records. We're just observing the greatest timestamp in each partition as reported by the broker. So we don't need to worry about "lagging" partitions in the usual Kafka sense (i.e. a consumer cannot keep up).

The big gap I see is that we totally ignore records with old timestamps. But it would be probably worse if we tried to trigger a backfill every time old records came through. That probably isn't what Kafka consumers would expect.

ahh yess, that makes sense 💡

…FrontierSource) Per PR review: the Kafka/OpenHouse sources report the *latest data-time seen* — an optimistic frontier — not a conservative completeness watermark. The design is "optimistic frontier + late-repair": fire as data appears, heal late/out-of-order writes behind the cursor via changesSince/backfill. Rename the SPI to say what it is, which also frees "watermark" to mean only the trigger's completed-processing cursor (status.watermark, left untouched). - InputWatermarkSource -> InputFrontierSource; watermark(String) -> frontier(String). - HoptimatorJdbcSchema.inputWatermarkSource() -> inputFrontierSource(). - WatermarkSourceResolver -> FrontierSourceResolver. - ClusterSchema implements InputFrontierSource; interface contract reworded (frontier, not completeness; completeness = frontier + repair). - Tests + docs renamed/reworded. CRD status.watermark / lateWatermark unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…optimism Address the review point that "frontier + repair" only closes the completeness gap for repair-capable sources. Kafka's ClusterSchema reports an optimistic max frontier but implements no changesSince, so late/out-of-order writes behind the cursor (lagging partitions; non-monotonic CreateTime) are silently dropped. Rather than make the OSS Kafka demo a full watermarker, say what it is: - InputFrontierSource contract: a source may report an optimistic frontier ONLY if it implements changesSince (completeness = frontier + repair); a source without repair MUST report a conservative frontier or it drops late data. - ClusterSchema javadoc: mark it best-effort/lossy, not a correctness reference; note the conservative-watermark approach a production source would take. - docs/extending + crd-reference: same framing; Kafka is a wiring reference only. No behavior change. OpenHouse (Jasper changesSince) remains the repair-capable, correctness-capable path. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…erator A backfill is a stateless one-off over an explicit window — it touches no cursor, so it is just a Kubernetes Job, not operator state. Both FIRE ... FROM/TO and automatic late-repair now render the trigger's template over the window and create a trigger-owned, window-named Job directly (shared K8sTriggerJobs), then forget it. The Job controller owns the lifecycle: retry via the template's backoffLimit, cleanup via ttlSecondsAfterFinished, and a failed backfill is its own inspectable record (a Failed Job labelled backfill=true). This dissolves the review's silent-loss bug (#2): there is no single backfill slot to serialize on, nothing written to status to advance-past-and-lose, and no delete-on-failure that discards K8s's retry budget. - New K8sTriggerJobs (hoptimator-k8s): render + deterministic naming + createBackfill, shared by the deployer (FIRE) and the reconciler (repair). - K8sTriggerDeployer.fire creates the backfill Job directly (drops the status round-trip and the backfill-in-flight rejection); validation is unchanged. - TableTriggerReconciler repair creates the Job (keyed by window + arrival); removes handleBackfill, the single-slot early-return, and its backfill helpers. - Removes status.backfillFrom/backfillTo from the CRD + generated model. - Tests + docs updated. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

fire/fire.from/fire.to were string keys in the generic Trigger.options() bag — the same bag that carries real pass-through config. Promote the control-plane intent to a first-class, typed field on the object being deployed. - New Trigger.Fire (from/to as OffsetDateTime, nullable): a plain fire has both null; a windowed fire has both set. `Trigger.fire()` returns it (null for any non-fire op). Remove FIRE_OPTION / FIRE_FROM_OPTION / FIRE_TO_OPTION constants. - HoptimatorDdlExecutor builds a Trigger.Fire; resolveFireBound now returns OffsetDateTime (typed end-to-end, no ISO round-trip through the options map). - K8sTriggerDeployer dispatches on `trigger.fire() != null` and reads the typed window (locals renamed from/to); validation unchanged. The executor still has no K8s dependency — it just hands the deployer a typed value object via the Deployable, exactly like the rest of the SPI. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…xcluded) Replace ClusterSchema's optimistic max frontier with a real, conservative completeness watermark: the per-partition minimum of maxTimestamp, held back by a lag B, excluding partitions more than B behind the leader (idle/straggler detection). B is a per-database connection property (frontier.lag.ms, default 5m). - Bounds within-partition out-of-orderness by B. - A partition more than B behind the leader is treated as idle and excluded, so a lagging/dead/sparse partition never stalls the topic. For a typical round-robin stream this is lossless: quiet partitions have no unprocessed data to drop, and when they resume they emit current-time records that land ahead of the cursor. - No repair path needed: completeness comes from the conservative watermark. (Kafka can't detect genuinely-late records via listOffsets anyway — an older-timestamp record doesn't move maxTimestamp — so repair would require a record-reading consumer, out of scope.) frontier.lag.ms is stripped from the AdminClient config so it doesn't warn. InputFrontierSource contract + extending docs updated: Kafka is now the conservative reference, not best-effort/lossy. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Pausing/resuming only sets spec.paused; it never clears status.timestamp/watermark, and the reconciler monitors an already-running job to completion rather than killing it. So "pause/resume to abort" was false — an in-flight execution only resolves when its job completes. Drop that advice from the message. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

A JobTemplate whose yaml references an unresolved {{variable}} renders to null (Template.render intentionally skips such templates, so one template can be selected from many). For CREATE TRIGGER the template is named explicitly via 'as <name>', so a non-render is a user error, not a skip. Previously the null was stored silently as empty trigger yaml. Now toK8sObject() throws a clear SQLException naming the trigger and pointing at the diagnostic WARN log. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

A JobTemplate used via CREATE TRIGGER is rendered twice: once at CREATE to fix the static vars (name/table/path/job.properties.*), and again per-fire to fill the fire window ([watermark, timestamp], or [FROM, TO] for a backfill). The window vars are not known at CREATE, so they must survive that first render. Add a token-level defer predicate to Template.render(env, defer): a deferred variable that the environment cannot resolve is re-emitted as its original {{...}} token verbatim -- transform and condition included -- so a per-fire render still resolves it correctly (e.g. {{watermarkDate toUpperCase}} is not mangled into {{WATERMARKDATE}}). A non-deferred missing variable still skips the whole template, so CREATE TRIGGER's empty-yaml guard still fires on a genuinely missing variable rather than deferring that failure to fire time. K8sTriggerDeployer renders the JobTemplate with K8sTriggerJobs::isWindowVar as the defer predicate (WINDOW_VARS is the single source of truth). Wire the retl-job-template sample to consume the window so backfills are testable. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…early on unresolved vars The first incremental fire has no prior watermark, so render(trigger, null, timestamp) skipped exporting the watermark family, leaving {{watermark}} in the stored Job yaml unresolved. render() then returned null and the reconciler's objFromYaml(null) threw an opaque NullPointerException. withInstantVars now exports a null instant's family as empty strings, so a template that references the window still renders on the first fire (the job decides how to treat the open lower bound). render() also now throws a clear, trigger-named SQLException when the template still resolves to nothing — surfacing a genuinely unprovided {{variable}} instead of letting a null reach objFromYaml. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…etl-job-template The empty-yaml guard (bf32aa1) makes CREATE TRIGGER fail when a JobTemplate resolves to nothing. LogicalTable's implicit offline-tier trigger renders the retl-job-template, which referenced {{job.properties.online.table.name}}. For an offline-only LogicalTable (e.g. LOGICAL-OFFLINE: nearline+offline, no online tier) that variable is unset, so the template skipped, rendered empty, and the guard aborted CREATE TABLE -- breaking logicalTableOfflineDdlScript and logicalTableGraphScript. Give the optional variable an empty default ({{...:}}) so the trigger still renders (with an empty ONLINE_TABLE_NAME) when there is no online tier. Window vars remain deferred verbatim for per-fire resolution. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

With the retl-job-template rendering fix, the implicit offline-tier trigger now carries a populated Job spec instead of empty yaml. The graph builder therefore surfaces the trigger's JobTemplate and container, so the mermaid trigger node gains 'template: retl-job-template-job' and 'container: hello' lines. Update the expected graph to match. Verified against the live kind cluster by rendering the LOGICAL-OFFLINE.MEMBERS graph directly. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…run triggers) The backfill guard assumed a forward-running trigger: it required a watermark and capped the window at it, so a trigger with no watermark could never be backfilled. But the offline-tier trigger is created *paused* by design — its only mode is on-demand backfills — so it never advances a watermark, and every backfill was rejected with 'no watermark yet'. A plain fire made it worse: it bumped timestamp past the (null) watermark, so the in-flight guard then blocked backfills too. A windowed fire is independent of the forward cursor: it launches a one-off Job over an explicit window and bypasses the paused-forward gate. So: - Skip the in-flight guard for windowed fires (it's about the forward path). - Apply the watermark cap/reject only when a watermark actually exists; with no watermark there is no cursor to protect, so run the requested window as-is. - Keep the inverted-window rejection, and keep the in-flight guard for plain fires. This unblocks backfilling an offline-tier logical table — the primary use case. Verified live: a windowed FIRE on a paused, watermark-less offline trigger now creates the backfill Job. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…kfills A windowed FIRE on a trigger with no watermark now runs the requested window instead of being rejected (see previous commit). Update the quidem script: the no-watermark windowed fires now succeed; the inverted-window and invalid-time-bound rejections still hold. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

ryannedolan and others added 7 commits July 13, 2026 16:06

Merge branch 'main' into tabletrigger-input-watermarks

1ee79c6

Fix stale integration test

a53ff95

Merge branch 'tabletrigger-input-watermarks' of personal:linkedin/Hop…

1a57500

…timator into tabletrigger-input-watermarks

Fix mismatched integration test

12638b5

Fix docs churn

4fb446a

Use kind for local testing instead of Docker Desktop

5e31575

srnand reviewed Jul 14, 2026

View reviewed changes

ryannedolan and others added 10 commits July 14, 2026 12:21

Create kind cluster in one place

f1f41ff

ryannedolan marked this pull request as ready for review July 15, 2026 04:36

ryannedolan requested a review from srnand July 15, 2026 04:36

ryannedolan and others added 2 commits July 15, 2026 09:06

ryannedolan force-pushed the tabletrigger-input-watermarks branch from 0f8253c to 163f6dd Compare July 15, 2026 15:02

srnand approved these changes Jul 15, 2026

View reviewed changes

ryannedolan and others added 2 commits July 15, 2026 17:16

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add data-availability triggering for TableTriggers#237

Add data-availability triggering for TableTriggers#237
ryannedolan wants to merge 22 commits into
mainfrom
tabletrigger-input-watermarks

ryannedolan commented Jul 1, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Jul 1, 2026 •

edited

Loading

Uh oh!

srnand left a comment

Uh oh!

srnand Jul 14, 2026

Uh oh!

ryannedolan Jul 14, 2026

Uh oh!

ryannedolan Jul 14, 2026

Uh oh!

srnand Jul 14, 2026

Uh oh!

ryannedolan Jul 14, 2026

Uh oh!

ryannedolan Jul 15, 2026

Uh oh!

srnand Jul 15, 2026

Uh oh!

ryannedolan Jul 15, 2026

Uh oh!

srnand Jul 15, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

ryannedolan commented Jul 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Details

Testing Done

Uh oh!

github-actions Bot commented Jul 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Code Coverage

Uh oh!

srnand left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ryannedolan commented Jul 1, 2026 •

edited

Loading

github-actions Bot commented Jul 1, 2026 •

edited

Loading