Stephen0421 commented on code in PR #10813:
URL: https://github.com/apache/paimon/pull/10813#discussion_r4386106214
##########
paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java:
##########
@@ -137,68 +180,103 @@ public void refreshFiles(
.computeIfAbsent(bucket, k -> new BucketLookupState());
state.lock.writeLock().lock();
try {
+ // Load before touching the levels so a failure leaves the
previous view intact.
+ // Compaction can mark a row deleted without rewriting its data
file. Streaming
+ // plans leave deletion files empty (SnapshotReaderImpl fills them
for batch plans
+ // only), and the updated file may not appear in this split at
all, so the bucket
+ // index is reloaded from the latest snapshot. That snapshot can
be newer than the
+ // splits just applied; a newer vector hides a row only until the
next refresh
+ // installs its data file. Unchanged index files are not read
again.
+ Map<String, DeletionVector> deletionVectors =
+ loadDeletionVectors(partition, bucket, state);
if (state.lookupLevels == null) {
// Initial phase: ignore beforeFiles as they represent
deletions from previous state
state.lookupLevels = createLookupLevels(partition, bucket,
dataFiles);
} else {
state.lookupLevels.getLevels().update(beforeFiles, dataFiles);
}
+ state.deletionVectors = deletionVectors;
} finally {
state.lock.writeLock().unlock();
}
}
- private LookupLevels<KeyValue> createLookupLevels(
+ private Map<String, DeletionVector> loadDeletionVectors(
+ BinaryRow partition, int bucket, BucketLookupState state) {
+ if (!deletionVectorsEnabled) {
+ return Collections.emptyMap();
+ }
+ List<IndexFileMeta> fileMetas = new ArrayList<>();
+ for (IndexManifestEntry entry : currentDeletionVectorEntries()) {
+ if (entry.bucket() == bucket &&
entry.partition().equals(partition)) {
+ fileMetas.add(entry.indexFile());
+ }
+ }
+ if (fileMetas.equals(state.deletionVectorFiles)) {
+ return state.deletionVectors;
+ }
+ if (fileMetas.isEmpty()) {
+ state.deletionVectorFiles = fileMetas;
+ return Collections.emptyMap();
+ }
+ // Record the file list only after the read succeeds. A failure leaves
the previous
+ // list and vectors in place, so the next refresh retries instead of
trusting them.
+ Map<String, DeletionVector> loaded =
+ indexFileHandler.readAllDeletionVectors(partition, bucket,
fileMetas);
+ state.deletionVectorFiles = fileMetas;
+ return loaded;
+ }
+
+ private List<IndexManifestEntry> currentDeletionVectorEntries() {
+ Snapshot latest = snapshotManager.latestSnapshot();
+ long snapshotId = latest == null ? -1L : latest.id();
Review Comment:
Fixed. `refreshFiles` now takes the snapshot id of the file event, and
deletion vectors are loaded from that snapshot instead of `latestSnapshot()`.
`QueryExecutorOperator` reads the snapshot id from field 0 of the
`FileMonitorTable` row and passes it through. A lagging executor applying
snapshot N therefore still sees N's deletion vectors, even if N+1 has already
rewritten the file and dropped the vector. The deleted row is not returned.
If that snapshot has expired, refresh throws `OutOfRangeException` instead
of serving the files with no deletion vectors. `FileStoreLookupFunction`
already catches this and reopens from a live snapshot.
##########
paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java:
##########
@@ -137,68 +180,103 @@ public void refreshFiles(
.computeIfAbsent(bucket, k -> new BucketLookupState());
state.lock.writeLock().lock();
try {
+ // Load before touching the levels so a failure leaves the
previous view intact.
+ // Compaction can mark a row deleted without rewriting its data
file. Streaming
+ // plans leave deletion files empty (SnapshotReaderImpl fills them
for batch plans
+ // only), and the updated file may not appear in this split at
all, so the bucket
+ // index is reloaded from the latest snapshot. That snapshot can
be newer than the
+ // splits just applied; a newer vector hides a row only until the
next refresh
+ // installs its data file. Unchanged index files are not read
again.
+ Map<String, DeletionVector> deletionVectors =
+ loadDeletionVectors(partition, bucket, state);
if (state.lookupLevels == null) {
// Initial phase: ignore beforeFiles as they represent
deletions from previous state
state.lookupLevels = createLookupLevels(partition, bucket,
dataFiles);
} else {
state.lookupLevels.getLevels().update(beforeFiles, dataFiles);
}
+ state.deletionVectors = deletionVectors;
} finally {
state.lock.writeLock().unlock();
}
}
- private LookupLevels<KeyValue> createLookupLevels(
+ private Map<String, DeletionVector> loadDeletionVectors(
+ BinaryRow partition, int bucket, BucketLookupState state) {
+ if (!deletionVectorsEnabled) {
+ return Collections.emptyMap();
+ }
+ List<IndexFileMeta> fileMetas = new ArrayList<>();
+ for (IndexManifestEntry entry : currentDeletionVectorEntries()) {
+ if (entry.bucket() == bucket &&
entry.partition().equals(partition)) {
+ fileMetas.add(entry.indexFile());
+ }
+ }
+ if (fileMetas.equals(state.deletionVectorFiles)) {
+ return state.deletionVectors;
+ }
+ if (fileMetas.isEmpty()) {
+ state.deletionVectorFiles = fileMetas;
+ return Collections.emptyMap();
+ }
+ // Record the file list only after the read succeeds. A failure leaves
the previous
+ // list and vectors in place, so the next refresh retries instead of
trusting them.
+ Map<String, DeletionVector> loaded =
+ indexFileHandler.readAllDeletionVectors(partition, bucket,
fileMetas);
+ state.deletionVectorFiles = fileMetas;
+ return loaded;
+ }
+
+ private List<IndexManifestEntry> currentDeletionVectorEntries() {
+ Snapshot latest = snapshotManager.latestSnapshot();
Review Comment:
Fixed. The event/split snapshot id is now passed into `refreshFiles`, and
deletion vectors are loaded from that snapshot together with the files.
`testDeletionVectorUsesTheFileSnapshot` holds the S1 file plan, writes `(1,
99)`, partially compacts S2, then refreshes with the held S1 files. Key 1 stays
`10` and key 2 stays `20`. The S2 vector is no longer applied to the S1 file.
`PrimaryKeyPartialLookupTable` passes `split.snapshotId()` for both
`DataSplit` and `IncrementalSplit`. A negative id means the files are not
committed yet, and no deletion vector is applied. If the snapshot file has
expired, refresh throws `OutOfRangeException` rather than looking up without
its deletion vectors.
--
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]
