Version 3 with cached cross chunk edges

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.18.3
+current_version = 3.0.10
 commit = True
 tag = True
 

.github/workflows/main.yml

@@ -4,9 +4,11 @@ on:
   push:
     branches:
       - "main"
+      - "pcgv3"
   pull_request:
     branches:
       - "main"
+      - "pcgv3"
 
 jobs:
   unit-tests:

pychunkedgraph/__init__.py

@@ -1 +1 @@
-__version__ = "2.18.3"
+__version__ = "3.0.10"

pychunkedgraph/app/__init__.py

@@ -105,6 +105,8 @@ def configure_app(app):
         with app.app_context():
             from ..ingest.rq_cli import init_rq_cmds
             from ..ingest.cli import init_ingest_cmds
+            from ..ingest.cli_upgrade import init_upgrade_cmds
 
             init_rq_cmds(app)
             init_ingest_cmds(app)
+            init_upgrade_cmds(app)

pychunkedgraph/debug/utils.py

@@ -1,7 +1,8 @@
+# pylint: disable=invalid-name, missing-docstring, bare-except, unidiomatic-typecheck
+
 import numpy as np
 
-from ..graph import ChunkedGraph
-from ..graph.utils.basetypes import NODE_ID
+from pychunkedgraph.graph.meta import ChunkedGraphMeta, GraphConfig
 
 
 def print_attrs(d):
@@ -16,28 +17,55 @@ def print_attrs(d):
             print(v)
 
 
-def print_node(
-    cg: ChunkedGraph,
-    node: NODE_ID,
-    indent: int = 0,
-    stop_layer: int = 2,
-) -> None:
+def print_node(cg, node: np.uint64, indent: int = 0, stop_layer: int = 2) -> None:
     children = cg.get_children(node)
     print(f"{' ' * indent}{node}[{len(children)}]")
     if cg.get_chunk_layer(node) <= stop_layer:
         return
     for child in children:
-        print_node(cg, child, indent=indent + 1, stop_layer=stop_layer)
+        print_node(cg, child, indent=indent + 4, stop_layer=stop_layer)
 
 
-def get_l2children(cg: ChunkedGraph, node: NODE_ID) -> np.ndarray:
-    nodes = np.array([node], dtype=NODE_ID)
+def get_l2children(cg, node: np.uint64) -> np.ndarray:
+    nodes = np.array([node], dtype=np.uint64)
     layers = cg.get_chunk_layers(nodes)
-    assert np.all(layers > 2), "nodes must be at layers > 2"
+    assert np.all(layers >= 2), "nodes must be at layers >= 2"
     l2children = []
     while nodes.size:
         children = cg.get_children(nodes, flatten=True)
         layers = cg.get_chunk_layers(children)
         l2children.append(children[layers == 2])
         nodes = children[layers > 2]
     return np.concatenate(l2children)
+
+
+def sanity_check(cg, new_roots, operation_id):
+    """
+    Check for duplicates in hierarchy, useful for debugging.
+    """
+    # print(f"{len(new_roots)} new ids from {operation_id}")
+    l2c_d = {}
+    for new_root in new_roots:
+        l2c_d[new_root] = get_l2children(cg, new_root)
+    success = True
+    for k, v in l2c_d.items():
+        success = success and (len(v) == np.unique(v).size)
+        # print(f"{k}: {np.unique(v).size}, {len(v)}")
+    if not success:
+        raise RuntimeError("Some ids are not valid.")
+
+
+def sanity_check_single(cg, node, operation_id):
+    v = get_l2children(cg, node)
+    msg = f"invalid node {node}:"
+    msg += f" found {len(v)} l2 ids, must be {np.unique(v).size}"
+    assert np.unique(v).size == len(v), f"{msg}, from {operation_id}."
+    return v
+
+
+def update_graph_id(cg, new_graph_id:str):
+    old_gc = cg.meta.graph_config._asdict()
+    old_gc["ID"] = new_graph_id
+    new_gc = GraphConfig(**old_gc)
+    new_meta = ChunkedGraphMeta(new_gc, cg.meta.data_source, cg.meta.custom_data)
+    cg.update_meta(new_meta, overwrite=True)

pychunkedgraph/graph/attributes.py

@@ -1,6 +1,9 @@
+# pylint: disable=invalid-name, missing-docstring, protected-access, raise-missing-from
+
 # TODO design to use these attributes across different clients
 # `family_id` is specific to bigtable
 
+from enum import Enum
 from typing import NamedTuple
 
 from .utils import serializers
@@ -101,20 +104,34 @@ class Connectivity:
         serializer=serializers.NumPyArray(dtype=basetypes.EDGE_AREA),
     )
 
-    CrossChunkEdge = _AttributeArray(
+    AtomicCrossChunkEdge = _AttributeArray(
         pattern=b"atomic_cross_edges_%d",
         family_id="3",
         serializer=serializers.NumPyArray(
             dtype=basetypes.NODE_ID, shape=(-1, 2), compression_level=22
         ),
     )
 
-    FakeEdges = _Attribute(
+    CrossChunkEdge = _AttributeArray(
+        pattern=b"cross_edges_%d",
+        family_id="4",
+        serializer=serializers.NumPyArray(
+            dtype=basetypes.NODE_ID, shape=(-1, 2), compression_level=22
+        ),
+    )
+
+    FakeEdgesCF3 = _Attribute(
         key=b"fake_edges",
         family_id="3",
         serializer=serializers.NumPyArray(dtype=basetypes.NODE_ID, shape=(-1, 2)),
     )
 
+    FakeEdges = _Attribute(
+        key=b"fake_edges",
+        family_id="4",
+        serializer=serializers.NumPyArray(dtype=basetypes.NODE_ID, shape=(-1, 2)),
+    )
+
 
 class Hierarchy:
     Child = _Attribute(
@@ -157,8 +174,6 @@ class GraphVersion:
 class OperationLogs:
     key = b"ioperations"
 
-    from enum import Enum
-
     class StatusCodes(Enum):
         SUCCESS = 0  # all is well, new changes persisted
         CREATED = 1  # log record created in storage