This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 866bca5d62 Flink: ConvertEqualityDeletes: Fix index consistency after
a failure or restore (#18674)
866bca5d62 is described below
commit 866bca5d6291d19b3480bb55529c9eebec82d0d2
Author: Maximilian Michels <[email protected]>
AuthorDate: Tue Nov 3 16:28:01 2026 +0100
Flink: ConvertEqualityDeletes: Fix index consistency after a failure or
restore (#18674)
---
docs/docs/flink-maintenance.md | 12 +-
.../operator/EqualityConvertCommitter.java | 56 +++-
.../maintenance/operator/EqualityConvertPlan.java | 19 +-
.../operator/EqualityConvertPlanner.java | 132 +++++---
.../api/TestConvertEqualityDeletes.java | 108 +++++++
.../api/TestConvertEqualityDeletesE2E.java | 111 ++++++-
.../operator/TestEqualityConvertCommitter.java | 198 +++++++++++-
.../operator/TestEqualityConvertDVWriter.java | 16 +-
.../operator/TestEqualityConvertPlanner.java | 338 ++++++++++++++++++++-
9 files changed, 907 insertions(+), 83 deletions(-)
diff --git a/docs/docs/flink-maintenance.md b/docs/docs/flink-maintenance.md
index 37be8445f7..eac8d84a08 100644
--- a/docs/docs/flink-maintenance.md
+++ b/docs/docs/flink-maintenance.md
@@ -113,10 +113,16 @@ Notes:
**State and restart**
-- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state and **maintains it incrementally across checkpoints**: each trigger
cycle only applies the commits added on the target branch since the last
indexed snapshot, and the resulting index is persisted with the next Flink
checkpoint. On failover, the worker resumes from the most recent checkpointed
index rather than rebuilding from scratch. Its size scales with the number of
live rows in the table for th [...]
-- Full (re)indexing from the target branch head is only triggered in a few
cases: cold start with no checkpoint, external commits that advance the target
past the currently-indexed snapshot (see below), when the planner detects that
the target branch has diverged from the indexed snapshot (rollback /
replace-main / expired marker), or when a staging snapshot is planned again
because the cycle that planned it did not commit.
+- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state, persisted with each Flink checkpoint. Its size scales with the
number of live rows in the table for the configured equality columns; size
Flink state backend, checkpoint storage, and TaskManager memory accordingly.
**RocksDB is the preferred state backend** because the PK index can grow beyond
the available heap and RocksDB spills to local disk instead of failing with an
`OutOfMemoryError`.
+- The index is **reused across triggers and restored on failover** while it
matches the head of the target branch: the head is the snapshot the index was
built from, or the converter's own commit for the last planned staging snapshot
directly on top of it. Each cycle applies its staging snapshot's changes to the
index, so consecutive conversions do not rebuild it.
+- Otherwise the next trigger **rebuilds the index** from the target branch
head, reading the equality columns of every live data file. This happens:
+ - on cold start without a checkpoint, and on the first trigger after
restoring state written by an older Iceberg version
+ - after any other commit on the target branch, e.g. compaction, external
writes, rollback, or replace-main (see *Concurrent commits on the target
branch* below)
+ - after restoring a checkpoint that predates the latest conversion commit
+ - when the last planned conversion did not commit
+ - when an earlier rebuild did not finish
- The `equalityFieldColumns` set is persisted in operator state. Reconfiguring
it across a restart from savepoint is **not supported** and the job will fail
fast on restore. Restart from a clean state (no savepoint) if the equality
columns change.
-- The committer is intentionally stateless. On restart, the planner
rediscovers its position by walking the target branch for the marker property
`equality-convert-staging-snapshot` written by the committer. If that marker is
no longer reachable on the target branch (target was rolled back,
replace-main'ed, or the marker snapshot was expired) the planner fails and
requires manual intervention.
+- On restart, the planner rediscovers its position by walking the target
branch for the marker property `equality-convert-staging-snapshot` written by
the committer. If that marker is no longer reachable on the target branch
(target was rolled back, replaced, or the marker snapshot was expired) the
planner fails and requires manual intervention.
**Concurrent commits on the target branch**
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
index 34d996c3d9..0e2fe43df4 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
@@ -22,8 +22,13 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -36,6 +41,7 @@ import
org.apache.iceberg.exceptions.CommitStateUnknownException;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.maintenance.api.Trigger;
import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.util.ContentFileUtil;
@@ -66,7 +72,13 @@ import org.slf4j.LoggerFactory;
* completes cleanly. Errors are reported via {@link
TaskResultAggregator#ERROR_STREAM} side output;
* the Aggregator collects them and surfaces failure on its own watermark.
*
- * <p>The committer is intentionally stateless: {@code bufferedResults} and
{@code planResult} are
+ * <p>The committer keeps the index generation of the last failed cycle as
checkpointed state. A
+ * failure in the readers or workers while the index is updated in a cycle
with nothing to convert
+ * may have left the index without some rows, but that cycle has no commit to
withhold. The
+ * committer withholds the next conversion commit on that generation instead:
the next trigger then
+ * plans that staging snapshot again, which makes the planner rebuild the
index on a new generation.
+ * The failure was reported where it happened, so the withheld cycle completes
without an error.
+ * Apart from that the committer is stateless: {@code bufferedResults} and
{@code planResult} are
* not checkpointed. The maintenance framework ensures mutual exclusive tasks,
and on restart the
* planner re-derives its position from {@link
#COMMITTED_STAGING_SNAPSHOT_PROPERTY} on main, so the
* committer never receives a plan for an already-committed staging snapshot.
@@ -93,6 +105,8 @@ public class EqualityConvertCommitter extends
AbstractStreamOperator<Trigger>
private transient Table table;
private transient List<DVWriteResult> bufferedResults;
private transient EqualityConvertPlan planResult;
+ private transient ListState<Long> failedIndexGenerationState;
+ private transient long failedIndexGeneration;
private transient Counter errorCounter;
private transient Counter addedDataFileNumCounter;
@@ -136,6 +150,32 @@ public class EqualityConvertCommitter extends
AbstractStreamOperator<Trigger>
this.commitDurationMsCounter =
taskMetricGroup.counter(COMMIT_DURATION_MS_METRIC);
}
+ @Override
+ public void initializeState(StateInitializationContext context) throws
Exception {
+ super.initializeState(context);
+ failedIndexGenerationState =
+ context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("failedIndexGeneration",
Types.LONG));
+
+ failedIndexGeneration = EqualityConvertPlan.NO_INDEX_GENERATION;
+ for (Long stateValue : failedIndexGenerationState.get()) {
+ Preconditions.checkState(
+ failedIndexGeneration == EqualityConvertPlan.NO_INDEX_GENERATION,
+ "failedIndexGeneration state should hold at most one value");
+ failedIndexGeneration = stateValue;
+ }
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ failedIndexGenerationState.clear();
+ if (failedIndexGeneration != EqualityConvertPlan.NO_INDEX_GENERATION) {
+ failedIndexGenerationState.add(failedIndexGeneration);
+ }
+ }
+
@Override
public void processElement1(StreamRecord<DVWriteResult> record) {
bufferedResults.add(record.getValue());
@@ -196,6 +236,7 @@ public class EqualityConvertCommitter extends
AbstractStreamOperator<Trigger>
"Skipping commit for table {} task {}: a DV writer reported an
error.",
tableName,
taskName);
+ failedIndexGeneration = planResult.indexGeneration();
deleteUncommittedDVs();
return;
}
@@ -209,6 +250,19 @@ public class EqualityConvertCommitter extends
AbstractStreamOperator<Trigger>
return;
}
+ if (planResult.indexGeneration() == failedIndexGeneration) {
+ LOG.warn(
+ "Not committing staging snapshot {} for table {} task {}: an index
update failed on "
+ + "index generation {}, so the next trigger rebuilds the index
and converts the "
+ + "snapshot again.",
+ planResult.stagingSnapshotId(),
+ tableName,
+ taskName,
+ failedIndexGeneration);
+ deleteUncommittedDVs();
+ return;
+ }
+
table.refresh();
List<DeleteFile> allDvFiles = Lists.newArrayList();
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
index 4aa326d19a..3226fbf4a2 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java
@@ -31,8 +31,8 @@ import
org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
* EqualityConvertCommitter} (for data files and staging deletes to commit).
*
* <p>A no-op cycle is encoded by {@link #stagingSnapshotId()} == {@link
- * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to
construct one and {@link
- * #noOp()} to check.
+ * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long, long)} to
construct one and
+ * {@link #noOp()} to check.
*
* <p>A cycle-start marker is a no-op plan whose done timestamp is never
reached. Use {@link
* #cycleStart(long)} to construct one.
@@ -50,6 +50,9 @@ import
org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
* @param triggerTimestamp original trigger timestamp, forwarded by the
Committer to the Aggregator
* @param doneTimestamp timestamp after which all phase watermarks have been
emitted; the DVWriter
* should only process the result when the watermark reaches or exceeds
this value
+ * @param indexGeneration generation of the worker index the cycle ran on. The
committer does not
+ * commit a cycle on a generation whose update failed. {@link
#NO_INDEX_GENERATION} for the
+ * cycle-start marker.
*/
@Internal
public record EqualityConvertPlan(
@@ -59,11 +62,14 @@ public record EqualityConvertPlan(
long stagingSnapshotId,
Long mainSnapshotId,
long triggerTimestamp,
- long doneTimestamp)
+ long doneTimestamp,
+ long indexGeneration)
implements Serializable {
public static final long NO_OP_STAGING_SNAPSHOT_ID = -1L;
+ public static final long NO_INDEX_GENERATION = -1L;
+
/** Returns {@code true} if this cycle has nothing to commit. */
public boolean noOp() {
return stagingSnapshotId == NO_OP_STAGING_SNAPSHOT_ID;
@@ -71,7 +77,7 @@ public record EqualityConvertPlan(
/** No-op cycle: empty file lists, sentinel staging snapshot id. */
public static EqualityConvertPlan noOp(
- Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) {
+ Long mainSnapshotId, long triggerTimestamp, long doneTimestamp, long
indexGeneration) {
return new EqualityConvertPlan(
ImmutableList.of(),
ImmutableList.of(),
@@ -79,7 +85,8 @@ public record EqualityConvertPlan(
NO_OP_STAGING_SNAPSHOT_ID,
mainSnapshotId,
triggerTimestamp,
- doneTimestamp);
+ doneTimestamp,
+ indexGeneration);
}
/**
@@ -87,6 +94,6 @@ public record EqualityConvertPlan(
* until the cycle's actual plan replaces it.
*/
public static EqualityConvertPlan cycleStart(long triggerTimestamp) {
- return noOp(null, triggerTimestamp, Long.MAX_VALUE);
+ return noOp(null, triggerTimestamp, Long.MAX_VALUE, NO_INDEX_GENERATION);
}
}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
index 96e5861f1c..1bf08b0976 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java
@@ -70,9 +70,8 @@ import org.slf4j.LoggerFactory;
* <ol>
* <li>{@link #refreshStagingCursor}: updates {@link #lastStagingSnapshotId}
from the most recent
* committer marker on the target branch.
- * <li>{@link #ensureIndexCurrent}: rebuilds the worker index from main when
there is no index
- * yet, when external commits (e.g. compaction) have advanced main, or
when the staging
- * snapshot picked for this cycle is the one the previous cycle planned.
+ * <li>{@link #ensureIndexCurrent}: rebuilds the worker index from main
unless it reflects the
+ * target branch head, see {@link #indexReflects}.
* <li>{@link #processStagingSnapshot}: resolve the chosen staging
snapshot's eq deletes against
* the (now-current) index, pass through any DV files, and index the
snapshot's new data files
* for the next cycle.
@@ -109,6 +108,8 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
private static final String SKIPPED_NO_OP_CYCLES_METRIC =
"skippedNoOpCycles";
private static final String REINDEX_COUNT_METRIC = "reindexCount";
+ private static final int STATE_VERSION = 1;
+
private final String tableName;
private final String taskName;
private final TableLoader tableLoader;
@@ -123,10 +124,12 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
private transient ListState<Long> indexSnapshotState;
// Generation of the worker's current index build.
private transient ListState<Long> indexGenerationState;
- // Staging snapshot the last emitted plan covered.
+ // Staging snapshot the index changed for until its conversion commit shows
up on the target.
private transient ListState<Long> pendingStagingSnapshotState;
// Equality field IDs the index was built with, allows to detect
reconfiguration.
private transient ListState<Integer> eqFieldIdsState;
+ // Absent in state of a version that recorded the index as current before it
was complete.
+ private transient ListState<Integer> stateVersionState;
private transient Table table;
@@ -134,8 +137,8 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
private transient Long lastStagingSnapshotId;
private transient Long indexSnapshotId;
private transient long indexGeneration;
- // Staging snapshot the last emitted plan covered, checkpointed so it
survives a restore taken
- // mid-cycle. Selecting it again means that cycle never committed.
+ // Staging snapshot the index changed for until its conversion commit shows
up on the target
+ // branch, checkpointed so it survives a restore taken mid-cycle.
private transient Long pendingStagingSnapshotId;
private transient long nextPhaseTs;
@@ -238,6 +241,20 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
"pendingStagingSnapshotId state should hold at most one value");
pendingStagingSnapshotId = stateValue;
}
+
+ stateVersionState =
+ context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("stateVersion",
Types.INT));
+
+ // The index recorded by an older version can be incomplete while it looks
current.
+ if (context.isRestored() && !stateVersionState.get().iterator().hasNext())
{
+ LOG.info(
+ "Restored state of an older version for table {} task {}, rebuilding
the worker index.",
+ tableName,
+ taskName);
+ indexSnapshotId = null;
+ }
}
@Override
@@ -260,6 +277,9 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
if (pendingStagingSnapshotId != null) {
pendingStagingSnapshotState.add(pendingStagingSnapshotId);
}
+
+ stateVersionState.clear();
+ stateVersionState.add(STATE_VERSION);
}
@Override
@@ -279,12 +299,12 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
Snapshot mainSnapshot = table.snapshot(targetBranch);
currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId()
: null;
- LastCommittedWork committedWork = refreshStagingCursor(mainSnapshot);
+ refreshStagingCursor(mainSnapshot);
Snapshot nextToProcess =
nextUnprocessedStagingSnapshot(table.snapshot(stagingBranch),
mainSnapshot);
- ensureIndexCurrent(mainSnapshot, committedWork, nextToProcess);
+ ensureIndexCurrent(mainSnapshot);
if (nextToProcess == null) {
LOG.info("Nothing new to convert on staging branch '{}'.",
stagingBranch);
@@ -303,45 +323,35 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
/**
* Updates {@link #lastStagingSnapshotId} from the most recent committer
marker on the target
- * branch. Returns the discovered work, or null when the target has not
moved since the last
- * trigger and the cursor therefore cannot have changed.
+ * branch. Does nothing when the target has not moved since the last trigger
and the cursor
+ * therefore cannot have changed.
*/
- private LastCommittedWork refreshStagingCursor(Snapshot mainSnapshot) {
+ private void refreshStagingCursor(Snapshot mainSnapshot) {
Long currentMainSnapshotId = mainSnapshot != null ?
mainSnapshot.snapshotId() : null;
if (Objects.equals(lastMainSnapshotId, currentMainSnapshotId)) {
- return null;
+ return;
}
- LastCommittedWork info = discoverLastCommittedWork(mainSnapshot);
- updateLastStagingSnapshotId(info);
- return info;
+ updateLastStagingSnapshotId(discoverLastCommittedWork(mainSnapshot));
}
/**
- * Rebuilds the worker index when it is missing, when external commits have
advanced the target
- * branch, or when {@code nextToProcess} is the staging snapshot the
previous plan covered.
- * Resolving an eq delete consumes the index entries it matches, so a cycle
that failed after its
- * delete phase left the index without them; the cursor only advances once
the committer's marker
- * is on the target branch, so planning the same staging snapshot again
means that cycle did not
- * commit and nothing else has rebuilt the index.
+ * Rebuilds the worker index unless it reflects {@code mainSnapshot}, see
{@link #indexReflects}.
*/
- private void ensureIndexCurrent(
- Snapshot mainSnapshot, LastCommittedWork committedWork, Snapshot
nextToProcess) {
+ private void ensureIndexCurrent(Snapshot mainSnapshot) {
if (mainSnapshot == null) {
+ // The target branch has no rows, so the index must not have any either.
A new generation
+ // makes the worker drop the rows indexed so far, e.g. by a cycle that
did not commit.
+ indexGeneration++;
+ indexSnapshotId = null;
+ pendingStagingSnapshotId = null;
lastMainSnapshotId = null;
return;
}
- boolean bootstrap = indexSnapshotId == null;
- boolean reindex =
- !bootstrap && committedWork != null &&
committedWork.externalCommitCount() > 0;
- boolean replan =
- !bootstrap
- && nextToProcess != null
- && Objects.equals(pendingStagingSnapshotId,
nextToProcess.snapshotId());
-
- if (bootstrap || reindex || replan) {
+ if (!indexReflects(mainSnapshot)) {
+ boolean bootstrap = indexSnapshotId == null;
LOG.info(
"{} worker index from main snapshot {} for field IDs {}.",
bootstrap ? "Bootstrapping" : "Reindexing",
@@ -350,20 +360,58 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
rebuildIndex(mainSnapshot, !bootstrap);
}
+ indexSnapshotId = mainSnapshot.snapshotId();
+ pendingStagingSnapshotId = null;
lastMainSnapshotId = mainSnapshot.snapshotId();
}
+ /**
+ * Returns whether the index reflects {@code mainSnapshot}.
+ *
+ * <p>Planning a staging snapshot changes the index before its conversion
commit exists: its eq
+ * deletes consume the entries they match and its data files are added. So
with a snapshot
+ * planned, the index reflects the head only if that snapshot's conversion
commit sits directly on
+ * top of the indexed snapshot. Without one, the head must be the indexed
snapshot. Any other head
+ * holds commits the index never saw (external commits, rollbacks, a restore
from before a
+ * conversion commit) or follows a cycle that did not commit.
+ *
+ * <p>After a restore, the conversion commit can come from a retry of the
planned cycle. A cycle
+ * that only failed to commit leaves the same index as its retry. A rebuild
that fails part way
+ * clears {@link #indexSnapshotId}, and the committer does not commit a
conversion on a generation
+ * whose update failed downstream.
+ */
+ private boolean indexReflects(Snapshot mainSnapshot) {
+ if (indexSnapshotId == null) {
+ return false;
+ }
+
+ if (pendingStagingSnapshotId == null) {
+ return mainSnapshot.snapshotId() == indexSnapshotId;
+ }
+
+ String committedStaging =
+
mainSnapshot.summary().get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
+ return committedStaging != null
+ && Long.parseLong(committedStaging) == pendingStagingSnapshotId
+ && Objects.equals(mainSnapshot.parentId(), indexSnapshotId);
+ }
+
/**
* Re-emits every data row on {@code mainSnapshot} so the worker's index
holds all their positions
* again, optionally preceded by a CLEAR_INDEX broadcast that evicts keyed
entries the re-emission
* will not re-add (e.g. a PK whose data file was removed by a CoW commit).
A bootstrap has no
- * earlier index and so nothing to evict.
+ * complete earlier index. Entries of an incomplete one are ignored at the
new generation.
*
* <p>The worker detects stale state by comparing the generation stamped on
the commands it
* receives with the one it stored, so every rebuild hands out a higher
generation, including a
* rebuild while the target branch stands still.
+ *
+ * <p>Clears {@link #indexSnapshotId} before emitting anything, and the
caller records the head
+ * only after this returns. Neither the next trigger nor a restore from a
checkpoint taken after a
+ * rebuild that failed part way trusts the index it left.
*/
private void rebuildIndex(Snapshot mainSnapshot, boolean evictStaleKeys) {
+ indexSnapshotId = null;
long generation = indexGeneration + 1;
if (evictStaleKeys) {
@@ -374,7 +422,6 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
reindexCounter.inc();
}
- indexSnapshotId = mainSnapshot.snapshotId();
indexGeneration = generation;
emitMainDataReadCommands(mainSnapshot);
}
@@ -399,7 +446,7 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
/**
* Walks main back from head looking for the most recent snapshot tagged
with {@link
* EqualityConvertCommitter#COMMITTED_STAGING_SNAPSHOT_PROPERTY}. Returns
the staging snapshot id
- * recorded there (or {@code null} if not reached) and the count of
intervening external commits.
+ * recorded there, or {@code null} if not reached.
*
* <p>The walk stops at whichever comes first:
*
@@ -410,7 +457,6 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
*/
private LastCommittedWork discoverLastCommittedWork(Snapshot mainSnapshot) {
Long lastCommittedStaging = null;
- int externalCount = 0;
boolean reachedLastInspected = false;
Snapshot current = mainSnapshot;
while (current != null) {
@@ -426,15 +472,13 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
break;
}
- externalCount++;
current = parentOf(current);
}
- return new LastCommittedWork(lastCommittedStaging, externalCount,
reachedLastInspected);
+ return new LastCommittedWork(lastCommittedStaging, reachedLastInspected);
}
- private record LastCommittedWork(
- Long lastCommittedStaging, int externalCommitCount, boolean
reachedLastInspected) {}
+ private record LastCommittedWork(Long lastCommittedStaging, boolean
reachedLastInspected) {}
/**
* Walks staging history from head back to the stop point and returns the
oldest unprocessed
@@ -533,7 +577,8 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
stagingSnapshot.snapshotId(),
currentMainSnapshotId,
triggerTs,
- nextPhaseTs)));
+ nextPhaseTs,
+ indexGeneration)));
advancePhase();
}
@@ -680,7 +725,8 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
output.collect(
METADATA_STREAM,
new StreamRecord<>(
- EqualityConvertPlan.noOp(currentMainSnapshotId, triggerTimestamp,
nextPhaseTs)));
+ EqualityConvertPlan.noOp(
+ currentMainSnapshotId, triggerTimestamp, nextPhaseTs,
indexGeneration)));
advancePhase();
}
@@ -700,7 +746,7 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
output.collect(
new StreamRecord<>(
ReadCommand.dataFile(
- task, indexSnapshotId, indexGeneration,
dataSequenceNumber(task.file())),
+ task, commitSnapshotId, indexGeneration,
dataSequenceNumber(task.file())),
nextPhaseTs));
}
} catch (IOException e) {
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
index 2e03a0655e..486ab0fe42 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
@@ -1473,6 +1473,114 @@ class TestConvertEqualityDeletes extends
MaintenanceTaskTestBase {
}
}
+ @Test
+ void testIndexUpdateFailureOnIdleTriggerIsRepaired() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile firstDelete = writeEqualityDelete(table, 1, "a");
+
table.newRowDelta().addDeletes(firstDelete).toBranch(STAGING_BRANCH).commit();
+
+ appendConvertTask(STAGING_BRANCH);
+
+ JobClient jobClient = null;
+ try {
+ jobClient = infra.env().executeAsync();
+
+ long time1 = System.currentTimeMillis();
+ infra.source().sendRecord(Trigger.create(time1, 0), time1);
+ assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+ // An external commit lands while staging is idle, and its data file is
unreadable while the
+ // next trigger updates the index.
+ DataFile externalData = writeDataFile(table, createRecord(2, "b"));
+ table.newAppend().appendFile(externalData).commit();
+ File externalFile = new File(externalData.location().replace("file:",
""));
+ File hiddenFile = new File(externalFile.getPath() + ".hidden");
+ assertThat(externalFile.renameTo(hiddenFile)).isTrue();
+
+ long time2 = time1 + 1;
+ infra.source().sendRecord(Trigger.create(time2, 0), time2);
+
assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isFalse();
+
+ assertThat(hiddenFile.renameTo(externalFile)).isTrue();
+
+ DeleteFile secondDelete = writeEqualityDelete(table, 2, "b");
+
table.newRowDelta().addDeletes(secondDelete).toBranch(STAGING_BRANCH).commit();
+
+ // The index may miss the external row, so the next conversion is not
committed.
+ table.refresh();
+ long mainSnapshotId = table.currentSnapshot().snapshotId();
+ long time3 = time2 + 1;
+ infra.source().sendRecord(Trigger.create(time3, 0), time3);
+ assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+ table.refresh();
+
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+
+ // Planned again, the conversion rebuilds the index and deletes the
external row.
+ long time4 = time3 + 1;
+ infra.source().sendRecord(Trigger.create(time4, 0), time4);
+ assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+ assertRecords(table, ImmutableList.of());
+ } finally {
+ closeJobClient(jobClient);
+ }
+ }
+
+ @Test
+ void testFailedConversionOnEmptyTargetIsRetried() throws Exception {
+ Table table = createTableWithDelete(3);
+
+ // The target branch has no snapshot until the first conversion commit.
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ // An upsert of (1, "a") and an insert whose data file is unreadable
during the first cycle.
+ DataFile reinsert = writeDataFile(table, createRecord(1, "a"));
+ DataFile insert = writeDataFile(table, createRecord(2, "b"));
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table
+ .newRowDelta()
+ .addRows(reinsert)
+ .addRows(insert)
+ .addDeletes(eqDelete)
+ .toBranch(STAGING_BRANCH)
+ .commit();
+
+ File insertFile = new File(insert.location().replace("file:", ""));
+ File hiddenFile = new File(insertFile.getPath() + ".hidden");
+ assertThat(insertFile.renameTo(hiddenFile)).isTrue();
+
+ appendConvertTask(STAGING_BRANCH);
+
+ JobClient jobClient = null;
+ try {
+ jobClient = infra.env().executeAsync();
+
+ // The first cycle indexes the re-insert, fails on the unreadable file
and does not commit.
+ long time1 = System.currentTimeMillis();
+ infra.source().sendRecord(Trigger.create(time1, 0), time1);
+
assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isFalse();
+ table.refresh();
+ assertThat(table.snapshot(SnapshotRef.MAIN_BRANCH)).isNull();
+
+ assertThat(hiddenFile.renameTo(insertFile)).isTrue();
+
+ // The retry must not delete the re-insert indexed by the first cycle.
+ long time2 = time1 + 1;
+ infra.source().sendRecord(Trigger.create(time2, 0), time2);
+ assertThat(infra.sink().poll(Duration.ofSeconds(10)).success()).isTrue();
+
+ assertRecords(table, ImmutableList.of(createRecord(1, "a"),
createRecord(2, "b")));
+ } finally {
+ closeJobClient(jobClient);
+ }
+ }
+
private void appendConvertTask() {
appendConvertTask(STAGING_BRANCH);
}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
index e39d05b66f..d18f81a54a 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.flink.maintenance.api;
import static org.apache.iceberg.flink.SimpleDataUtil.createRecord;
import static
org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME;
+import static
org.apache.iceberg.flink.maintenance.operator.EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY;
import static
org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER;
import static org.assertj.core.api.Assertions.assertThat;
@@ -30,12 +31,18 @@ import java.nio.file.Path;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.test.junit5.InjectClusterClient;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileContent;
@@ -104,18 +111,7 @@ class TestConvertEqualityDeletesE2E extends
OperatorTestBase {
table.refresh();
assertThat(dvCountOnMain(table)).isZero();
- TableMaintenance.forTable(env, tableLoader(), LOCK_FACTORY)
- .uidSuffix("ConvertEqualityDeletesE2EUID-" + stagingBranch)
- .rateLimit(Duration.ofMillis(50))
- .lockCheckDelay(Duration.ofMillis(50))
- .add(
- ConvertEqualityDeletes.builder()
- .scheduleOnInterval(Duration.ofMillis(100))
- .stagingBranch(stagingBranch)
- .targetBranch(SnapshotRef.MAIN_BRANCH)
- .equalityFieldColumns(ImmutableList.of("id", "data"))
- .parallelism(2))
- .append();
+ appendConvertTask(env, stagingBranch, "ConvertEqualityDeletesE2EUID-" +
stagingBranch);
JobClient jobClient = null;
try {
@@ -211,6 +207,97 @@ class TestConvertEqualityDeletesE2E extends
OperatorTestBase {
}
}
+ /**
+ * Restoring from a savepoint taken before a committed cycle leaves the
converter's index without
+ * the rows that cycle copied to the target. The next conversion must
rebuild it, or a delete for
+ * one of those rows resolves to nothing.
+ */
+ @Test
+ void reindexesAfterRestoreFromBeforeCommittedCycle(
+ @TempDir File savepointDir, @InjectClusterClient ClusterClient<?>
clusterClient)
+ throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table
+ .newRowDelta()
+ .addDeletes(writeEqualityDelete(table, 1, "a"))
+ .toBranch(STAGING_BRANCH)
+ .commit();
+ table.refresh();
+
+ appendConvertTask(env, STAGING_BRANCH, "ConvertEqualityDeletesRestoreUID");
+ String savepointPath;
+ JobClient jobClient = null;
+ try {
+ jobClient = env.executeAsync();
+ awaitConverted(table, table.snapshot(STAGING_BRANCH).snapshotId());
+ savepointPath =
+ jobClient.triggerSavepoint(savepointDir.getPath(),
SavepointFormatType.CANONICAL).get();
+
+ // This cycle copies S2's row to the target after the savepoint.
+ table.newAppend().appendFile(writeDataFile(table, 2,
"b")).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+ awaitConverted(table, table.snapshot(STAGING_BRANCH).snapshotId());
+ } finally {
+ closeJobClient(jobClient);
+ }
+
+ table
+ .newRowDelta()
+ .addDeletes(writeEqualityDelete(table, 2, "b"))
+ .toBranch(STAGING_BRANCH)
+ .commit();
+ table.refresh();
+ long deleteSnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+ StreamExecutionEnvironment restoredEnv =
StreamExecutionEnvironment.getExecutionEnvironment();
+ appendConvertTask(restoredEnv, STAGING_BRANCH,
"ConvertEqualityDeletesRestoreUID");
+ StreamGraph streamGraph = restoredEnv.getStreamGraph();
+ // Submitting the graph directly skips the executor, which is what
normally assigns the job id
+ streamGraph.setJobId(JobID.generate());
+
streamGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savepointPath));
+ JobID restoredJobId = clusterClient.submitJob(streamGraph).get();
+ try {
+ awaitConverted(table, deleteSnapshotId);
+ SimpleDataUtil.assertTableRecords(table, ImmutableList.of());
+ } finally {
+ clusterClient.cancel(restoredJobId).get();
+ Awaitility.await()
+ .until(() ->
clusterClient.getJobStatus(restoredJobId).get().isTerminalState());
+ }
+ }
+
+ private void appendConvertTask(
+ StreamExecutionEnvironment environment, String stagingBranch, String
uidSuffix)
+ throws IOException {
+ TableMaintenance.forTable(environment, tableLoader(), LOCK_FACTORY)
+ .uidSuffix(uidSuffix)
+ .rateLimit(Duration.ofMillis(50))
+ .lockCheckDelay(Duration.ofMillis(50))
+ .add(
+ ConvertEqualityDeletes.builder()
+ .scheduleOnInterval(Duration.ofMillis(100))
+ .stagingBranch(stagingBranch)
+ .targetBranch(SnapshotRef.MAIN_BRANCH)
+ .equalityFieldColumns(ImmutableList.of("id", "data"))
+ .parallelism(2))
+ .append();
+ }
+
+ private static void awaitConverted(Table table, long stagingSnapshotId) {
+ Awaitility.await()
+ .atMost(Duration.ofMinutes(5))
+ .pollInterval(Duration.ofMillis(200))
+ .untilAsserted(
+ () -> {
+ table.refresh();
+ assertThat(table.currentSnapshot().summary())
+ .containsEntry(
+ COMMITTED_STAGING_SNAPSHOT_PROPERTY,
String.valueOf(stagingSnapshotId));
+ });
+ }
+
private static void awaitSucceeded(List<String> counterKey, long expected) {
Awaitility.await()
.atMost(Duration.ofMinutes(3))
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
index 35aa9908bc..e52f53f942 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
@@ -25,6 +25,7 @@ import java.io.IOException;
import java.nio.file.Path;
import java.util.List;
import java.util.stream.Collectors;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
@@ -54,6 +55,8 @@ import org.junit.jupiter.api.io.TempDir;
class TestEqualityConvertCommitter extends OperatorTestBase {
+ private static final long INDEX_GENERATION = 1L;
+
@TempDir private Path tempDir;
@Test
@@ -78,7 +81,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
stagingSnapshotId,
snapshotIdBefore,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
harness.processBothWatermarks(new Watermark(doneTs));
@@ -116,7 +120,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
42L,
snapshotIdBefore,
doneTs - 2,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement2(new StreamRecord<>(planResult, doneTs - 2));
@@ -165,7 +170,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
42L,
snapshotIdBefore,
triggerTs,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement2(new StreamRecord<>(planResult));
harness.processBothWatermarks(new Watermark(doneTs));
@@ -203,7 +209,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
harness.open();
long doneTs = System.currentTimeMillis();
- EqualityConvertPlan planResult = EqualityConvertPlan.noOp(null, doneTs -
1, doneTs);
+ EqualityConvertPlan planResult =
+ EqualityConvertPlan.noOp(null, doneTs - 1, doneTs, INDEX_GENERATION);
harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
harness.processBothWatermarks(new Watermark(doneTs));
@@ -235,7 +242,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
42L,
snapshotIdBefore,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
// Receive an abort signal from the DVWriter followed by the planning
planResult.
harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, doneTs));
@@ -250,6 +258,129 @@ class TestEqualityConvertCommitter extends
OperatorTestBase {
}
}
+ @Test
+ void withholdsNextConversionCommitAfterFailedIdleCycle() throws Exception {
+ Table table = createTable(3, FileFormat.PARQUET);
+ insert(table, 1, "a");
+
+ DataFile stagingDataFile = getFirstDataFile(table);
+ long mainSnapshotId = table.currentSnapshot().snapshotId();
+ DeleteFile writtenDv = writeDV(table, stagingDataFile.location(), 0L);
+
+ try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan,
Trigger> harness =
+ createHarness()) {
+ harness.open();
+
+ failIdleRebuild(harness, mainSnapshotId);
+
+ // The next conversion is not committed, and the DV it wrote is deleted.
+ harness.processElement1(
+ new StreamRecord<>(
+ new DVWriteResult(Lists.newArrayList(writtenDv),
Lists.newArrayList()), 20L));
+ harness.processElement2(
+ new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId,
20L, 1L), 19L));
+ harness.processBothWatermarks(new Watermark(20L));
+
+ table.refresh();
+
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+
assertThat(table.io().newInputFile(writtenDv.location()).exists()).isFalse();
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+
+ // Planned again with a rebuilt index on a new generation, the
conversion commits.
+ harness.processElement2(
+ new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId,
30L, 2L), 29L));
+ harness.processBothWatermarks(new Watermark(30L));
+
+ assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+ }
+ }
+
+ @Test
+ void withholdsNextConversionCommitAfterRestore() throws Exception {
+ Table table = createTable(3, FileFormat.PARQUET);
+ insert(table, 1, "a");
+
+ DataFile stagingDataFile = getFirstDataFile(table);
+ long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+ OperatorSubtaskState state;
+ try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan,
Trigger> harness =
+ createHarness()) {
+ harness.open();
+
+ failIdleRebuild(harness, mainSnapshotId);
+
+ state = harness.snapshot(1, 11L);
+ }
+
+ try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan,
Trigger> harness =
+ createHarness()) {
+ harness.initializeState(state);
+ harness.open();
+
+ harness.processElement2(
+ new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId,
20L, 1L), 19L));
+ harness.processBothWatermarks(new Watermark(20L));
+
+ table.refresh();
+
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId);
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ }
+ }
+
+ @Test
+ void commitsReplannedConversionAfterFailedConversion() throws Exception {
+ Table table = createTable(3, FileFormat.PARQUET);
+ insert(table, 1, "a");
+
+ DataFile stagingDataFile = getFirstDataFile(table);
+ long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+ try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan,
Trigger> harness =
+ createHarness()) {
+ harness.open();
+
+ // A failed conversion is planned again with a rebuilt index, so nothing
is withheld.
+ harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, 10L));
+ harness.processElement2(
+ new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId,
10L, 1L), 9L));
+ harness.processBothWatermarks(new Watermark(10L));
+
+ harness.processElement2(
+ new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId,
20L, 2L), 19L));
+ harness.processBothWatermarks(new Watermark(20L));
+
+ assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+ }
+ }
+
+ @Test
+ void commitsNextConversionAfterCleanRebuild() throws Exception {
+ Table table = createTable(3, FileFormat.PARQUET);
+ insert(table, 1, "a");
+
+ DataFile stagingDataFile = getFirstDataFile(table);
+ long mainSnapshotId = table.currentSnapshot().snapshotId();
+
+ try (TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan,
Trigger> harness =
+ createHarness()) {
+ harness.open();
+
+ failIdleRebuild(harness, mainSnapshotId);
+
+ // An external commit makes the next idle cycle rebuild the index, which
repairs the failure.
+ harness.processElement2(
+ new StreamRecord<>(EqualityConvertPlan.noOp(mainSnapshotId, 19L,
20L, 2L), 19L));
+ harness.processBothWatermarks(new Watermark(20L));
+
+ harness.processElement2(
+ new StreamRecord<>(conversionPlan(stagingDataFile, mainSnapshotId,
30L, 2L), 29L));
+ harness.processBothWatermarks(new Watermark(30L));
+
+ assertThat(committedStagingSnapshot(table)).isEqualTo("42");
+ }
+ }
+
@Test
void failsCommitWhenExternalCommitLandsAfterPlan() throws Exception {
Table table = createTable(3, FileFormat.PARQUET);
@@ -272,7 +403,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
stagingSnapshotId,
mainSnapshotIdAtPlan,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
@@ -330,7 +462,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
stagingSnapshotId,
mainSnapshotIdAtPlan,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
// Cycle 1 commits, advancing the target past mainSnapshotIdAtPlan and
writing the marker.
harness.processElement2(new StreamRecord<>(planResult, doneTs - 1));
@@ -353,7 +486,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
stagingSnapshotId,
mainSnapshotIdAtPlan,
doneTs2 - 1,
- doneTs2);
+ doneTs2,
+ INDEX_GENERATION);
harness.processElement2(new StreamRecord<>(replay, doneTs2 - 1));
harness.processBothWatermarks(new Watermark(doneTs2));
@@ -399,7 +533,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
777L,
mainSnapshotIdAtPlan,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement1(
new StreamRecord<>(
@@ -450,7 +585,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
42L,
snapshotIdBefore,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
// One writer task wrote a DV (rewriting an existing one); a sibling
task aborted.
harness.processElement1(
@@ -505,7 +641,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
888L,
mainSnapshotIdAtPlan,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement1(
new StreamRecord<>(
@@ -557,7 +694,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
123L,
mainSnapshotId,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement1(
new StreamRecord<>(
@@ -616,7 +754,8 @@ class TestEqualityConvertCommitter extends OperatorTestBase
{
123L,
mainSnapshotId,
doneTs - 1,
- doneTs);
+ doneTs,
+ INDEX_GENERATION);
harness.processElement1(
new StreamRecord<>(
@@ -655,6 +794,39 @@ class TestEqualityConvertCommitter extends
OperatorTestBase {
SnapshotRef.MAIN_BRANCH));
}
+ // The planner rebuilt the index to generation 1 in a cycle with nothing to
convert, and a worker
+ // failed to update it.
+ private static void failIdleRebuild(
+ TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan,
Trigger> harness,
+ long mainSnapshotId)
+ throws Exception {
+ harness.processElement1(new StreamRecord<>(DVWriteResult.ABORT, 10L));
+ harness.processElement2(
+ new StreamRecord<>(EqualityConvertPlan.noOp(mainSnapshotId, 9L, 10L,
1L), 9L));
+ harness.processBothWatermarks(new Watermark(10L));
+ }
+
+ private static String committedStagingSnapshot(Table table) {
+ table.refresh();
+ return table
+ .currentSnapshot()
+ .summary()
+ .get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
+ }
+
+ private static EqualityConvertPlan conversionPlan(
+ DataFile stagingDataFile, long mainSnapshotId, long doneTs, long
indexGeneration) {
+ return new EqualityConvertPlan(
+ Lists.newArrayList(stagingDataFile),
+ Lists.newArrayList(),
+ Lists.newArrayList(),
+ 42L,
+ mainSnapshotId,
+ doneTs - 1,
+ doneTs,
+ indexGeneration);
+ }
+
private static List<Watermark>
watermarks(TwoInputStreamOperatorTestHarness<?, ?, ?> harness) {
return harness.getOutput().stream()
.filter(Watermark.class::isInstance)
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
index 93614fe8ab..b7dd1e308d 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java
@@ -40,6 +40,7 @@ import org.junit.jupiter.api.Test;
class TestEqualityConvertDVWriter extends OperatorTestBase {
private static final byte[] EMPTY_PARTITION = new byte[0];
+ private static final long INDEX_GENERATION = 1L;
@Test
void writesDVFileForSinglePosition() throws Exception {
@@ -247,7 +248,8 @@ class TestEqualityConvertDVWriter extends OperatorTestBase {
1L,
null,
0L,
- 0L);
+ 0L,
+ INDEX_GENERATION);
// New position 1 in the same file. Must merge it with the staged
position 0.
harness.processElement1(
@@ -288,7 +290,8 @@ class TestEqualityConvertDVWriter extends OperatorTestBase {
1L,
plannedSnapshotId,
0L,
- 0L);
+ 0L,
+ INDEX_GENERATION);
harness.processElement1(
new StreamRecord<>(new DVPosition(dataFilePath, 0, 0,
EMPTY_PARTITION, 0L), time));
@@ -491,7 +494,14 @@ class TestEqualityConvertDVWriter extends OperatorTestBase
{
private static EqualityConvertPlan emptyEqualityConvertPlan() {
return new EqualityConvertPlan(
- Lists.newArrayList(), Lists.newArrayList(), Lists.newArrayList(), 1L,
null, 0L, 0L);
+ Lists.newArrayList(),
+ Lists.newArrayList(),
+ Lists.newArrayList(),
+ 1L,
+ null,
+ 0L,
+ 0L,
+ INDEX_GENERATION);
}
private static String getFirstDataFilePath(Table table) {
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
index 1482ce8726..c98de623e7 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java
@@ -26,9 +26,13 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.StateObjectCollection;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.OperatorStreamStateHandle;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
@@ -45,6 +49,7 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RewriteFiles;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.FileHelpers;
import org.apache.iceberg.data.GenericAppenderHelper;
@@ -53,9 +58,12 @@ import org.apache.iceberg.data.Record;
import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.flink.maintenance.api.Trigger;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
class TestEqualityConvertPlanner extends OperatorTestBase {
@@ -353,6 +361,13 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
sendTrigger(harness);
assertThat(harness.extractOutputValues()).hasSize(firstTriggerCount);
assertThat(planner(harness).skippedNoOpCycles()).isEqualTo(1);
+
+ // Each trigger emits a cycle start marker followed by the plan. The
bootstrap built
+ // generation 1, and the second trigger reused it.
+ List<StreamRecord<EqualityConvertPlan>> metadata =
+
Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM));
+ assertThat(metadata.get(1).getValue().indexGeneration()).isEqualTo(1L);
+ assertThat(metadata.get(3).getValue().indexGeneration()).isEqualTo(1L);
}
}
@@ -668,6 +683,38 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
}
}
+ @Test
+ void startsNewIndexGenerationWhenReplanningOnEmptyTarget() throws Exception {
+ Table table = createTableWithDelete(3);
+
+ // The target branch has no snapshot until the first conversion commit.
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DataFile reinsert = writeDataFile(table, createRecord(1, "a"));
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+
table.newRowDelta().addRows(reinsert).addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ // The first cycle indexes the re-insert but does not commit, so the
same staging snapshot is
+ // planned again. Its eq delete must not resolve against the re-insert
of the first cycle.
+ sendTrigger(harness);
+ sendTrigger(harness);
+
+ List<ReadCommand> eqDeletes =
+ harness.extractOutputValues().stream()
+ .filter(cmd -> cmd.task() instanceof EqualityDeleteFileScanTask)
+ .collect(Collectors.toList());
+ assertThat(eqDeletes).hasSize(2);
+ assertThat(eqDeletes.get(1).indexGeneration())
+ .isGreaterThan(eqDeletes.get(0).indexGeneration());
+ }
+ }
+
@Test
void rebuildsIndexAfterRestoreOfUncommittedCycle() throws Exception {
Table table = createTableWithDelete(3);
@@ -687,8 +734,7 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
harness.open();
// First cycle converts S1 and commits, so the newest commit on the
target is the converter's
- // own marker. The existing reindex path counts only commits newer than
that marker, so it
- // cannot fire on the restore below and the replan check is the only
thing that can rebuild.
+ // own marker, which the planner sees on the next trigger.
sendTrigger(harness);
assertThat(harness.extractOutputValues()).hasSize(3);
simulateConvertCommit(table,
table.snapshot(STAGING_BRANCH).snapshotId());
@@ -722,6 +768,206 @@ class TestEqualityConvertPlanner extends OperatorTestBase
{
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void rebuildsIndexAfterRestoreFromBeforeCommittedCycle(boolean
plannerSawConversionCommit)
+ throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DataFile stagingData;
+ OperatorSubtaskState state;
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ // The checkpoint follows the first cycle, which converts S1 and
commits. The planner sees
+ // that commit only if another trigger follows.
+ sendTrigger(harness);
+ simulateConvertCommit(table,
table.snapshot(STAGING_BRANCH).snapshotId());
+ if (plannerSawConversionCommit) {
+ sendTrigger(harness);
+ }
+
+ state = harness.snapshot(1, System.currentTimeMillis());
+
+ // The second cycle indexes S2's data file as staging data and copies it
to the target. No
+ // checkpoint follows it.
+ stagingData = writeDataFile(table, createRecord(2, "b"));
+
table.newAppend().appendFile(stagingData).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+ long s2SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+ sendTrigger(harness);
+ table
+ .newRowDelta()
+ .addRows(stagingData)
+ .set(
+ EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY,
+ String.valueOf(s2SnapshotId))
+ .commit();
+ table.refresh();
+ }
+
+ DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b");
+
table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.initializeState(state);
+ harness.open();
+
+ // The index restored with the planner never held S2's row, so the
planner rebuilds it from
+ // the target before resolving the delete for that row.
+ sendTrigger(harness);
+ List<ReadCommand> commands = harness.extractOutputValues();
+
+ assertThat(commands).anyMatch(cmd ->
filePath(cmd).equals(stagingData.location()));
+ assertThat(countEqDeleteTasks(commands)).isEqualTo(1);
+ assertThat(planner(harness).reindexCount()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void rebuildsIndexAfterRestoreWhenExternalCommitIsBelowConversionCommit()
throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+ long s1SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+ OperatorSubtaskState state;
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ // The first cycle plans S1 but does not commit, then a checkpoint is
taken.
+ sendTrigger(harness);
+ state = harness.snapshot(1, System.currentTimeMillis());
+
+ // An external commit lands, the planner indexes it and the second cycle
commits S1 on top.
+ insert(table, 2, "b");
+ sendTrigger(harness);
+ assertThat(planner(harness).reindexCount()).isEqualTo(1);
+ simulateConvertCommit(table, s1SnapshotId);
+ }
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.initializeState(state);
+ harness.open();
+
+ // The newest conversion commit is for the planned snapshot, but the
external commit below it
+ // is not in the restored index.
+ sendTrigger(harness);
+
+ // Two inserts plus simulateConvertCommit's marker file = 3 data files
on the target.
+
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(3);
+ assertThat(planner(harness).reindexCount()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void rebuildsIndexAfterRestoreFromFailedRebuild() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+ long s1SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId();
+
+ List<File> manifests = dataManifestFiles(table);
+
+ OperatorSubtaskState state;
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ // The first cycle plans S1 but its commit fails.
+ sendTrigger(harness);
+
+ // The retry's rebuild fails part way, then a checkpoint is taken.
+ hide(manifests);
+ sendTrigger(harness);
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ restore(manifests);
+ state = harness.snapshot(1, System.currentTimeMillis());
+
+ // The next retry rebuilds the index and commits S1 on the same parent.
+ sendTrigger(harness);
+ simulateConvertCommit(table, s1SnapshotId);
+ }
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.initializeState(state);
+ harness.open();
+
+ // The commit has the planned snapshot and parent, but the restored
index is the one the
+ // failed rebuild left.
+ sendTrigger(harness);
+
+ // One insert plus simulateConvertCommit's marker file = 2 data files on
the target.
+
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2);
+ }
+ }
+
+ @Test
+ void bootstrapsAgainAfterRestoreFromFailedBootstrap() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ // The newest commit is a conversion commit, so no external commit makes
the planner rebuild.
+ simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).snapshotId());
+
+ List<File> manifests = dataManifestFiles(table);
+
+ OperatorSubtaskState state;
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ hide(manifests);
+ sendTrigger(harness);
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ state = harness.snapshot(1, System.currentTimeMillis());
+ }
+
+ restore(manifests);
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.initializeState(state);
+ harness.open();
+
+ // The checkpoint must not claim the index covers the head: the
bootstrap did not finish.
+ sendTrigger(harness);
+
+ // One insert plus simulateConvertCommit's marker file = 2 data files on
the target.
+
assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2);
+ }
+ }
+
@Test
void noMainReEmitAfterOwnCommit() throws Exception {
Table table = createTableWithDelete(3);
@@ -963,6 +1209,55 @@ class TestEqualityConvertPlanner extends OperatorTestBase
{
}
}
+ @Test
+ void rebuildsIndexAfterRestoreFromOlderVersion() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ OperatorSubtaskState state;
+ long generationBeforeRestore;
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.open();
+
+ // The same state as in stateRestoredAfterCheckpoint, where the index is
reused.
+ sendTrigger(harness);
+ simulateConvertCommit(table,
table.snapshot(STAGING_BRANCH).snapshotId());
+ sendTrigger(harness);
+ generationBeforeRestore =
+ harness.extractOutputValues().stream()
+ .mapToLong(ReadCommand::indexGeneration)
+ .max()
+ .getAsLong();
+ state = harness.snapshot(1, System.currentTimeMillis());
+ }
+
+ try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness =
+ createHarness(STAGING_BRANCH)) {
+ harness.initializeState(stateOfOlderVersion(state));
+ harness.open();
+
+ // An older version can record the head as indexed while the index is
incomplete.
+ sendTrigger(harness);
+ List<ReadCommand> commands = harness.extractOutputValues();
+
+ long dataFileCount =
+ Long.parseLong(
+
table.currentSnapshot().summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP));
+ assertThat(countDataFileTasks(commands)).isEqualTo(dataFileCount);
+
+ // The workers ignore index entries of an older generation.
+ assertThat(commands).allMatch(cmd -> cmd.indexGeneration() >
generationBeforeRestore);
+ }
+ }
+
@Test
void failsOnV2PosDeleteOnStagingBranch() throws Exception {
Table table = createTableWithDelete(2);
@@ -1175,10 +1470,49 @@ class TestEqualityConvertPlanner extends
OperatorTestBase {
3);
}
+ private static List<File> dataManifestFiles(Table table) {
+ return table.currentSnapshot().dataManifests(table.io()).stream()
+ .map(manifest -> new File(manifest.path().replace("file:", "")))
+ .collect(Collectors.toList());
+ }
+
+ // Makes reads of the files fail until restore is called.
+ private static void hide(List<File> files) {
+ assertThat(files.stream().map(file ->
file.renameTo(hidden(file))).collect(Collectors.toList()))
+ .containsOnly(true);
+ }
+
+ private static void restore(List<File> files) {
+ assertThat(files.stream().map(file ->
hidden(file).renameTo(file)).collect(Collectors.toList()))
+ .containsOnly(true);
+ }
+
+ private static File hidden(File file) {
+ return new File(file.getPath() + ".hidden");
+ }
+
private static long countDataFileTasks(List<ReadCommand> commands) {
return commands.stream().filter(c -> c.task() instanceof
FileScanTask).count();
}
+ // Keeps the operator state entries that the version before the stateVersion
entry wrote.
+ private static OperatorSubtaskState stateOfOlderVersion(OperatorSubtaskState
state) {
+ Set<String> olderStateNames =
+ Sets.newHashSet(
+ "indexSnapshotId", "indexGeneration", "eqFieldIds",
"pendingStagingSnapshotId");
+ List<OperatorStateHandle> handles = Lists.newArrayList();
+ for (OperatorStateHandle handle : state.getManagedOperatorState()) {
+ Map<String, OperatorStateHandle.StateMetaInfo> stateMetaInfos =
+ Maps.newHashMap(handle.getStateNameToPartitionOffsets());
+ stateMetaInfos.keySet().retainAll(olderStateNames);
+ handles.add(new OperatorStreamStateHandle(stateMetaInfos,
handle.getDelegateStateHandle()));
+ }
+
+ return OperatorSubtaskState.builder()
+ .setManagedOperatorState(new StateObjectCollection<>(handles))
+ .build();
+ }
+
private static long countEqDeleteTasks(List<ReadCommand> commands) {
return
commands.stream().filter(TestEqualityConvertPlanner::isEqDelete).count();
}
