Skip to content

[Issue #1413] Route GC primary-index switches and rollback by key bucket #1414

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 #1413] Route GC primary-index switches and rollback by key bucket #1414

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
61 changes: 43 additions & 18 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 @@ -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;
Expand Down Expand Up @@ -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);
}
}

Expand Down Expand Up @@ -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<Long> updateSinglePointIndex(RewriteResult result, long tableId, long primaryIndexId,
IndexOption indexOption, long newRowIdStart) throws Exception
long newRowIdStart) throws Exception
{
int size = result.pendingIndexEntries.size();
List<IndexProto.IndexKey> keys = new ArrayList<>(size);
Expand All @@ -1176,27 +1177,43 @@ private List<Long> updateSinglePointIndex(RewriteResult result, long tableId, lo
.build());
}

List<Optional<ResolvedPrimary>> resolved =
indexService.resolvePrimary(tableId, primaryIndexId, keys, indexOption);
List<Long> oldRowIds = new ArrayList<>(size);
Map<Integer, List<Integer>> 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<Long> oldRowIds = new ArrayList<>(Collections.nCopies(size, -1L));
for (Map.Entry<Integer, List<Integer>> bucket : buckets.entrySet())
{
List<IndexProto.IndexKey> bucketKeys = new ArrayList<>();
for (int i : bucket.getValue()) bucketKeys.add(keys.get(i));
List<Optional<ResolvedPrimary>> 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<Integer, List<Integer>> bucket : buckets.entrySet())
{
List<IndexProto.PrimaryIndexEntry> 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;
}

Expand Down Expand Up @@ -1358,6 +1375,15 @@ void rollback(RewriteResult result)
}
}

static Map<Integer, List<RollbackEntry>> indexRollbackBuckets(List<RollbackEntry> entries)
{
Map<Integer, List<RollbackEntry>> 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
Expand All @@ -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
Expand Down Expand Up @@ -1399,8 +1423,9 @@ private void rollbackSinglePointIndex(RewriteResult result)
}
if (!rollbackEntries.isEmpty())
{
indexService.restorePrimaryIndexEntries(
result.group.tableId, primaryIndexId, rollbackEntries, indexOption);
for (Map.Entry<Integer, List<RollbackEntry>> bucket : indexRollbackBuckets(rollbackEntries).entrySet())
indexService.restorePrimaryIndexEntries(result.group.tableId, primaryIndexId,
bucket.getValue(), IndexOption.builder().vNodeId(bucket.getKey()).build());
}
}
catch (Exception e)
Expand Down
9 changes: 5 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 @@ -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;
Expand Down Expand Up @@ -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<Integer, List<RollbackEntry>> bucket :
StorageGarbageCollector.indexRollbackBuckets(entries).entrySet())
indexService.restorePrimaryIndexEntries(task.getTableId(), primaryIndexId,
bucket.getValue(), IndexOption.builder().vNodeId(bucket.getKey()).build());
}
catch (IndexException | UnsupportedOperationException e)
{
Expand Down
55 changes: 55 additions & 0 deletions pixels-retina/src/test/java/io/pixelsdb/pixels/retina/TestStorageGcIndexBuckets.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<StorageGarbageCollector.PendingIndexEntry> 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<Optional<ResolvedPrimary>> out=new ArrayList<>();
for(IndexProto.IndexKey key:(List<IndexProto.IndexKey>)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<IndexProto.PrimaryIndexEntry>)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<RollbackEntry> 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<Integer,List<RollbackEntry>> 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();
}
}
Loading