update to icebug-disk v1

.github/workflows/ci.yml

@@ -7,7 +7,7 @@ on:
 
 jobs:
   test:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
 
     steps:
       - uses: actions/checkout@v4
@@ -18,9 +18,9 @@ jobs:
 
       - name: Install dependencies
         run: |
-          uv pip install --system duckdb "pyarrow>=21"
-          uv pip install --system --no-deps -e .
-          uv pip install --system pytest
+          uv venv
+          uv sync
+          uv pip install -e .
 
       - name: Run tests
-        run: pytest
+        run: uv run --no-sync pytest

examples/karate/duckdb/schema.cypher

@@ -0,0 +1,2 @@
+CREATE NODE TABLE nodes(id INT64, club STRING, PRIMARY KEY(id));
+CREATE REL TABLE edges(FROM nodes TO nodes);

examples/karate/icebug-disk/schema.cypher

@@ -0,0 +1,2 @@
+CREATE NODE TABLE nodes(id INT64, club STRING, PRIMARY KEY(id)) WITH (storage = '', format = 'icebug-disk');
+CREATE REL TABLE edges(FROM nodes TO nodes) WITH (storage = '', format = 'icebug-disk');

icebug_format/graphar.py

@@ -9,9 +9,8 @@
 The conversion process:
 1. Reads vertex data from GraphAr parquet files
 2. Reads edge adjacency lists and properties from GraphAr parquet files
-3. Creates a mapping from original vertex IDs to contiguous indices
-4. Converts edges to CSR (Compressed Sparse Row) format
-5. Saves to DuckDB and exports to parquet format with schema.cypher
+3. Converts edges to CSR (Compressed Sparse Row) format
+4. Saves to DuckDB and exports to parquet format with schema.cypher
 
 Usage Examples:
     # Convert graphar graph to icebug-format
@@ -24,7 +23,7 @@
 import duckdb
 import pyarrow.parquet as pq
 
-from icebug_format.cli import set_memory_limit
+from icebug_format.cli import ICEBUG_DISK_VERSION, _write_parquet_with_icebug_metadata, set_memory_limit
 
 
 def duckdb_type_to_cypher_type(duckdb_type: str) -> str:
@@ -96,7 +95,7 @@ def read_graphar_vertices(vertex_info, graph_path: Path) -> dict:
     return vertices
 
 
-def read_graphar_edges(edge_info, graph_path: Path, vertex_indices: dict) -> dict:
+def read_graphar_edges(edge_info, graph_path: Path) -> dict:
     """
     Read edges from GraphAr parquet files.
 
@@ -199,7 +198,6 @@ def convert_graphar_to_graph_std(
     # Read vertices
     print("\nStep 1: Reading vertices...")
     vertex_type_to_table = {}
-    vertex_indices = {}
 
     for i in range(graph_info.vertex_info_num()):
         vertex_info = graph_info.get_vertex_info_by_index(i)
@@ -260,20 +258,7 @@ def convert_graphar_to_graph_std(
 
             print(f"  Created {node_table_name} with {len(vertex_rows)} vertices")
 
-            # Create mapping table
-            pk_col = prop_group_names[0]
-            mapping_table_name = f"{csr_table_name}_mapping_{vertex_type}"
-            con.execute(f"""
-                CREATE TABLE {mapping_table_name} AS
-                SELECT row_number() OVER (ORDER BY "{pk_col}") - 1 AS csr_index,
-                       "{pk_col}" AS original_node_id
-                FROM {node_table_name}
-                ORDER BY csr_index
-            """)
-            print(f"  Created {mapping_table_name}")
-
             vertex_type_to_table[vertex_type] = node_table_name
-            vertex_indices[vertex_type] = mapping_table_name
         else:
             print(f"  Warning: No vertices found for type {vertex_type}")
 
@@ -288,21 +273,16 @@ def convert_graphar_to_graph_std(
 
         print(f"\n  Processing edge {edge_type}: {src_type} -> {dst_type}")
 
-        # Get source and destination mapping tables
-        src_mapping = vertex_indices.get(src_type)
-        dst_mapping = vertex_indices.get(dst_type)
+        # Get source and destination node tables
+        src_table = vertex_type_to_table.get(src_type)
+        dst_table = vertex_type_to_table.get(dst_type)
 
-        if not src_mapping or not dst_mapping:
-            print(f"    Warning: Missing mapping tables for {src_type} -> {dst_type}")
+        if not src_table or not dst_table:
+            print(f"    Warning: Missing node tables for {src_type} -> {dst_type}")
             continue
 
         # Get vertex counts
-        src_table = vertex_type_to_table.get(src_type)
-        num_src_nodes = (
-            con.execute(f"SELECT COUNT(*) FROM {src_table}").fetchone()[0]
-            if src_table
-            else 0
-        )
+        num_src_nodes = con.execute(f"SELECT COUNT(*) FROM {src_table}").fetchone()[0]
         print(f"    Source nodes: {num_src_nodes}")
 
         # Read edge data
@@ -465,12 +445,6 @@ def convert_graphar_to_graph_std(
         if result:
             total_edges += result[0]
 
-    # Create global metadata
-    con.execute(f"""
-        CREATE TABLE {csr_table_name}_metadata AS
-        SELECT {total_nodes}::BIGINT AS n_nodes, {total_edges}::BIGINT AS n_edges, {directed}::BOOLEAN AS directed
-    """)
-
     print(f"\n✅ Conversion complete: {total_nodes} nodes, {total_edges} edges")
 
     # Export to parquet and generate schema.cypher
@@ -482,24 +456,35 @@ def convert_graphar_to_graph_std(
 
     print(f"Parquet output directory: {parquet_dir}")
 
-    # Get all tables
-    result = con.execute("SHOW TABLES").fetchall()
-    all_tables = [row[0] for row in result]
+    # Compute storage path (points to the parquet directory)
+    storage_path = f"./{parquet_dir.name}"
+
+    # Export node tables: nodes_<vertex_type>.parquet
+    for vertex_type, src_table in vertex_type_to_table.items():
+        csr_node_table = f"{csr_table_name}_{src_table}"
+        parquet_file = parquet_dir / f"nodes_{vertex_type}.parquet"
+        _write_parquet_with_icebug_metadata(con, csr_node_table, parquet_file)
+        print(f"  Exported: {csr_node_table} -> {parquet_file.name}")
+
+    # Export edge tables: indices_<edge_type>.parquet, indptr_<edge_type>.parquet
+    for i in range(graph_info.edge_info_num()):
+        edge_info_item = graph_info.get_edge_info_by_index(i)
+        edge_type = edge_info_item.get_edge_type()
 
-    # Export each table
-    for table_name in all_tables:
-        parquet_file = parquet_dir / f"{table_name}.parquet"
-        con.execute(f"COPY {table_name} TO '{parquet_file}' (FORMAT 'parquet')")
-        print(f"  Exported: {table_name} -> {parquet_file.name}")
+        indices_table = f"{csr_table_name}_indices_{edge_type}"
+        indices_file = parquet_dir / f"indices_{edge_type}.parquet"
+        _write_parquet_with_icebug_metadata(con, indices_table, indices_file)
+        print(f"  Exported: {indices_table} -> {indices_file.name}")
+
+        indptr_table = f"{csr_table_name}_indptr_{edge_type}"
+        indptr_file = parquet_dir / f"indptr_{edge_type}.parquet"
+        _write_parquet_with_icebug_metadata(con, indptr_table, indptr_file)
+        print(f"  Exported: {indptr_table} -> {indptr_file.name}")
 
     # Generate schema.cypher
     schema_lines = []
 
-    # Compute storage path
-    storage_path = f"./{parquet_dir.name}/{csr_table_name}"
-
     # Generate NODE TABLE definitions
-    schema_lines = []
     for vertex_type, src_table in vertex_type_to_table.items():
         table_name = f"{csr_table_name}_{src_table}"
         try:
@@ -515,7 +500,8 @@ def convert_graphar_to_graph_std(
 
             cols_str = ", ".join(col_defs)
             schema_lines.append(
-                f"CREATE NODE TABLE {vertex_type}({cols_str}, PRIMARY KEY({pk_col})) WITH (storage = '{storage_path}');"
+                f"CREATE NODE TABLE {vertex_type}({cols_str}, PRIMARY KEY({pk_col})) "
+                f"WITH (storage = '{storage_path}', format = 'icebug-disk');"
             )
         except Exception as e:
             print(
@@ -524,14 +510,10 @@ def convert_graphar_to_graph_std(
 
     # Generate REL TABLE definitions
     for i in range(graph_info.edge_info_num()):
-        edge_info = graph_info.get_edge_info_by_index(i)
-        edge_type = edge_info.get_edge_type()
-        src_type = edge_info.get_src_type()
-        dst_type = edge_info.get_dst_type()
-        rel_name = edge_type
-
-        src_table = vertex_type_to_table.get(src_type, f"nodes_{src_type}")
-        vertex_type_to_table.get(dst_type, f"nodes_{dst_type}")
+        edge_info_item = graph_info.get_edge_info_by_index(i)
+        edge_type = edge_info_item.get_edge_type()
+        src_type = edge_info_item.get_src_type()
+        dst_type = edge_info_item.get_dst_type()
 
         indices_table = f"{csr_table_name}_indices_{edge_type}"
         try:
@@ -546,11 +528,11 @@ def convert_graphar_to_graph_std(
 
             props_str = ", ".join(col_defs)
             schema_lines.append(
-                f"CREATE REL TABLE {rel_name}(FROM {src_type} TO {dst_type}"
-                f"{', ' + props_str if props_str else ''}) WITH (storage = '{storage_path}');"
+                f"CREATE REL TABLE {edge_type}(FROM {src_type} TO {dst_type}"
+                f"{', ' + props_str if props_str else ''}) WITH (storage = '{storage_path}', format = 'icebug-disk');"
             )
         except Exception as e:
-            print(f"Warning: Could not generate schema for rel table {rel_name}: {e}")
+            print(f"Warning: Could not generate schema for rel table {edge_type}: {e}")
 
     schema_cypher = "\n".join(schema_lines) + "\n"
     schema_file = parquet_dir / "schema.cypher"

icebug_format/test_csr_duckdb.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+"""
+Script to scan graph data in icebug-disk format from parquet files and print metadata, node tables, and reconstructed edge tables.
+
+Usage:
+    uv run scan.py --input demo-db_csr
+"""
+
+import argparse
+import re
+from pathlib import Path
+
+import duckdb
+
+
+def parse_schema_cypher(schema_path: Path) -> dict:
+    """
+    Parse schema.cypher to extract edge relationships (FROM/TO node types).
+
+    Returns:
+        Dictionary mapping edge names to (from_node_type, to_node_type) tuples
+    """
+    edge_relationships = {}
+
+    if not schema_path.exists():
+        return edge_relationships
+
+    content = schema_path.read_text()
+
+    # Parse REL TABLE definitions: CREATE REL TABLE Follows(FROM User TO User, ...);
+    # Also handles backtick-quoted identifiers: CREATE REL TABLE `edges` (FROM `nodes` TO `nodes`, ...)
+    rel_pattern = (
+        r"CREATE\s+REL\s+TABLE\s+`?(\w+)`?\s*\(\s*FROM\s+`?(\w+)`?\s+TO\s+`?(\w+)`?"
+    )
+    for match in re.finditer(rel_pattern, content, re.IGNORECASE):
+        edge_name = match.group(1).lower()
+        from_node = match.group(2).lower()
+        to_node = match.group(3).lower()
+        edge_relationships[edge_name] = (from_node, to_node)
+
+    return edge_relationships
+
+
+def scan_icebug_disk(input_dir: Path, schema_path: Path | None = None):
+    """
+    Scan the graph data in icebug-disk format from parquet files and print nodes and edges.
+    """
+    con = duckdb.connect()  # In-memory connection
+
+    try:
+        # Node tables: nodes_*.parquet
+        node_parquets = sorted(input_dir.glob("nodes_*.parquet"))
+        print("Node Tables:")
+        for np in node_parquets:
+            nt = np.stem  # e.g. "nodes_city"
+            print(f"\nTable: {nt}")
+
+            # verify metadata
+            metadataQuery = f"""
+                SELECT CAST(value AS VARCHAR) AS metadata_value
+                FROM parquet_kv_metadata('{np}')
+                WHERE key = 'icebug_disk_version'
+            """
+            metadata = con.execute(metadataQuery).fetchone()
+
+            if not metadata or metadata[0].lower() != "v1":
+                print(f"Warning: {np} has missing or incompatible icebug_disk_version metadata")
+
+            rows = con.execute(f"SELECT * FROM '{np}'").fetchall()
+            for row in rows:
+                print(row)
+
+        # Edge tables - reconstruct from CSR using node table row order as ID map
+        print("\nEdge Tables (reconstructed from CSR):")
+
+        # Build node-type -> pk-values-by-row-order map from nodes_*.parquet
+        node_id_map: dict[str, list] = {}
+        for np in node_parquets:
+            node_type = np.stem[len("nodes_"):]  # e.g. "city"
+            rows = con.execute(f"SELECT * FROM '{np}'").fetchall()
+            node_id_map[node_type] = [row[0] for row in rows]
+
+        # Parse schema for edge FROM/TO relationships
+        edge_relationships = {}
+        if schema_path:
+            edge_relationships = parse_schema_cypher(schema_path)
+
+        for indptr_p in sorted(input_dir.glob("indptr_*.parquet")):
+            edge_name = indptr_p.stem[len("indptr_"):]
+            indices_p = input_dir / f"indices_{edge_name}.parquet"
+
+            if not indices_p.exists():
+                print(f"\nSkipping {edge_name}: indices parquet not found")
+                continue
+
+            from_node, to_node = edge_relationships.get(edge_name, (None, None))
+            if not from_node or not to_node:
+                print(f"\nSkipping {edge_name}: no relationship info in schema.cypher")
+                continue
+
+            source_ids = node_id_map.get(from_node)
+            target_ids = node_id_map.get(to_node)
+
+            if source_ids is None:
+                print(f"\nSkipping {edge_name}: no node table for '{from_node}'")
+                continue
+            if target_ids is None:
+                print(f"\nSkipping {edge_name}: no node table for '{to_node}'")
+                continue
+
+            print(f"\nTable: {edge_name} (FROM {from_node} TO {to_node})")
+
+            indptr = [row[0] for row in con.execute(f"SELECT ptr FROM '{indptr_p}'").fetchall()]
+            indices_result = con.execute(f"SELECT * FROM '{indices_p}'").fetchall()
+
+            for i in range(len(indptr) - 1):
+                start = indptr[i]
+                end = indptr[i + 1]
+                source_orig = source_ids[i]
+                for j in range(start, end):
+                    row = indices_result[j]
+                    target_orig = target_ids[row[0]]
+                    edge_data = [source_orig, target_orig]
+                    if len(row) > 1:
+                        edge_data.extend(row[1:])
+                    print(tuple(edge_data))
+
+    finally:
+        con.close()
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Scan CSR graph data from parquet files"
+    )
+    parser.add_argument(
+        "--input", required=True, help="Input directory containing parquet files"
+    )
+
+    args = parser.parse_args()
+
+    input_dir = Path(args.input)
+    if not input_dir.is_dir():
+        print(f"Directory {input_dir} not found")
+        return
+
+    schema_path = input_dir / "schema.cypher"
+    if not schema_path.exists():
+        schema_path = None
+
+    scan_icebug_disk(input_dir, schema_path)
+
+
+if __name__ == "__main__":
+    main()