Skip to content

[Issue #1415] Retire fully deleted GC groups with checkpoint protection #1416

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

Draft
AntiO2 wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

[Issue #1415] Retire fully deleted GC groups with checkpoint protection #1416

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
2 changes: 1 addition & 1 deletion pixels-common/src/main/java/io/pixelsdb/pixels/common/metadata/MetadataService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1676,7 +1676,7 @@ public File getFileById(long fileId) throws MetadataException

/**
* Atomically promote a temporary GC file to REGULAR and retire the old files.
* @param newFileId the id of the new temporary GC file to promote
* @param newFileId the new temporary GC file to promote, or zero to retire without a replacement
* @param oldFileIds the ids of old files to retire
* @param cleanupAt the cleanup deadline to write on retired old files
* @throws MetadataException if the request fails
Expand Down
3 changes: 3 additions & 0 deletions pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/dao/impl/RdbFileDao.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,10 +351,13 @@ public boolean deleteByIds(List<Long> ids)
@Override
public boolean atomicSwapFiles(long newFileId, List<Long> oldFileIds, long cleanupAt)
{
if (newFileId < 0 || (newFileId == 0 && (oldFileIds == null || oldFileIds.isEmpty())))
return false;
Connection conn = db.getConnection();
try
{
conn.setAutoCommit(false);
if (newFileId > 0)
try (PreparedStatement pst = conn.prepareStatement(
"UPDATE FILES SET FILE_TYPE=?, FILE_CLEANUP_AT=NULL WHERE FILE_ID=?"))
{
Expand Down
2 changes: 1 addition & 1 deletion pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,7 +261,7 @@ private RecoveryResult recoverRetinaState(RecoveryContext context) throws Retina
recoverableFileIds.add(fileId);
}

context.storageGcWalRecoveryHandler.recover(recoverableFileIds);
context.storageGcWalRecoveryHandler.recover(recoverableFileIds, body.getCheckpointAppliedTs());

for (VisibilityEntry ve : validRgEntries)
{
Expand Down
22 changes: 22 additions & 0 deletions pixels-daemon/src/test/java/io/pixelsdb/pixels/daemon/metadata/dao/TestRdbFileDao.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -604,4 +604,26 @@ private static Connection swapConnection(Connection replacement) throws Exceptio
f.set(MetaDBUtil.Instance(), replacement);
return previous;
}

@Test
public void atomicSwapFiles_retiresFullyDeletedGroupWithoutReplacement() throws Exception
{
PreparedStatement statement = mock(PreparedStatement.class);
when(mockConn.prepareStatement(anyString())).thenReturn(statement);
assertTrue(dao.atomicSwapFiles(0, Arrays.asList(11L, 12L), 123456L));
ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class);
verify(mockConn).prepareStatement(sql.capture());
assertFalse(sql.getValue().contains("FILE_CLEANUP_AT=NULL"));
verify(statement).setLong(3, 11L);
verify(statement).setLong(3, 12L);
verify(statement).executeBatch();
verify(mockConn).commit();
}

@Test
public void atomicSwapFiles_rejectsRetirementWithoutOldFiles() throws Exception
{
assertFalse(dao.atomicSwapFiles(0, Collections.emptyList(), 123456L));
verify(mockConn, never()).commit();
}
}
12 changes: 10 additions & 2 deletions pixels-retina/src/main/java/io/pixelsdb/pixels/retina/RetinaResourceManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -412,9 +412,16 @@ public void processRetiredFiles()
{
// In-memory queue for files retired in this process.
long now = System.currentTimeMillis();
final Set<Long> pending;
try { pending = storageGcWal.collectPendingFileIds(); }
catch (RuntimeException e)
{
logger.warn("processRetiredFiles: cannot establish checkpoint protection; skipping cleanup", e);
return;
}
retiredFiles.removeIf(rf ->
{
if (now <= rf.retireTimestamp)
if (now <= rf.retireTimestamp || pending.contains(rf.fileId))
{
return false;
}
Expand Down Expand Up @@ -459,6 +466,7 @@ public void processRetiredFiles()
}
for (File file : dueFiles)
{
if (pending.contains(file.getId())) continue;
RetiredPathInfo pathInfo = pathsById.get(file.getPathId());
if (pathInfo == null)
{
Expand Down Expand Up @@ -1041,7 +1049,7 @@ private void runGC()
for (StorageGcWal.Task task : storageGcWal.listAllTasks())
{
if (task.getState() != StorageGcWal.State.SWAPPED_NOT_CHECKPOINTED
|| !checkpointFileIds.contains(task.getNewFileId()))
|| !task.isCoveredByCheckpoint(checkpointFileIds, timestamp))
{
continue;
}
Expand Down
78 changes: 72 additions & 6 deletions pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGarbageCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.logging.log4j.LogManager;
Expand Down Expand Up @@ -481,14 +482,16 @@ List<FileGroup> groupAndMerge(List<FileCandidate> candidates)
{
int vNodeId = vnodeEntry.getKey();
List<FileCandidate> files = vnodeEntry.getValue();
files.sort(Comparator.comparingDouble((FileCandidate c) -> c.invalidRatio).reversed());
files.sort(Comparator.comparingDouble((FileCandidate c) -> c.invalidRatio).reversed()
.thenComparingLong(c -> c.fileId));
splitIntoGroups(groups, tableId, vNodeId, files);
}
}

groups.sort(Comparator.comparingDouble(
(FileGroup g) -> g.files.stream().mapToDouble(c -> c.invalidRatio).average().orElse(0.0))
.reversed());
.reversed().thenComparingLong(g -> g.files.stream().mapToLong(c -> c.fileId)
.min().orElse(Long.MAX_VALUE)));

if (groups.size() > maxFileGroupsPerRun)
{
Expand Down Expand Up @@ -828,12 +831,10 @@ RewriteResult rewriteFileGroup(FileGroup group, long safeGcTs,
}
}

// Edge case: all rows in the group were deleted — skip catalog registration,
// delete the empty file, and return early. The old files will be cleaned up
// by the delayed-cleanup phase once it is implemented.
// Fully deleted groups need no replacement; the caller journals and retires the old files.
if (globalNewRowOffset == 0)
{
logger.info("StorageGC: all rows deleted for table={}, vNodeId={}, skipping empty file",
logger.info("StorageGC: all rows deleted for table={}, vNodeId={}, retiring without replacement",
group.tableId, group.virtualNodeId);
try
{
Expand Down Expand Up @@ -1414,6 +1415,68 @@ private void rollbackSinglePointIndex(RewriteResult result)
// Per-group orchestration
// -------------------------------------------------------------------------

void retireEmptyFileGroup(RewriteResult result, long safeGcTs) throws Exception
{
if (result.newFileId > 0 || result.newFileRgCount != 0 || !result.pendingIndexEntries.isEmpty())
throw new IllegalArgumentException("Retirement requires a fully deleted rewrite result");
List<Long> oldFileIds = result.group.files.stream().map(fc -> fc.fileId).collect(Collectors.toList());
String taskId = "empty_" + RetinaUtils.buildStorageGcJournalTaskId(
result.group.tableId, result.group.virtualNodeId, oldFileIds.get(0));
long deadline = System.currentTimeMillis() + retireDelayMs;
result.walWriter = wal.createTask(taskId, result.group.tableId, result.group.virtualNodeId,
oldFileIds, 0, "", 0, 0);
try
{
result.walWriter.appendRetirementOnly(safeGcTs);
result.walWriter.flush();
try
{
metadataService.atomicSwapFiles(0, oldFileIds, deadline);
}
catch (Exception e)
{
for (long id : oldFileIds)
{
File file = metadataService.getFileById(id);
if (file == null || file.getType() != File.Type.RETIRED
|| !Objects.equals(file.getCleanupAt(), deadline)) throw e;
}
}
result.walWriter.markSwapped();
result.walWriter = null;
for (FileCandidate file : result.group.files)
resourceManager.scheduleRetiredFile(new RetinaResourceManager.RetiredFile(
file.fileId, file.rgCount, file.filePath, deadline, Collections.emptyList()));
}
catch (Exception e)
{
try
{
for (long id : oldFileIds)
{
File file = metadataService.getFileById(id);
if (file == null) throw new MetadataException("Missing retirement rollback file " + id);
if (file.getType() == File.Type.RETIRED)
{
file.setType(File.Type.REGULAR);
file.setCleanupAt(null);
if (!metadataService.updateFile(file))
throw new MetadataException("Cannot roll back retirement file " + id);
}
}
if (result.walWriter != null) result.walWriter.markAborted();
}
catch (Exception rollbackError) { e.addSuppressed(rollbackError); }
finally
{
if (result.walWriter != null) result.walWriter.close();
result.walWriter = null;
}
throw e;
}
}


/**
* Processes a single file group through the complete GC pipeline:
* rewrite → dual-write → visibility sync → index sync → commit.
Expand All @@ -1429,6 +1492,9 @@ void processFileGroup(FileGroup group, long safeGcTs,

if (result.newFileId <= 0)
{
retireEmptyFileGroup(result, safeGcTs);
logger.info("StorageGC completed for FileGroup tableId={}, vNodeId={}, retirementOnly=true",
group.tableId, group.virtualNodeId);
return;
}

Expand Down
68 changes: 64 additions & 4 deletions pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGcWal.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@ public final class StorageGcWal
static final byte RT_CREATE = 1;
static final byte RT_ROLLBACK_ENTRY = 2;
static final byte RT_STATE_TRANSITION = 3;
static final byte RT_RETIREMENT_ONLY = 4;

// ─── State ──────────────────────────────────────────────────────────────

Expand Down Expand Up @@ -149,11 +150,12 @@ public static final class Task
private final int newRowCount;
private final List<RollbackEntry> rollbackEntries;
private final State state;
private final long retirementSafeGcTs;

Task(String taskId, long tableId, int virtualNodeId,
List<Long> oldFileIds, long newFileId, String newFilePath,
long newRowIdStart, int newRowCount,
List<RollbackEntry> rollbackEntries, State state)
List<RollbackEntry> rollbackEntries, State state, long retirementSafeGcTs)
{
this.taskId = taskId;
this.tableId = tableId;
Expand All @@ -165,6 +167,7 @@ public static final class Task
this.newRowCount = newRowCount;
this.rollbackEntries = Collections.unmodifiableList(new ArrayList<>(rollbackEntries));
this.state = state;
this.retirementSafeGcTs = retirementSafeGcTs;
}

public String getTaskId() { return taskId; }
Expand All @@ -177,6 +180,14 @@ public static final class Task
public int getNewRowCount() { return newRowCount; }
public List<RollbackEntry> getRollbackEntries() { return rollbackEntries; }
public State getState() { return state; }
public boolean isRetirementOnly() { return newFileId == 0 && newRowCount == 0; }
public long getRetirementSafeGcTs() { return retirementSafeGcTs; }
public boolean isCoveredByCheckpoint(Set<Long> visibleFileIds, long checkpointAppliedTs)
{
return isRetirementOnly()
? retirementSafeGcTs >= 0 && checkpointAppliedTs >= retirementSafeGcTs
: visibleFileIds.contains(newFileId);
}
}

// ─── Writer (per-task write handle) ─────────────────────────────────────
Expand Down Expand Up @@ -230,6 +241,15 @@ public void flush() throws IOException
out.flush();
}

/** Records the visibility-fold timestamp before retiring a fully deleted group. */
public void appendRetirementOnly(long safeGcTs) throws IOException
{
if (safeGcTs < 0) throw new IllegalArgumentException("Negative retirement timestamp");
out.writeByte(RT_RETIREMENT_ONLY);
out.writeInt(Long.BYTES);
out.writeLong(safeGcTs);
}

/** Writes {@code SWAPPED_NOT_CHECKPOINTED} record and closes the stream. */
public void markSwapped() throws IOException
{
Expand Down Expand Up @@ -397,7 +417,7 @@ public Set<Long> collectPendingFileIds()
{
if (task.getState().isPending())
{
fileIds.add(task.getNewFileId());
if (task.getNewFileId() > 0) fileIds.add(task.getNewFileId());
fileIds.addAll(task.getOldFileIds());
}
}
Expand Down Expand Up @@ -540,6 +560,7 @@ private Task replayWal(String path) throws IOException
// ── Subsequent records (rollback entries + state transitions) ──
List<RollbackEntry> rollbackEntries = new ArrayList<>();
State state = State.INDEX_SWITCHING;
long retirementSafeGcTs = -1;

while (true)
{
Expand Down Expand Up @@ -575,6 +596,14 @@ private Task replayWal(String path) throws IOException
rollbackEntries.add(new RollbackEntry(
IndexProto.IndexKey.parseFrom(keyBytes), oldRowId, newRowId));
}
else if ((byte) rt == RT_RETIREMENT_ONLY)
{
if (newFileId != 0 || newRowCount != 0 || rLen != Long.BYTES)
throw new IllegalStateException("Invalid retirement-only WAL record");
retirementSafeGcTs = new DataInputStream(new ByteArrayInputStream(rPayload)).readLong();
if (retirementSafeGcTs < 0)
throw new IllegalStateException("Negative retirement timestamp in WAL");
}
else if ((byte) rt == RT_STATE_TRANSITION)
{
state = State.fromCode(rPayload[0]);
Expand All @@ -587,7 +616,7 @@ else if ((byte) rt == RT_STATE_TRANSITION)

return new Task(taskId, tableId, virtualNodeId, oldFileIds,
newFileId, newFilePath, newRowIdStart, newRowCount,
rollbackEntries, state);
rollbackEntries, state, retirementSafeGcTs);
}
}

Expand All @@ -614,6 +643,11 @@ public RecoveryHandler(StorageGcWal wal,
}

public void recover(Set<Long> baselineVisibleFileIds) throws RetinaException
{
recover(baselineVisibleFileIds, -1L);
}

public void recover(Set<Long> baselineVisibleFileIds, long checkpointAppliedTs) throws RetinaException
{
Set<Long> baseline = baselineVisibleFileIds == null
? Collections.emptySet() : baselineVisibleFileIds;
Expand All @@ -626,6 +660,11 @@ public void recover(Set<Long> baselineVisibleFileIds) throws RetinaException
// back, restoring old files only when the catalog confirms the swap committed.
for (Task task : wal.listAllTasks())
{
if (task.isRetirementOnly())
{
recoverRetirement(task, checkpointAppliedTs);
continue;
}
switch (task.getState())
{
case CHECKPOINTED:
Expand Down Expand Up @@ -692,6 +731,26 @@ public void recover(Set<Long> baselineVisibleFileIds) throws RetinaException
}
}

private void recoverRetirement(Task task, long checkpointAppliedTs) throws RetinaException
{
boolean covered = task.isCoveredByCheckpoint(Collections.emptySet(), checkpointAppliedTs);
if (task.getState() == State.ABORTED) return;
if (task.getState() == State.CHECKPOINTED)
{
if (!covered) throw new RetinaException("Checkpoint predates retirement task " + task.getTaskId());
return;
}
if (task.getState() == State.SWAPPED_NOT_CHECKPOINTED && covered)
{
try { wal.markCheckpointed(task.getTaskId()); }
catch (IOException e) { throw new RetinaException("Cannot checkpoint retirement " + task.getTaskId(), e); }
return;
}
if (task.getState() == State.INDEX_SWITCHING && covered)
throw new RetinaException("Unfinished retirement appears in checkpoint: " + task.getTaskId());
rollbackTask(task, isSwapCommitted(task));
}

private void rollbackTask(Task task, boolean restoreOldFiles) throws RetinaException
{
restorePrimaryIndex(task);
Expand Down Expand Up @@ -784,6 +843,7 @@ private void restoreOldFileCatalog(Task task) throws RetinaException

private void cleanupNewFile(Task task) throws RetinaException
{
if (task.isRetirementOnly()) return;
if (task.getNewRowCount() > 0)
{
try
Expand Down Expand Up @@ -847,7 +907,7 @@ private void cleanupNewFile(Task task) throws RetinaException

private boolean isSwapCommitted(Task task) throws RetinaException
{
File newFile = loadOptionalFile(task.getNewFileId());
File newFile = task.isRetirementOnly() ? null : loadOptionalFile(task.getNewFileId());
if (newFile != null && newFile.getType() == File.Type.REGULAR)
{
return true;
Expand Down
Loading
Loading