Skip to content

Process deletes before adds when committing OrderedList state updates. #40659

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shunping merged 1 commit into from
Oct 19, 2026
Merged

Process deletes before adds when committing OrderedList state updates. #40659

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 14 additions & 14 deletions ...main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillOrderedList.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,20 @@ public Windmill.WorkItemCommitRequest persistDirectly(WindmillStateCache.ForKeyA
cleared = false;
}

if (!pendingDeletes.isEmpty()) {

@shunping shunping Oct 19, 2026

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.

I agree that we need to apply pendingDeletes before pendingAdds. When clearRange() is called, it already clears any existing pendingAdds that fall within the deleted range (see the logic in clearRange()).

Therefore, if there is a remaining pendingAdd that overlaps with a pendingDelete range, it means the add operation occurred after the clear operation. Applying the deletes first ensures that these subsequent additions are correctly preserved.

for (Range<Instant> range : pendingDeletes.asRanges()) {
Windmill.TagSortedListDeleteRequest.Builder deletesBuilder =
updatesBuilder.addDeletesBuilder();
deletesBuilder.setRange(
Windmill.SortedListRange.newBuilder()
.setStart(WindmillTimeUtils.harnessToWindmillTimestamp(range.lowerEndpoint()))
.setLimit(WindmillTimeUtils.harnessToWindmillTimestamp(range.upperEndpoint())));
deletesBuilder.build();
idTracker.remove(range);
}
pendingDeletes.clear();
}

if (!pendingAdds.isEmpty()) {
// TODO(reuvenlax): Once we start caching data, we should remove this line. We have it
// here now
Expand Down Expand Up @@ -261,20 +275,6 @@ public Windmill.WorkItemCommitRequest persistDirectly(WindmillStateCache.ForKeyA
pendingAdds.clear();
insertBuilder.build();
}

if (!pendingDeletes.isEmpty()) {
for (Range<Instant> range : pendingDeletes.asRanges()) {
Windmill.TagSortedListDeleteRequest.Builder deletesBuilder =
updatesBuilder.addDeletesBuilder();
deletesBuilder.setRange(
Windmill.SortedListRange.newBuilder()
.setStart(WindmillTimeUtils.harnessToWindmillTimestamp(range.lowerEndpoint()))
.setLimit(WindmillTimeUtils.harnessToWindmillTimestamp(range.upperEndpoint())));
deletesBuilder.build();
idTracker.remove(range);
}
pendingDeletes.clear();
}
} catch (ExecutionException | InterruptedException e) {
throw new RuntimeException(e);
}
Expand Down
105 changes: 105 additions & 0 deletions ...va/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillStateInternalsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2386,6 +2386,111 @@ public void testOrderedListDeleteRangePersist() {
assertBuildable(commitBuilder);
}

@Test
public void testOrderedListDeleteRangeThenAddPersist() throws Exception {
SettableFuture<Map<Range<Instant>, RangeSet<Long>>> orderedListFuture = SettableFuture.create();
orderedListFuture.set(null);
SettableFuture<Map<Range<Instant>, RangeSet<Instant>>> deletionsFuture =
SettableFuture.create();
deletionsFuture.set(null);
when(mockReader.valueFuture(
systemKey(NAMESPACE, "orderedList" + IdTracker.IDS_AVAILABLE_STR),
STATE_FAMILY,
IdTracker.IDS_AVAILABLE_CODER))
.thenReturn(orderedListFuture);
when(mockReader.valueFuture(
systemKey(NAMESPACE, "orderedList" + IdTracker.DELETIONS_STR),
STATE_FAMILY,
IdTracker.SUBRANGE_DELETIONS_CODER))
.thenReturn(deletionsFuture);

StateTag<OrderedListState<String>> addr =
StateTags.orderedList("orderedList", StringUtf8Coder.of());
OrderedListState<String> orderedListState = underTest.state(NAMESPACE, addr);

// Clear a sub-range [2ms, 4ms) and then add an element at 2ms within that cleared range.
orderedListState.clearRange(Instant.ofEpochMilli(2), Instant.ofEpochMilli(4));
orderedListState.add(TimestampedValue.of("hello", Instant.ofEpochMilli(2)));

Windmill.WorkItemCommitRequest.Builder commitBuilder =
Windmill.WorkItemCommitRequest.newBuilder();
underTest.persist(commitBuilder);

assertEquals(1, commitBuilder.getSortedListUpdatesCount());
TagSortedListUpdateRequest updates = commitBuilder.getSortedListUpdates(0);
assertEquals(1, updates.getDeletesCount());
assertEquals(2000, updates.getDeletes(0).getRange().getStart());
assertEquals(4000, updates.getDeletes(0).getRange().getLimit());
assertEquals(1, updates.getInsertsCount());
assertEquals(1, updates.getInserts(0).getEntriesCount());
assertEquals("hello", updates.getInserts(0).getEntries(0).getValue().toStringUtf8());
assertEquals(2000, updates.getInserts(0).getEntries(0).getSortKey());
assertEquals(IdTracker.NEW_RANGE_MIN_ID, updates.getInserts(0).getEntries(0).getId());
assertBuildable(commitBuilder);

// Verify that the sub-range deletion for [2ms, 3ms) was removed when "hello" was added at 2ms,
// so clearing the rest of the 5-minute bucket [0, 2ms) and [4ms, 5m) in the next work item
// does not wipe out id tracking for the bucket.
resetUnderTest();
orderedListState = underTest.state(NAMESPACE, addr);
orderedListState.clearRange(Instant.EPOCH, Instant.ofEpochMilli(2));
orderedListState.clearRange(Instant.ofEpochMilli(4), Instant.ofEpochSecond(300));

commitBuilder = Windmill.WorkItemCommitRequest.newBuilder();
underTest.persist(commitBuilder);
assertBuildable(commitBuilder);

// Adding another element at 2ms in a subsequent work item should get the next id (1), not 0.
resetUnderTest();
orderedListState = underTest.state(NAMESPACE, addr);
orderedListState.add(TimestampedValue.of("world", Instant.ofEpochMilli(2)));

commitBuilder = Windmill.WorkItemCommitRequest.newBuilder();
underTest.persist(commitBuilder);

updates = commitBuilder.getSortedListUpdates(0);
assertEquals(1, updates.getInsertsCount());
assertEquals(1, updates.getInserts(0).getEntriesCount());
assertEquals("world", updates.getInserts(0).getEntries(0).getValue().toStringUtf8());
assertEquals(2000, updates.getInserts(0).getEntries(0).getSortKey());
assertEquals(IdTracker.NEW_RANGE_MIN_ID + 1, updates.getInserts(0).getEntries(0).getId());
assertBuildable(commitBuilder);

// Now clear the entire 5-minute bucket [0, 5m) and add a new element in the same work item.
// The delete should be processed before the add, resetting the bucket's ids to start at
// NEW_RANGE_MIN_ID (0) for the new element while preserving its id tracking for future adds.
resetUnderTest();
orderedListState = underTest.state(NAMESPACE, addr);
orderedListState.clearRange(Instant.EPOCH, Instant.ofEpochSecond(300));
orderedListState.add(TimestampedValue.of("afterFullClear1", Instant.ofEpochMilli(2)));

commitBuilder = Windmill.WorkItemCommitRequest.newBuilder();
underTest.persist(commitBuilder);

updates = commitBuilder.getSortedListUpdates(0);
assertEquals(1, updates.getDeletesCount());
assertEquals(1, updates.getInsertsCount());
assertEquals(1, updates.getInserts(0).getEntriesCount());
assertEquals("afterFullClear1", updates.getInserts(0).getEntries(0).getValue().toStringUtf8());
assertEquals(IdTracker.NEW_RANGE_MIN_ID, updates.getInserts(0).getEntries(0).getId());
assertBuildable(commitBuilder);

// A subsequent add in the same bucket should receive NEW_RANGE_MIN_ID + 1.
resetUnderTest();
orderedListState = underTest.state(NAMESPACE, addr);
orderedListState.add(TimestampedValue.of("afterFullClear2", Instant.ofEpochMilli(2)));

commitBuilder = Windmill.WorkItemCommitRequest.newBuilder();
underTest.persist(commitBuilder);

updates = commitBuilder.getSortedListUpdates(0);
assertEquals(1, updates.getInsertsCount());
assertEquals(1, updates.getInserts(0).getEntriesCount());
assertEquals("afterFullClear2", updates.getInserts(0).getEntries(0).getValue().toStringUtf8());
assertEquals(IdTracker.NEW_RANGE_MIN_ID + 1, updates.getInserts(0).getEntries(0).getId());
assertBuildable(commitBuilder);
}

@Test
public void testOrderedListMergePendingAdds() {
SettableFuture<Map<Range<Instant>, RangeSet<Long>>> orderedListFuture = SettableFuture.create();
Expand Down
Loading