swapna267 commented on code in PR #18717:
URL: https://github.com/apache/iceberg/pull/18717#discussion_r4519274866
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java:
##########
@@ -256,6 +269,26 @@ private void applyBufferedRows() throws Exception {
bufferedRows.clear();
}
+ private void removeFromIndex(DVPosition removed) throws Exception {
+ // Only dataRowPositions is scanned: the maintenance lock serializes
cycles, so the staging rows
+ // a previous cycle buffered have joined the index before this cycle's
commands arrive.
+ List<DVPosition> kept = Lists.newArrayList();
+ boolean found = false;
+ for (DVPosition position : dataRowPositions.get()) {
Review Comment:
As this involves retrieving the entire state for that key , can we do
something like write a removal marker here ? Can we apply it and write back the
updated state when resolving deletes ?
A side effect i can think of is slightly large state size if it's not being
cleaned up, if no new deletes are coming. We can handle that with stale timer
registration ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
