From db25c9767a17ab2762884065cf22ed41ef66b508 Mon Sep 17 00:00:00 2001 From: AntiO2 Date: Sun, 11 Oct 2026 16:49:07 +0800 Subject: [PATCH] [Issue #1413] Route GC primary-index switches and rollback by key bucket --- .../retina/StorageGarbageCollector.java | 61 +++++++++++++------ .../pixelsdb/pixels/retina/StorageGcWal.java | 9 +-- .../retina/TestStorageGcIndexBuckets.java | 55 +++++++++++++++++ 3 files changed, 103 insertions(+), 22 deletions(-) create mode 100644 pixels-retina/src/test/java/io/pixelsdb/pixels/retina/TestStorageGcIndexBuckets.java diff --git a/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGarbageCollector.java b/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGarbageCollector.java index 2ac5a71fa..cc14afcbf 100644 --- a/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGarbageCollector.java +++ b/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGarbageCollector.java @@ -37,6 +37,7 @@ import io.pixelsdb.pixels.common.physical.Storage; import io.pixelsdb.pixels.common.physical.StorageFactory; import io.pixelsdb.pixels.common.utils.NetUtils; +import io.pixelsdb.pixels.common.utils.IndexUtils; import io.pixelsdb.pixels.common.utils.PixelsFileNameUtils; import io.pixelsdb.pixels.common.utils.RetinaUtils; import io.pixelsdb.pixels.core.PixelsFooterCache; @@ -1113,7 +1114,7 @@ void syncIndex(RewriteResult result, long tableId) throws Exception if (!result.pendingIndexEntries.isEmpty()) { - result.oldRowIds = updateSinglePointIndex(result, tableId, primaryIndexId, indexOption, newRowIdStart); + result.oldRowIds = updateSinglePointIndex(result, tableId, primaryIndexId, newRowIdStart); } } @@ -1159,7 +1160,7 @@ private void insertMainIndexEntries(RewriteResult result, long tableId, long pri * {@code IndexService} grows a batch API that returns the rowIds atomically replaced. */ private List updateSinglePointIndex(RewriteResult result, long tableId, long primaryIndexId, - IndexOption indexOption, long newRowIdStart) throws Exception + long newRowIdStart) throws Exception { int size = result.pendingIndexEntries.size(); List keys = new ArrayList<>(size); @@ -1176,27 +1177,43 @@ private List updateSinglePointIndex(RewriteResult result, long tableId, lo .build()); } - List> resolved = - indexService.resolvePrimary(tableId, primaryIndexId, keys, indexOption); - List oldRowIds = new ArrayList<>(size); + Map> buckets = new LinkedHashMap<>(); for (int i = 0; i < size; i++) { - long oldRowId = resolved.get(i).map(ResolvedPrimary::getRowId).orElse(-1L); - oldRowIds.add(oldRowId); - if (oldRowId < 0) + int bucket = IndexUtils.getBucketIdFromByteBuffer(keys.get(i).getKey()); + buckets.computeIfAbsent(bucket, ignored -> new ArrayList<>()).add(i); + } + List oldRowIds = new ArrayList<>(Collections.nCopies(size, -1L)); + for (Map.Entry> bucket : buckets.entrySet()) + { + List bucketKeys = new ArrayList<>(); + for (int i : bucket.getValue()) bucketKeys.add(keys.get(i)); + List> resolved = indexService.resolvePrimary(tableId, primaryIndexId, + bucketKeys, IndexOption.builder().vNodeId(bucket.getKey()).build()); + if (resolved.size() != bucketKeys.size()) throw new IllegalStateException("GC primary resolve count mismatch"); + for (int i = 0; i < resolved.size(); i++) + oldRowIds.set(bucket.getValue().get(i), resolved.get(i).map(ResolvedPrimary::getRowId).orElse(-1L)); + } + // Persist every rollback entry before switching any bucket; preserve row-order alignment. + for (int i = 0; i < size; i++) + { + if (oldRowIds.get(i) < 0) { logger.warn("StorageGC syncIndex: no resolvable primary for tableId={}, " + "newGlobalRowOffset={} — index may be inconsistent", tableId, result.pendingIndexEntries.get(i).newGlobalRowOffset); } - result.walWriter.appendRollbackEntry( - keys.get(i), oldRowId, entries.get(i).getRowId()); + result.walWriter.appendRollbackEntry(keys.get(i), oldRowIds.get(i), entries.get(i).getRowId()); } - // Flush all rollback entries to durable storage before the batch index update so - // that recovery can restore the old rowIds if we crash between this flush and commit. result.walWriter.flush(); - - indexService.updatePrimaryIndexEntriesOnly(tableId, primaryIndexId, entries, indexOption); + result.oldRowIds = oldRowIds; + for (Map.Entry> bucket : buckets.entrySet()) + { + List bucketEntries = new ArrayList<>(); + for (int i : bucket.getValue()) bucketEntries.add(entries.get(i)); + indexService.updatePrimaryIndexEntriesOnly(tableId, primaryIndexId, bucketEntries, + IndexOption.builder().vNodeId(bucket.getKey()).build()); + } return oldRowIds; } @@ -1358,6 +1375,15 @@ void rollback(RewriteResult result) } } + static Map> indexRollbackBuckets(List entries) + { + Map> buckets = new LinkedHashMap<>(); + for (RollbackEntry entry : entries) + buckets.computeIfAbsent(IndexUtils.getBucketIdFromByteBuffer(entry.getIndexKey().getKey()), + ignored -> new ArrayList<>()).add(entry); + return buckets; + } + private void rollbackSinglePointIndex(RewriteResult result) { try @@ -1369,8 +1395,6 @@ private void rollbackSinglePointIndex(RewriteResult result) return; } long primaryIndexId = primaryIndex.getId(); - IndexOption indexOption = IndexOption.builder().vNodeId(result.group.virtualNodeId).build(); - // Alignment invariant: oldRowIds.size() == pendingIndexEntries.size() // (established in updateSinglePointIndex). Walk them in lockstep by // position and skip -1 placeholders — those slots had no prior entry @@ -1399,8 +1423,9 @@ private void rollbackSinglePointIndex(RewriteResult result) } if (!rollbackEntries.isEmpty()) { - indexService.restorePrimaryIndexEntries( - result.group.tableId, primaryIndexId, rollbackEntries, indexOption); + for (Map.Entry> bucket : indexRollbackBuckets(rollbackEntries).entrySet()) + indexService.restorePrimaryIndexEntries(result.group.tableId, primaryIndexId, + bucket.getValue(), IndexOption.builder().vNodeId(bucket.getKey()).build()); } } catch (Exception e) diff --git a/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGcWal.java b/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGcWal.java index 4d0423497..805ed0bce 100644 --- a/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGcWal.java +++ b/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/StorageGcWal.java @@ -46,6 +46,7 @@ import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.Set; @@ -724,12 +725,12 @@ private void restorePrimaryIndex(Task task) throws RetinaException return; } long primaryIndexId = getPrimaryIndexId(task.getTableId()); - IndexOption indexOption = IndexOption.builder() - .vNodeId(task.getVirtualNodeId()).build(); try { - indexService.restorePrimaryIndexEntries( - task.getTableId(), primaryIndexId, entries, indexOption); + for (Map.Entry> bucket : + StorageGarbageCollector.indexRollbackBuckets(entries).entrySet()) + indexService.restorePrimaryIndexEntries(task.getTableId(), primaryIndexId, + bucket.getValue(), IndexOption.builder().vNodeId(bucket.getKey()).build()); } catch (IndexException | UnsupportedOperationException e) { diff --git a/pixels-retina/src/test/java/io/pixelsdb/pixels/retina/TestStorageGcIndexBuckets.java b/pixels-retina/src/test/java/io/pixelsdb/pixels/retina/TestStorageGcIndexBuckets.java new file mode 100644 index 000000000..75579fb3f --- /dev/null +++ b/pixels-retina/src/test/java/io/pixelsdb/pixels/retina/TestStorageGcIndexBuckets.java @@ -0,0 +1,55 @@ +package io.pixelsdb.pixels.retina; +import com.google.protobuf.ByteString; +import io.pixelsdb.pixels.common.index.*; +import io.pixelsdb.pixels.common.index.service.IndexService; +import io.pixelsdb.pixels.common.utils.*; +import io.pixelsdb.pixels.core.encoding.EncodingLevel; +import io.pixelsdb.pixels.index.IndexProto; +import java.lang.reflect.*; +import java.nio.ByteBuffer; +import java.nio.file.*; +import java.util.*; +import org.junit.Test; +import static org.junit.Assert.*; +public class TestStorageGcIndexBuckets { + @Test public void rewriteUsesIndexBucketsAndKeepsRollbackAlignedBeforePartialFailure() throws Exception { + ConfigFactory.Instance().addProperty("index.bucket.num","16"); + Path dir=Files.createTempDirectory("gc-index-buckets-"); + ConfigFactory.Instance().addProperty("retina.storage.gc.journal.dir",dir.toUri().toString()); + StorageGcWal wal=new StorageGcWal(); + List pending=new ArrayList<>(); + for(int i=0;i<100;i++)pending.add(new StorageGarbageCollector.PendingIndexEntry(i,ByteString.copyFrom(ByteBuffer.allocate(4).putInt(i).array()),10)); + StorageGarbageCollector.RewriteResult result=new StorageGarbageCollector.RewriteResult(new StorageGarbageCollector.FileGroup(17,3,Collections.emptyList()),"file:///new.pxl",55,1,new int[]{100},new int[]{0,100},Collections.emptyMap(),Collections.emptyList(),pending); + result.walWriter=wal.createTask("test",17,3,Collections.singletonList(44L),55,"file:///new.pxl",1000,100); + final int[] updates={0}; + IndexService index=(IndexService)Proxy.newProxyInstance(IndexService.class.getClassLoader(),new Class[]{IndexService.class},(proxy,method,args)->{ + IndexOption option=(IndexOption)args[3]; + if(method.getName().equals("resolvePrimary")){ + List> out=new ArrayList<>(); + for(IndexProto.IndexKey key:(List)args[2]){ + assertEquals(IndexUtils.getBucketIdFromByteBuffer(key.getKey()),option.getVNodeId()); + int n=key.getKey().asReadOnlyByteBuffer().getInt();out.add(Optional.of(new ResolvedPrimary(100+n,IndexProto.RowLocation.newBuilder().setFileId(44).setRgRowOffset(n).build()))); + }return out; + } + if(method.getName().equals("updatePrimaryIndexEntriesOnly")){ + assertEquals(100,wal.getTask("test").get().getRollbackEntries().size()); + for(IndexProto.PrimaryIndexEntry entry:(List)args[2])assertEquals(IndexUtils.getBucketIdFromByteBuffer(entry.getIndexKey().getKey()),option.getVNodeId()); + if(++updates[0]==2)throw new io.pixelsdb.pixels.common.exception.IndexException("Injected partial failure"); + return true; + } + throw new AssertionError(method.getName()); + }); + StorageGarbageCollector gc=new StorageGarbageCollector(null,null,index,.5,1024,20,10,100,EncodingLevel.from(2),0,wal); + Method update=StorageGarbageCollector.class.getDeclaredMethod("updateSinglePointIndex",StorageGarbageCollector.RewriteResult.class,long.class,long.class,long.class);update.setAccessible(true); + try{update.invoke(gc,result,17L,16L,1000L);fail("Expected injected failure");} + catch(InvocationTargetException e){assertTrue(e.getCause() instanceof io.pixelsdb.pixels.common.exception.IndexException);} + assertEquals(2,updates[0]);assertEquals(100,result.oldRowIds.size()); + List rollback=wal.getTask("test").get().getRollbackEntries(); + for(int i=0;i<100;i++){assertEquals(100L+i,(long)result.oldRowIds.get(i));assertEquals(100L+i,rollback.get(i).getOldRowId());assertEquals(1000L+i,rollback.get(i).getNewRowId());} + int restored=0; + for(Map.Entry> bucket:StorageGarbageCollector.indexRollbackBuckets(rollback).entrySet())for(RollbackEntry entry:bucket.getValue()){ + assertEquals(IndexUtils.getBucketIdFromByteBuffer(entry.getIndexKey().getKey()),(int)bucket.getKey());restored++; + } + assertEquals(100,restored);result.walWriter.close(); + } +}