Use trigger state to know if a window is new and optimize reads by arunpandianp · Pull Request #37574 · apache/beam

arunpandianp · 2026-02-12T01:56:04Z

If the trigger finished set is not present, then there is no state associated with the window and the window is new
We persist the finished set even if it is empty, so the next time we read it it won't be null
When the window is new, we clear all state associated with the window. The cleared state can be cached in user worker and windmill caches.

The optimization is not enabled by default and is behind an experiment unstable_not_update_compatible_new_window_optimization. The optimization is enabled only for non merging windows.

This change also fixes a bug where the trigger finished bit sets are always written/cleared due to missing equals() method.

gemini-code-assist · 2026-02-12T01:56:31Z

Summary of Changes

Hello @arunpandianp, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces optimizations for state management within Apache Beam's windowing system, specifically targeting new windows. By leveraging trigger state to determine if a window is being processed for the first time, the system can avoid redundant state reads and writes, leading to improved performance, especially in distributed execution environments like Windmill. The changes include adding methods to identify new windows and to mark state as locally empty, along with corresponding logic in the ReduceFnRunner to utilize these optimizations.

Highlights

New Window Identification: Introduced a mechanism to identify 'new' windows using trigger state, allowing for optimized state management by preventing unnecessary state reads and writes for windows processed for the first time.
Optimized Watermark Hold Clearing: Added a setKnownEmpty method to the WatermarkHoldState interface and its implementations, enabling local marking of watermark state as empty without requiring immediate backend clearing, which can reduce I/O operations.
Conditional State Clearing in ReduceFnRunner: Modified ReduceFnRunner to leverage the new isNew window check, conditionally clearing state for new windows to improve performance, especially in distributed runners like Windmill.
Sentinel Bit for Trigger State: Enhanced TriggerStateMachineRunner to use a sentinel bit when persisting trigger state, allowing the system to accurately distinguish between a truly new window (no state ever persisted) and a window with previously cleared or empty trigger state.
New Test Coverage: Added new test cases to ReduceFnRunnerTest and WindmillStateInternalsTest to validate the new window optimization and the setKnownEmpty functionality.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Changelog

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java
- Added setKnownEmpty() method to InstantState implementation.
runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
- Added getTriggerRunner() and getContextFactory() methods for testing purposes.
- Implemented conditional state clearing for new windows within the processElement method using triggerRunner.isNew().
runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
- Added setKnownEmpty() method to allow local marking of watermark holds as empty without clearing them in the backend.
runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/FinishedTriggersBitSet.java
- Imported @Nullable for nullability annotations.
- Added equals() and hashCode() methods for proper object comparison.
runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java
- Added isNew() method to determine if a window is new based on its trigger state.
- Added getFinishedBits() method for testing.
- Modified persistFinishedSet() to write a sentinel bit for empty trigger states to distinguish new windows from seen-but-empty ones.
- Added isEquivalent() helper method for comparing BitSets, accounting for the sentinel bit.
runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnRunnerTest.java
- Imported BitSet and StateStyle classes.
- Added testNewWindowOptimization() to verify the new window state clearing logic and sentinel bit handling.
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillWatermarkHold.java
- Implemented the setKnownEmpty() method for Windmill-specific watermark holds, optimizing state management.
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternalsTest.java
- Added testWatermarkSetKnownEmptyBeforeRead() to verify setKnownEmpty behavior before reading.
- Added testWatermarkSetKnownEmptyPersist() to confirm setKnownEmpty results in no-op persistence for cleared holds.
sdks/java/core/src/main/java/org/apache/beam/sdk/state/WatermarkHoldState.java
- Added a default setKnownEmpty() method to the WatermarkHoldState interface, marked as internal.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

SentinelBitSetCoder is same as BitSetCoder except that it encodes empty bitset as a single element 0 byte array. This allows checking if the finished bitset is empty or missing. SentinelBitSetCoder and BitSetCoder are state compatible. Both coders can decode encoded bytes from the other coder successfully.

arunpandianp · 2026-04-10T18:07:49Z

-      }
+    @Nullable BitSet currentBits = finishedSetState.read();
+    if (currentBits == null || !currentBits.equals(modifiedFinishedSet.getBitSet())) {
+      finishedSetState.write(modifiedFinishedSet.getBitSet());


Need to make sure this doesn't cause unbounded state growth with global windows

IIUC, repeated triggers in global windows will cause unbounded state growth. It can be avoided by differentiating clear vs delete bitset. Added a TODO, will tackle it separately. Since the change is behind an unstable experiment, it is safe to merge.

I'm not seeing how this change is guarded by the experiment.

Thanks! Fixed it. PTAL.

arunpandianp · 2026-04-27T22:17:23Z

R: @scwhittle

github-actions · 2026-04-27T22:18:38Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

codecov · 2026-04-27T23:59:17Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 55.60%. Comparing base (e01f711) to head (3d8f6c5).
⚠️ Report is 11 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #37574      +/-   ##
============================================
- Coverage     57.23%   55.60%   -1.64%     
+ Complexity     5120     2095    -3025     
============================================
  Files          1396     1102     -294     
  Lines        197997   172412   -25585     
  Branches       4826     1349    -3477     
============================================
- Hits         113326    95870   -17456     
+ Misses        80835    74163    -6672     
+ Partials       3836     2379    -1457

Flag	Coverage Δ
java	`75.91% <ø> (+5.03%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

arunpandianp · 2026-05-02T00:06:17Z

R: @scwhittle

scwhittle · 2026-05-05T20:02:47Z

-      }
+    @Nullable BitSet currentBits = finishedSetState.read();
+    if (currentBits == null || !currentBits.equals(modifiedFinishedSet.getBitSet())) {
+      finishedSetState.write(modifiedFinishedSet.getBitSet());


I'm not seeing how this change is guarded by the experiment.

arunpandianp · 2026-05-14T09:52:28Z

+      return;
+    }
+
    if (!readFinishedBits(finishedSetState).equals(modifiedFinishedSet)) {


Note: before this PR this check always returned true due to equals method missing in FinishedTriggersBitSet. With the added equals method, won't be writing finishedSetState every time.

github-actions Bot added java runners dataflow core labels Feb 12, 2026

Use trigger state to know if a window is new

28957f5

arunpandianp force-pushed the MArkNewWindow branch from baaf7ef to 28957f5 Compare February 14, 2026 03:39

arunpandianp added 2 commits April 10, 2026 04:43

fix style

a4ff2c6

arunpandianp mentioned this pull request Apr 10, 2026

Change TriggerState finished bitset coder to a SentinelBitSetCoder #38139

Merged

arunpandianp added 4 commits April 10, 2026 05:29

fix style

c55b4cb

fix style

cf5050e

fix style

848617b

fix style

493fa8f

arunpandianp commented Apr 10, 2026

View reviewed changes

arunpandianp added 7 commits April 11, 2026 09:01

address comment

0d929f1

Merge branch 'sentinelbitsetcoder' into MArkNewWindow

3ffeca7

Merge remote-tracking branch 'beam/master' into MArkNewWindow

c1e0eba

add tests and improve setKnownEmpty

55b300c

doc fix

8ffe18b

improve persist

ae4a5b2

improve persist

64f8899

arunpandianp changed the title ~~[WIP] Use trigger state to know if a window is new~~ Use trigger state to know if a window is new and optimize reads Apr 27, 2026

arunpandianp marked this pull request as ready for review April 27, 2026 22:17

arunpandianp added 2 commits April 27, 2026 23:05

Enable isNewWindowOptimization on non merging windows only

de2a32e

Add todo to prevent global window state growth

3d8f6c5

spotless

9bb4861

Merge remote-tracking branch 'beam/master' into MArkNewWindow

b2c9e74

scwhittle requested changes May 5, 2026

View reviewed changes

Address comments

168431d

arunpandianp commented May 14, 2026

View reviewed changes

arunpandianp added 2 commits May 14, 2026 09:56

Refactor

ba4766f

Trigger validate runner

ebb5d6f

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use trigger state to know if a window is new and optimize reads#37574

Use trigger state to know if a window is new and optimize reads#37574
arunpandianp wants to merge 21 commits into
apache:masterfrom
arunpandianp:MArkNewWindow

arunpandianp commented Feb 12, 2026 •

edited

Loading

Uh oh!

gemini-code-assist Bot commented Feb 12, 2026

Uh oh!

arunpandianp Apr 10, 2026

Uh oh!

arunpandianp Apr 27, 2026

Uh oh!

scwhittle May 5, 2026

Uh oh!

arunpandianp May 14, 2026

Uh oh!

arunpandianp commented Apr 27, 2026

Uh oh!

github-actions Bot commented Apr 27, 2026

Uh oh!

codecov Bot commented Apr 27, 2026

Uh oh!

arunpandianp commented May 2, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

scwhittle May 5, 2026

Uh oh!

arunpandianp May 14, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

arunpandianp commented Feb 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gemini-code-assist Bot commented Feb 12, 2026

Summary of Changes

Highlights

Footnotes

Uh oh!

arunpandianp Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

arunpandianp Apr 27, 2026

Choose a reason for hiding this comment

Uh oh!

scwhittle May 5, 2026

Choose a reason for hiding this comment

Uh oh!

arunpandianp May 14, 2026

Choose a reason for hiding this comment

Uh oh!

arunpandianp commented Apr 27, 2026

Uh oh!

github-actions Bot commented Apr 27, 2026

Uh oh!

codecov Bot commented Apr 27, 2026

Codecov Report

Uh oh!

arunpandianp commented May 2, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

scwhittle May 5, 2026

Choose a reason for hiding this comment

Uh oh!

arunpandianp May 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

arunpandianp commented Feb 12, 2026 •

edited

Loading

arunpandianp May 14, 2026 •

edited

Loading