Skip to content

Copy relocated leaf pages as bytes during compaction #74

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

Closed
venkat1701 wants to merge 3 commits into from
Closed

Copy relocated leaf pages as bytes during compaction #74

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
dee4e8e
perf(engine): copy relocated leaf pages as bytes
venkat1701 Oct 7, 2026
7aedab8
test(engine): a corrupt page is not copied with a fresh checksum
venkat1701 Oct 7, 2026
9cf85d2
fix(engine): copy a relocated page before giving it a new id
venkat1701 Oct 8, 2026
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
28 changes: 22 additions & 6 deletions engine/src/main/java/io/hstore/engine/tree/TreeWalker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import java.util.Map;
import java.util.function.LongConsumer;
import java.util.function.LongPredicate;
import java.util.function.UnaryOperator;

public final class TreeWalker {

Expand Down Expand Up @@ -50,20 +51,35 @@ public Ref relocate(Ref root, TreeSchema<?> schema, LongPredicate moving, WriteS
}

public Ref relocate(Ref root, TreeSchema<?> schema, LongPredicate moving, WriteScope scope, LongConsumer reached) {
return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope, reached);
return relocate(root, schema, moving, scope, reached, leaf -> leaf);
}

private Ref relocate(Ref ref, TreeSchema<?> schema, int height, LongPredicate inVictim, WriteScope scope, LongConsumer reached) {
if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) {
public Ref relocate(Ref root, TreeSchema<?> schema, LongPredicate moving, WriteScope scope, LongConsumer reached,
UnaryOperator<Ref.Stored> copyLeaf) {
return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope, reached, copyLeaf);
}

private Ref relocate(Ref ref, TreeSchema<?> schema, int height, LongPredicate inVictim, WriteScope scope, LongConsumer reached,
UnaryOperator<Ref.Stored> copyLeaf) {
if (!(ref instanceof Ref.Stored stored)) {
return ref;
}
long pageId = stored.pageId();
reached.accept(pageId);
boolean moving = inVictim.test(pageId);
if (height == 0 && !moving && !schema.codec().holdsRefs()) {
return ref;
if (!schema.codec().holdsRefs()) {
if (height == 0 && !moving) {
return ref;
}
if (moving && height <= 0) {
Ref.Stored copy = copyLeaf.apply(stored);
if (copy != stored) {
return copy;
}
}
}
return rebuild(ref, pageId, schema, moving, scope, (child, childSchema, childHeight) ->
relocate(child, childSchema, childHeight, inVictim, scope, reached));
relocate(child, childSchema, childHeight, inVictim, scope, reached, copyLeaf));
}

public void pairMoves(Ref before, Ref after, TreeSchema<?> schema, Map<Long, Ref> moved) {
Expand Down
33 changes: 27 additions & 6 deletions engine/src/main/java/io/hstore/engine/txn/TransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
import io.hstore.engine.feed.CommitEvent;
import io.hstore.engine.feed.FeedCodec;
import io.hstore.engine.page.Checksums;
import io.hstore.engine.page.PageHeader;
import io.hstore.engine.page.PageId;
import io.hstore.engine.page.PageStore;
import io.hstore.engine.topology.MemberChange;
import io.hstore.engine.tree.Hashing;
Expand Down Expand Up @@ -326,12 +328,15 @@ public long relocate(LongPredicate moving) {
List<PageWrite> writes = new ArrayList<>();
Materializer materializer = materializer(snapshot.id() + 1, txnId, records, writes);
WriteScope copies = new WriteScope();
List<Runnable> pairings = new ArrayList<>();
snapshot.branches().values().stream().filter(Branch::isActive).forEach(branch -> branch.roots().roots().forEach((slot, ref) -> {
TreeSchema<?> schema = storage.slots().slot(slot).schema();
Ref copy = materializer.materialize(walker.relocate(ref, schema, moving, copies, reached::add), schema);
walker.pairMoves(ref, copy, schema, moved);
Ref copy = materializer.materialize(walker.relocate(ref, schema, moving, copies, reached::add,
leaf -> copyPage(leaf, snapshot.id() + 1, txnId, records, writes)), schema);
pairings.add(() -> walker.pairMoves(ref, copy, schema, moved));
}));
writes.forEach(write -> storage.pages().write(write.pageId(), write.image()));
pairings.forEach(Runnable::run);
pagesWritten.add(materializer.pagesWritten());
Set<Integer> swapped = new HashSet<>();
List<WalRecord> unlogged = records;
Expand Down Expand Up @@ -409,15 +414,31 @@ public long allocate(int length) {

@Override
public void accept(long pageId, MemorySegment image, Node frozen) {
records.add(storage.walMode() == WalMode.PAGE_IMAGES
? new WalRecord.Page(txnId, pageId, image.toArray(ValueLayout.JAVA_BYTE))
: new WalRecord.PageRef(txnId, pageId, Math.toIntExact(image.byteSize()), Checksums.crc32c(image)));
writes.add(new PageWrite(pageId, image));
stage(txnId, pageId, image, records, writes);
storage.source().admit(pageId, frozen);
}
}, epoch, storage.pages().pageSize());
}

private void stage(long txnId, long pageId, MemorySegment image, List<WalRecord> records, List<PageWrite> writes) {
records.add(storage.walMode() == WalMode.PAGE_IMAGES
? new WalRecord.Page(txnId, pageId, image.toArray(ValueLayout.JAVA_BYTE))
: new WalRecord.PageRef(txnId, pageId, Math.toIntExact(image.byteSize()), Checksums.crc32c(image)));
writes.add(new PageWrite(pageId, image));
}

private Ref.Stored copyPage(Ref.Stored stored, long epoch, long txnId, List<WalRecord> records, List<PageWrite> writes) {
MemorySegment page = storage.pages().read(stored.pageId());
if (PageHeader.verify(page, stored.pageId()).height() != 0) {
return stored;
}
MemorySegment image = MemorySegment.ofArray(page.asSlice(0, PageHeader.SIZE + PageHeader.payloadLength(page)).toArray(ValueLayout.JAVA_BYTE));
long pageId = storage.pages().allocate(epoch, Math.toIntExact(image.byteSize()));
PageHeader.assign(image, pageId);
stage(txnId, pageId, image, records, writes);
return new Ref.Stored(pageId, PageId.unitsFor(image.byteSize()), stored.summary());
}

private Appended append(Generation latest, Branch changed, long txnId, List<MemberChange> members,
List<SlotChange<?>> slotChanges, List<WalRecord> meta, long pageBudget) {
long pendingPages = changed.roots().roots().entrySet().stream()
Expand Down
36 changes: 36 additions & 0 deletions engine/src/test/java/io/hstore/engine/CompactionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,14 @@

import io.hstore.engine.catalog.Branch;
import io.hstore.engine.maintenance.Compactor;
import io.hstore.engine.page.PageHeader;
import io.hstore.engine.page.PageId;
import io.hstore.engine.page.SegmentInfo;
import io.hstore.engine.page.SegmentState;
import io.hstore.engine.topology.EdgeKind;
import io.hstore.engine.topology.Incidence;
import io.hstore.engine.tree.Leaf;
import io.hstore.engine.tree.Ref;
import io.hstore.engine.tree.TreeWalker;
import io.hstore.engine.txn.CrashPoint;
import io.hstore.engine.txn.Snapshot;
Expand All @@ -19,7 +22,10 @@
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;

import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
Expand Down Expand Up @@ -238,6 +244,36 @@ void commitsOnTwoBranchesDuringCompactionAreKeptAndOldSegmentsGoAway() throws Ex
}
}

@Test
void aCorruptPageIsNotCopiedWithAFreshChecksum() throws Exception {
try (StorageEngine engine = StorageEngine.open(directory, options())) {
Map<Long, TreeSet<Long>> oracle = load(engine);
Set<Integer> sealed = engine.stats().segments().stream().filter(segment -> segment.state() == SegmentState.SEALED)
.map(SegmentInfo::id).collect(Collectors.toSet());
long victim;
try (Snapshot snapshot = engine.snapshot(Branch.MAIN)) {
victim = oracle.keySet().stream()
.map(edge -> snapshot.requireEdge(edge).membership().root())
.filter(root -> root instanceof Ref.Stored(long pageId, int _, var _) && sealed.contains(PageId.segmentOf(pageId)))
.mapToLong(root -> ((Ref.Stored) root).pageId())
.filter(pageId -> engine.transactions().source().load(pageId, snapshot.requireEdge(oracle.keySet().iterator().next()).membership().schema()) instanceof Leaf)
.findFirst().orElse(0L);
}
assertTrue(victim != 0, "no single-leaf membership tree in a sealed segment");
Path segment = directory.resolve("data").resolve("segments").resolve("%08x.seg".formatted(PageId.segmentOf(victim)));
try (FileChannel channel = FileChannel.open(segment, StandardOpenOption.READ, StandardOpenOption.WRITE)) {
long at = (long) PageId.offsetOf(victim) * PageId.UNIT_BYTES + PageHeader.SIZE + 1;
ByteBuffer one = ByteBuffer.allocate(1);
channel.read(one, at);
channel.write(ByteBuffer.wrap(new byte[]{(byte) (one.get(0) ^ 0x5A)}), at);
}
int corrupted = PageId.segmentOf(victim);
HStoreException failure = assertThrows(HStoreException.class,
() -> engine.transactions().relocate(pageId -> PageId.segmentOf(pageId) == corrupted));
assertTrue(failure.getMessage().toLowerCase().contains("checksum"), failure.getMessage());
}
}

@ParameterizedTest
@EnumSource(value = CrashPoint.class, names = {"PAGE", "WAL_APPEND", "DATA_WRITE", "COMMIT_APPEND", "DATA_SYNC", "WAL_SYNC", "CATALOG_PUBLISH"})
void aCrashWhileSwappingInTheCopiesLosesNothing(CrashPoint point) {
Expand Down