Use commit_id as segment id

Author: lalinsky

src/FileSegment.zig

@@ -8,7 +8,7 @@ const SearchResults = common.SearchResults;
 const SegmentId = common.SegmentId;
 const KeepOrDelete = common.KeepOrDelete;
 
-const MemorySegment = @import("MemorySegment.zig");
+const SegmentInfo = @import("segment.zig").SegmentInfo;
 
 const filefmt = @import("filefmt.zig");
 
@@ -20,8 +20,7 @@ pub const Options = struct {
 
 allocator: std.mem.Allocator,
 dir: std.fs.Dir,
-id: SegmentId = .{ .version = 0, .included_merges = 0 },
-max_commit_id: u64 = 0,
+info: SegmentInfo = .{},
 docs: std.AutoHashMap(u32, bool),
 index: std.ArrayList(u32),
 block_size: usize = 0,
@@ -82,19 +81,19 @@ pub fn search(self: Self, sorted_hashes: []const u32, results: *SearchResults) !
             }
             const matches = std.sort.equalRange(Item, Item{ .hash = hash, .id = 0 }, block_items.items, {}, Item.cmpByHash);
             for (matches[0]..matches[1]) |i| {
-                try results.incr(block_items.items[i].id, self.id.version);
+                try results.incr(block_items.items[i].id, self.info.version);
             }
         }
     }
 }
 
-pub fn open(self: *Self, id: SegmentId) !void {
-    try filefmt.readSegmentFile(self.dir, id, self);
+pub fn open(self: *Self, info: SegmentInfo) !void {
+    try filefmt.readSegmentFile(self.dir, info, self);
 }
 
 pub fn delete(self: *Self) void {
     var file_name_buf: [filefmt.max_file_name_size]u8 = undefined;
-    const file_name = filefmt.buildSegmentFileName(&file_name_buf, self.id);
+    const file_name = filefmt.buildSegmentFileName(&file_name_buf, self.info);
 
     log.info("deleting segment file {s}", .{file_name});
 
@@ -113,18 +112,20 @@ pub fn merge(self: *Self, source: anytype) !void {
 
 pub fn build(self: *Self, source: anytype) !void {
     var file_name_buf: [filefmt.max_file_name_size]u8 = undefined;
-    const file_name = filefmt.buildSegmentFileName(&file_name_buf, source.segment.id);
+    const file_name = filefmt.buildSegmentFileName(&file_name_buf, source.segment.info);
 
     try filefmt.writeSegmentFile(self.dir, source);
 
     errdefer self.dir.deleteFile(file_name) catch |err| {
         log.err("failed to clean up segment file {s}: {}", .{ file_name, err });
     };
 
-    try filefmt.readSegmentFile(self.dir, source.segment.id, self);
+    try filefmt.readSegmentFile(self.dir, source.segment.info, self);
 }
 
 test "build" {
+    const MemorySegment = @import("MemorySegment.zig");
+
     var tmp_dir = std.testing.tmpDir(.{});
     defer tmp_dir.cleanup();
 
@@ -134,7 +135,7 @@ test "build" {
     var source = MemorySegment.init(std.testing.allocator, .{});
     defer source.deinit(.delete);
 
-    source.id.version = 1;
+    source.info = .{ .version = 1 };
     source.frozen = true;
     try source.docs.put(1, true);
     try source.items.append(.{ .id = 1, .hash = 1 });
@@ -148,8 +149,7 @@ test "build" {
 
     try segment.build(&source_reader);
 
-    try std.testing.expectEqual(1, segment.id.version);
-    try std.testing.expectEqual(0, segment.id.included_merges);
+    try std.testing.expectEqualDeep(SegmentInfo{ .version = 1, .merges = 0 }, segment.info);
     try std.testing.expectEqual(1, segment.docs.count());
     try std.testing.expectEqual(1, segment.index.items.len);
 }

src/Index.zig

@@ -8,7 +8,7 @@ const Deadline = @import("utils/Deadline.zig");
 const Change = @import("change.zig").Change;
 const SearchResult = @import("common.zig").SearchResult;
 const SearchResults = @import("common.zig").SearchResults;
-const SegmentId = @import("common.zig").SegmentId;
+const SegmentInfo = @import("segment.zig").SegmentInfo;
 
 const Oplog = @import("Oplog.zig");
 
@@ -142,11 +142,11 @@ pub fn deinit(self: *Self) void {
     self.dir.close();
 }
 
-fn loadSegment(self: *Self, segment_id: SegmentId) !FileSegmentNode {
+fn loadSegment(self: *Self, info: SegmentInfo) !FileSegmentNode {
     var node = try FileSegmentList.createSegment(self.allocator, .{ .dir = self.dir });
     errdefer FileSegmentList.destroySegment(self.allocator, &node);
 
-    try node.value.open(segment_id);
+    try node.value.open(info);
 
     return node;
 }
@@ -169,13 +169,13 @@ fn loadSegments(self: *Self) !u64 {
 
     try self.file_segments.segments.value.nodes.ensureTotalCapacity(self.allocator, segment_ids.len);
 
-    var max_commit_id: u64 = 0;
+    var last_commit_id: u64 = 0;
     for (segment_ids) |segment_id| {
         const node = try self.loadSegment(segment_id);
         self.file_segments.segments.value.nodes.appendAssumeCapacity(node);
-        max_commit_id = @max(max_commit_id, node.value.max_commit_id);
+        last_commit_id = node.value.info.getLastCommitId();
     }
-    return max_commit_id;
+    return last_commit_id;
 }
 
 fn doCheckpoint(self: *Self) !bool {
@@ -260,7 +260,7 @@ fn stopCheckpointThread(self: *Self) void {
 }
 
 fn updateIndexFile(self: *Self, segments: *FileSegmentList) !void {
-    var ids = try segments.getIds(self.allocator);
+    var ids = try segments.getInfos(self.allocator);
     defer ids.deinit(self.allocator);
 
     try filefmt.writeIndexFile(self.dir, ids.items);
@@ -359,14 +359,14 @@ fn stopMemorySegmentMergeThread(self: *Self) void {
 }
 
 pub fn open(self: *Self) !void {
-    const max_commit_id = try self.loadSegments();
+    const last_commit_id = try self.loadSegments();
 
     // start these threads after loading file segments, but before replaying oplog to memory segments
     try self.startFileSegmentMergeThread();
     try self.startMemorySegmentMergeThread();
     try self.startCheckpointThread();
 
-    try self.oplog.open(max_commit_id + 1, updateInternal, self);
+    try self.oplog.open(last_commit_id + 1, updateInternal, self);
 
     log.info("index loaded", .{});
 }
@@ -406,21 +406,11 @@ pub fn updateInternal(self: *Self, changes: []const Change, commit_id: ?u64) !vo
     var upd = try self.memory_segments.beginUpdate(self.allocator);
     defer self.memory_segments.cleanupAfterUpdate(self.allocator, &upd);
 
-    target.value.max_commit_id = commit_id orelse try self.oplog.write(changes);
+    target.value.info.version = commit_id orelse try self.oplog.write(changes);
 
     self.segments_lock.lock();
     defer self.segments_lock.unlock();
 
-    target.value.id = blk: {
-        if (self.memory_segments.segments.value.getLast()) |n| {
-            break :blk n.value.id.next();
-        } else if (self.file_segments.segments.value.getLast()) |n| {
-            break :blk n.value.id.next();
-        } else {
-            break :blk SegmentId.first();
-        }
-    };
-
     upd.appendSegment(target);
 
     self.memory_segments.commitUpdate(&upd);

src/MemorySegment.zig

@@ -4,8 +4,8 @@ const log = std.log;
 const common = @import("common.zig");
 const Item = common.Item;
 const SearchResults = common.SearchResults;
-const SegmentId = common.SegmentId;
 const KeepOrDelete = common.KeepOrDelete;
+const SegmentInfo = @import("segment.zig").SegmentInfo;
 
 const Change = @import("change.zig").Change;
 
@@ -18,8 +18,7 @@ const Self = @This();
 pub const Options = struct {};
 
 allocator: std.mem.Allocator,
-id: SegmentId = .{ .version = 0, .included_merges = 0 },
-max_commit_id: u64 = 0,
+info: SegmentInfo = .{},
 docs: std.AutoHashMap(u32, bool),
 items: std.ArrayList(Item),
 frozen: bool = false,
@@ -44,7 +43,7 @@ pub fn search(self: Self, sorted_hashes: []const u32, results: *SearchResults) !
     for (sorted_hashes) |hash| {
         const matches = std.sort.equalRange(Item, Item{ .hash = hash, .id = 0 }, items, {}, Item.cmpByHash);
         for (matches[0]..matches[1]) |i| {
-            try results.incr(items[i].id, self.id.version);
+            try results.incr(items[i].id, self.info.version);
         }
         items = items[matches[1]..];
     }
@@ -112,8 +111,7 @@ pub fn cleanup(self: *Self) void {
 }
 
 pub fn merge(self: *Self, merger: *SegmentMerger(Self)) !void {
-    self.id = merger.segment.id;
-    self.max_commit_id = merger.segment.max_commit_id;
+    self.info = merger.segment.info;
 
     self.docs.deinit();
     self.docs = merger.segment.docs.move();

src/common.zig

@@ -2,6 +2,7 @@ const std = @import("std");
 const testing = std.testing;
 
 const msgpack = @import("msgpack");
+const SegmentInfo = @import("segment.zig").SegmentInfo;
 
 pub const KeepOrDelete = enum {
     keep,
@@ -59,7 +60,7 @@ test "Item array sort" {
 pub const SearchResult = struct {
     id: u32,
     score: u32,
-    version: u32,
+    version: u64,
 
     pub fn cmp(_: void, a: SearchResult, b: SearchResult) bool {
         return a.score > b.score or (a.score == b.score and a.id > b.id);
@@ -81,7 +82,7 @@ pub const SearchResults = struct {
         self.results.deinit();
     }
 
-    pub fn incr(self: *SearchResults, id: u32, version: u32) !void {
+    pub fn incr(self: *SearchResults, id: u32, version: u64) !void {
         const r = try self.results.getOrPut(id);
         if (!r.found_existing or r.value_ptr.version < version) {
             r.value_ptr.id = id;
@@ -140,73 +141,3 @@ test "sort search results" {
         SearchResult{ .id = 1, .score = 1, .version = 1 },
     }, results.values());
 }
-
-pub const SegmentId = packed struct(u64) {
-    version: u32,
-    included_merges: u32 = 0,
-
-    pub fn cmp(_: void, a: SegmentId, b: SegmentId) bool {
-        const xa: u64 = @bitCast(a);
-        const xb: u64 = @bitCast(b);
-        return xa < xb;
-    }
-
-    pub fn eq(a: SegmentId, b: SegmentId) bool {
-        const xa: u64 = @bitCast(a);
-        const xb: u64 = @bitCast(b);
-        return xa == xb;
-    }
-
-    pub fn contains(self: SegmentId, other: SegmentId) bool {
-        const start = self.version;
-        const end = self.version + self.included_merges;
-
-        const other_start = other.version;
-        const other_end = other.version + other.included_merges;
-
-        return other_start >= start and other_end <= end;
-    }
-
-    pub fn first() SegmentId {
-        return .{
-            .version = 1,
-            .included_merges = 0,
-        };
-    }
-
-    pub fn next(a: SegmentId) SegmentId {
-        return .{
-            .version = a.version + a.included_merges + 1,
-            .included_merges = 0,
-        };
-    }
-
-    pub fn merge(a: SegmentId, b: SegmentId) SegmentId {
-        return .{
-            .version = @min(a.version, b.version),
-            .included_merges = 1 + a.included_merges + b.included_merges,
-        };
-    }
-
-    pub fn msgpackFormat() msgpack.StructFormat {
-        return .{ .as_array = .{} };
-    }
-};
-
-test "SegmentId.contains" {
-    const a = SegmentId{ .version = 1, .included_merges = 0 };
-    const b = SegmentId{ .version = 2, .included_merges = 0 };
-    const c = SegmentId{ .version = 1, .included_merges = 1 };
-
-    try std.testing.expect(a.contains(a));
-    try std.testing.expect(!a.contains(b));
-    try std.testing.expect(!a.contains(c));
-
-    try std.testing.expect(!b.contains(a));
-    try std.testing.expect(b.contains(b));
-    try std.testing.expect(!b.contains(c));
-
-    try std.testing.expect(c.contains(a));
-    try std.testing.expect(c.contains(b));
-    try std.testing.expect(c.contains(c));
-}