feat: finalize change set file format

flake.nix

@@ -16,7 +16,7 @@
             let
               buildSystems = {
                 rocksdb = [ "setuptools" "cython" "pkgconfig" ];
-                cprotobuf = [ "setuptools" ];
+                cprotobuf = [ "setuptools" "cython" ];
               };
             in
             lib.mapAttrs

iavl/cli.py

@@ -1,6 +1,7 @@
 import binascii
 import hashlib
 import json
+import mmap
 import sys
 from pathlib import Path
 from typing import List, Optional
@@ -9,7 +10,7 @@
 from hexbytes import HexBytes
 
 from . import dbm, diff
-from .iavl import NodeDB, Tree, delete_version
+from .iavl import DEFAULT_CACHE_SIZE, NodeDB, Tree, delete_version
 from .utils import (
     decode_fast_node,
     diff_iterators,
@@ -376,30 +377,71 @@ def visualize(db, version, store=None, include_prev_version=False):
     type=click.Path(exists=True),
     required=True,
 )
-def dump_changesets(db, start_version, end_version, store: Optional[str], out_dir: str):
+@click.option(
+    "--cache-size",
+    help="the output directory to save the data files",
+    default=DEFAULT_CACHE_SIZE,
+)
+def dump_changesets(
+    db, start_version, end_version, store: Optional[str], out_dir: str, cache_size: int
+):
     """
     extract changeset by comparing iavl versions and save in files
     with compatible format with file streamer.
     end_version is exclusive.
     """
     db = dbm.open(str(db), read_only=True)
     prefix = store_prefix(store) if store is not None else b""
-    ndb = NodeDB(db, prefix=prefix)
-    for _, v, _, changeset in iter_state_changes(
-        db, ndb, start_version=start_version, end_version=end_version, prefix=prefix
-    ):
-        with (Path(out_dir) / f"block-{v}-data").open("wb") as fp:
-            diff.write_change_set(fp, changeset)
+    ndb = NodeDB(db, prefix=prefix, cache_size=cache_size)
+
+    last_version = None
+    offset = 0
+    output = Path(out_dir) / f"block-{start_version}"
+    if output.exists():
+        with output.open("rb") as fp:
+            last_version, offset = diff.seek_last_version(fp)
+
+    with output.open("ab") as fp:
+        fp.seek(offset)
+        fp.truncate()
+        if offset == 0:
+            fp.write(diff.VERSIONDB_MAGIC)
+        if last_version is not None:
+            start_version = last_version + 1
+            print("continue from", start_version)
+        else:
+            print("start from", start_version)
+        for _, v, _, changeset in iter_state_changes(
+            db, ndb, start_version=start_version, end_version=end_version, prefix=prefix
+        ):
+            diff.append_change_set(fp, v, changeset)
 
 
 @cli.command()
 @click.argument("file", type=click.Path(exists=True))
-def print_changeset(file):
+@click.option(
+    "--parse-kv-pairs/--no-parse-kv-pairs",
+    default=True,
+    help="if parse the changeset kv pairs",
+)
+def print_changesets(file, parse_kv_pairs):
     """
     decode and print the content of changeset files
     """
-    for item in diff.parse_change_set(Path(file).read_bytes()):
-        print(json.dumps(item.as_json()))
+    with Path(file).open("rb") as fp:
+        with mmap.mmap(fp.fileno(), 0, access=mmap.ACCESS_READ) as data:
+            if parse_kv_pairs:
+                data.madvise(mmap.MADV_NORMAL)
+            else:
+                data.madvise(mmap.MADV_RANDOM)
+            for version, items in diff.parse_change_set(
+                memoryview(data), parse_kv_pairs
+            ):
+                print("version:", version)
+                if items is None:
+                    continue
+                for item in items:
+                    print(json.dumps(item.as_json()))
 
 
 @cli.command()