diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8c8d3a9..6644723 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/karate/karate_random.duckdb b/examples/karate/duckdb/karate_random.duckdb similarity index 100% rename from karate/karate_random.duckdb rename to examples/karate/duckdb/karate_random.duckdb diff --git a/examples/karate/duckdb/schema.cypher b/examples/karate/duckdb/schema.cypher new file mode 100644 index 0000000..9106b42 --- /dev/null +++ b/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); diff --git a/examples/karate/icebug-disk/indices_edges.parquet b/examples/karate/icebug-disk/indices_edges.parquet new file mode 100644 index 0000000..fb2d43d Binary files /dev/null and b/examples/karate/icebug-disk/indices_edges.parquet differ diff --git a/examples/karate/icebug-disk/indptr_edges.parquet b/examples/karate/icebug-disk/indptr_edges.parquet new file mode 100644 index 0000000..6a9fa81 Binary files /dev/null and b/examples/karate/icebug-disk/indptr_edges.parquet differ diff --git a/examples/karate/icebug-disk/karate_csr.duckdb b/examples/karate/icebug-disk/karate_csr.duckdb new file mode 100644 index 0000000..0da29f6 Binary files /dev/null and b/examples/karate/icebug-disk/karate_csr.duckdb differ diff --git a/examples/karate/icebug-disk/nodes_nodes.parquet b/examples/karate/icebug-disk/nodes_nodes.parquet new file mode 100644 index 0000000..9050ecd Binary files /dev/null and b/examples/karate/icebug-disk/nodes_nodes.parquet differ diff --git a/examples/karate/icebug-disk/schema.cypher b/examples/karate/icebug-disk/schema.cypher new file mode 100644 index 0000000..44fb5cb --- /dev/null +++ b/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'); diff --git a/icebug_format/cli.py b/icebug_format/cli.py index b166814..256687c 100644 --- a/icebug_format/cli.py +++ b/icebug_format/cli.py @@ -11,7 +11,7 @@ 2. Handles sparse node IDs by creating a dense mapping (original_id -> csr_index) 3. Converts edges to CSR (Compressed Sparse Row) format 4. Pre-sorts edges by source using DuckDB for memory efficiency -5. Saves CSR data and node mapping to DuckDB for reuse +5. Saves CSR data to DuckDB for reuse 6. Exports to parquet format and generates schema.cypher for ladybugdb Key Features: @@ -22,10 +22,10 @@ Usage Examples: # Convert edges in karate_random.duckdb to CSR format and save to csr_graph.db - python convert_csr.py --source-db karate_random.duckdb --output-db csr_graph.db + python icebug-format.py --source-db karate_random.duckdb --output-db csr_graph.db # Convert with limited data for testing - python convert_csr.py --source-db karate_random.duckdb --test --limit 50000 --output-db test.db + python icebug-format.py --source-db karate_random.duckdb --test --limit 50000 --output-db test.db """ import argparse @@ -34,6 +34,18 @@ from pathlib import Path import duckdb +import pyarrow.parquet as pq + +ICEBUG_DISK_VERSION = "v1" + + +def _write_parquet_with_icebug_metadata(con, table_name: str, output_path: Path) -> None: + """Export a DuckDB table to parquet with icebug_disk_version metadata.""" + arrow_table = con.execute(f"SELECT * FROM {table_name}").arrow().read_all() + existing_metadata = arrow_table.schema.metadata or {} + new_metadata = {**existing_metadata, b"icebug_disk_version": ICEBUG_DISK_VERSION.encode()} + arrow_table = arrow_table.replace_schema_metadata(new_metadata) + pq.write_table(arrow_table, str(output_path)) def parse_schema_cypher(schema_path: Path) -> dict: @@ -227,7 +239,7 @@ def get_edge_display_name(table_name: str) -> str: display_name = node_display_names[node_table] lines.append( f"CREATE NODE TABLE {display_name}({cols_str}, PRIMARY KEY({pk_col})) " - f"WITH (storage = '{storage_path}');" + f"WITH (storage = '{storage_path}', format = 'icebug-disk');" ) except Exception as e: print( @@ -271,7 +283,7 @@ def get_edge_display_name(table_name: str) -> str: props_str = ", ".join(col_defs) lines.append( f"CREATE REL TABLE {rel_name}(FROM {src_table} TO {dst_table}" - f"{', ' + props_str if props_str else ''}) WITH (storage = '{storage_path}');" + 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}") @@ -311,17 +323,38 @@ def export_to_parquet_and_cypher( # Compute storage path if not provided if storage_path is None: - storage_path = f"./{output_path.stem}/{csr_table_name}" + storage_path = f"./{output_path.stem}" + + # Helper to get node display name from original table name + def get_display_name(table_name: str, prefix: str) -> str: + if table_name == prefix: + return prefix + if table_name.startswith(f"{prefix}_"): + return table_name[len(prefix) + 1:].lower() + return table_name.lower() - # Get all tables to export - result = con.execute("SHOW TABLES").fetchall() - all_tables = [row[0] for row in result] + # Export node tables: nodes_.parquet + for node_table in node_tables: + display_name = get_display_name(node_table, "nodes") + csr_node_table = f"{csr_table_name}_{node_table}" + parquet_file = parquet_dir / f"nodes_{display_name}.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_.parquet, indptr_.parquet + for edge_table in edge_tables: + edge_name = ( + edge_table[6:].lower() if edge_table.startswith("edges_") else edge_table.lower() + ) + indices_table = f"{csr_table_name}_indices_{edge_name}" + indices_file = parquet_dir / f"indices_{edge_name}.parquet" + _write_parquet_with_icebug_metadata(con, indices_table, indices_file) + print(f" Exported: {indices_table} -> {indices_file.name}") - # Export each table to parquet (lowercase filenames) - for table_name in all_tables: - parquet_file = parquet_dir / f"{table_name.lower()}.parquet" - con.execute(f"COPY {table_name} TO '{parquet_file}' (FORMAT 'parquet')") - print(f" Exported: {table_name} -> {parquet_file.name}") + indptr_table = f"{csr_table_name}_indptr_{edge_name}" + indptr_file = parquet_dir / f"indptr_{edge_name}.parquet" + _write_parquet_with_icebug_metadata(con, indptr_table, indptr_file) + print(f" Exported: {indptr_table} -> {indptr_file.name}") # Generate schema.cypher schema_cypher = generate_schema_cypher( @@ -431,33 +464,20 @@ def create_csr_graph_to_duckdb( print(f"Node type to table mapping: {node_type_to_table}") - # Copy all node tables with proper prefixing and create per-table mappings + # Copy all node tables with proper prefixing node_counts = {} # Track node counts per table + node_pk_cols = {} # pk column name per node table for nt in node_tables: try: - # Get the primary key column (first column of original node table) cols = con.execute(f"DESCRIBE orig.{nt}").fetchall() pk_col = cols[0][0] if cols else "id" + node_pk_cols[nt] = pk_col con.execute( f"CREATE TABLE {csr_table_name}_{nt} AS SELECT * FROM orig.{nt} ORDER BY {pk_col};" ) print(f" Copied node table: {nt} -> {csr_table_name}_{nt}") - # Create per-table node mapping - node_type = nt[6:].lower() if nt.startswith("nodes_") else nt.lower() - mapping_table = f"{csr_table_name}_mapping_{node_type}" - con.execute(f""" - CREATE TABLE {mapping_table} AS - SELECT - row_number() OVER (ORDER BY {pk_col}) - 1 AS csr_index, - {pk_col} AS original_node_id - FROM {csr_table_name}_{nt} - ORDER BY csr_index; - """) - print(f" Created node mapping: {mapping_table}") - - # Track node count result = con.execute( f"SELECT COUNT(*) FROM {csr_table_name}_{nt}" ).fetchone() @@ -469,77 +489,72 @@ def create_csr_graph_to_duckdb( print("\nStep 1: Building per-edge-table CSR structures...") for et in edge_tables: - # Determine source and target node types from schema - edge_name = ( - et[6:].lower() if et.startswith("edges_") else et.lower() - ) # Remove "edges_" prefix and lowercase - src_node_type, dst_node_type = edge_relationships.get( - edge_name, (None, None) - ) + edge_name = et[6:].lower() if et.startswith("edges_") else et.lower() + src_node_type, dst_node_type = edge_relationships.get(edge_name, (None, None)) - # Find the corresponding node tables src_table = node_type_to_table.get(src_node_type) dst_table = node_type_to_table.get(dst_node_type) - fallback_node_type = None if src_table and dst_table: - src_mapping = f"{csr_table_name}_mapping_{src_node_type}" - dst_mapping = f"{csr_table_name}_mapping_{dst_node_type}" num_src_nodes = node_counts.get(src_table, 0) - print( - f"\n Processing {et}: {src_node_type} ({num_src_nodes} nodes) -> {dst_node_type}" - ) + print(f"\n Processing {et}: {src_node_type} ({num_src_nodes} nodes) -> {dst_node_type}") else: - # Fallback: use first node table for both - fallback_table = node_tables[0] if node_tables else "nodes" - fallback_node_type = ( - fallback_table[6:].lower() - if fallback_table.startswith("nodes_") - else fallback_table.lower() - ) - src_mapping = f"{csr_table_name}_mapping_{fallback_node_type}" - dst_mapping = src_mapping - num_src_nodes = node_counts.get(fallback_table, 0) - print(f"\n Processing {et}: using fallback mapping {src_mapping}") + src_table = dst_table = node_tables[0] if node_tables else "nodes" + num_src_nodes = node_counts.get(src_table, 0) + print(f"\n Processing {et}: using fallback node table {src_table}") + + src_pk = node_pk_cols.get(src_table, "id") + dst_pk = node_pk_cols.get(dst_table, "id") + src_csr_table = f"{csr_table_name}_{src_table}" + dst_csr_table = f"{csr_table_name}_{dst_table}" + + # Inline id→csr_index mapping as CTEs — no separate mapping tables needed + map_cte = f""" + src_map AS ( + SELECT row_number() OVER (ORDER BY {src_pk}) - 1 AS csr_index, + {src_pk} AS original_node_id + FROM {src_csr_table} + ), + dst_map AS ( + SELECT row_number() OVER (ORDER BY {dst_pk}) - 1 AS csr_index, + {dst_pk} AS original_node_id + FROM {dst_csr_table} + )""" # Get edge columns excluding source and target edge_cols_result = con.execute(f"DESCRIBE orig.{et}").fetchall() edge_col_names = [col[0] for col in edge_cols_result] edge_cols = [c for c in edge_col_names if c not in ["source", "target"]] - # Prepare select column strings select_cols = "m1.csr_index AS csr_source, m2.csr_index AS csr_target" if edge_cols: select_cols += ", " + ", ".join([f"e.{c}" for c in edge_cols]) - reverse_select_cols = ( - "m2.csr_index AS csr_source, m1.csr_index AS csr_target" - ) + reverse_select_cols = "m2.csr_index AS csr_source, m1.csr_index AS csr_target" if edge_cols: reverse_select_cols += ", " + ", ".join([f"e.{c}" for c in edge_cols]) reverse_cols = "csr_target AS csr_source, csr_source AS csr_target" if edge_cols: reverse_cols += ", " + ", ".join(edge_cols) - # Create relations table for this edge type + join_clause = f""" + FROM orig.{et} e + JOIN src_map m1 ON e.source = m1.original_node_id + JOIN dst_map m2 ON e.target = m2.original_node_id + WHERE e.source != e.target""" + if limit_rels: limit_per_table = limit_rels // len(edge_tables) if directed: rel_query = f""" - SELECT {select_cols} - FROM orig.{et} e - JOIN {src_mapping} m1 ON e.source = m1.original_node_id - JOIN {dst_mapping} m2 ON e.target = m2.original_node_id - WHERE e.source != e.target + WITH {map_cte} + SELECT {select_cols} {join_clause} LIMIT {limit_per_table} """ else: rel_query = f""" - WITH limited AS ( - SELECT {select_cols} - FROM orig.{et} e - JOIN {src_mapping} m1 ON e.source = m1.original_node_id - JOIN {dst_mapping} m2 ON e.target = m2.original_node_id - WHERE e.source != e.target + WITH {map_cte}, + limited AS ( + SELECT {select_cols} {join_clause} LIMIT {limit_per_table} ) SELECT * FROM limited @@ -549,25 +564,15 @@ def create_csr_graph_to_duckdb( else: if directed: rel_query = f""" - SELECT {select_cols} - FROM orig.{et} e - JOIN {src_mapping} m1 ON e.source = m1.original_node_id - JOIN {dst_mapping} m2 ON e.target = m2.original_node_id - WHERE e.source != e.target + WITH {map_cte} + SELECT {select_cols} {join_clause} """ else: rel_query = f""" - SELECT {select_cols} - FROM orig.{et} e - JOIN {src_mapping} m1 ON e.source = m1.original_node_id - JOIN {dst_mapping} m2 ON e.target = m2.original_node_id - WHERE e.source != e.target + WITH {map_cte} + SELECT {select_cols} {join_clause} UNION ALL - SELECT {reverse_select_cols} - FROM orig.{et} e - JOIN {src_mapping} m1 ON e.source = m1.original_node_id - JOIN {dst_mapping} m2 ON e.target = m2.original_node_id - WHERE e.source != e.target + SELECT {reverse_select_cols} {join_clause} """ con.execute(f"CREATE TABLE relations_{edge_name} AS {rel_query};") @@ -640,27 +645,6 @@ def create_csr_graph_to_duckdb( ).fetchone() total_edges += result[0] if result else 0 - # Create global metadata - con.execute(f""" - CREATE TABLE {csr_table_name}_metadata AS - SELECT {total_nodes} AS n_nodes, {total_edges} AS n_edges, {directed} AS directed - """) - - # List per-table node mappings for output - node_mapping_tables = [ - f"{csr_table_name}_mapping_{nt[6:].lower() if nt.startswith('nodes_') else nt.lower()}" - for nt in node_tables - ] - - print("\n✅ CSR format built and cleaned up. Final tables:") - for mapping_table in node_mapping_tables: - print(f" - {mapping_table} (orig_id → mapped_id)") - for i, et in enumerate(edge_tables): - edge_name = et[6:].lower() if et.startswith("edges_") else et.lower() - print(f" - {csr_table_name}_indptr_{edge_name}") - print(f" - {csr_table_name}_indices_{edge_name}") - print(f" - {csr_table_name}_metadata (global)") - print( f"\n✓ Built CSR format: {total_nodes} nodes, {total_edges} edges across {len(edge_tables)} edge types" ) @@ -695,20 +679,18 @@ def main(): parser.add_argument( "--source-db", type=str, - default="karate_random.duckdb", - help="Source DuckDB database path (default: karate_random.duckdb)", + required=True, + help="Source DuckDB database path", ) parser.add_argument( "--output-db", type=str, - default="csr_graph.db", - help="Output DuckDB database path (default: csr_graph.db)", + help="Output DuckDB database path", ) parser.add_argument( "--csr-table", type=str, - default="csr_graph", - help="Table name prefix for CSR data (default: csr_graph)", + help="Table name prefix for CSR data", ) parser.add_argument( "--node-table", @@ -763,6 +745,13 @@ def main(): args = parser.parse_args() + # Infer --output-db and --csr-table from --source-db stem when not provided + source_stem = Path(args.source_db).stem + if args.output_db is None: + args.output_db = str(Path(args.source_db).parent / f"{source_stem}_csr.duckdb") + if args.csr_table is None: + args.csr_table = source_stem + if args.graphar: print("=== GraphAr to CSR Format Converter ===\n") print(f"GraphAr directory: {args.graphar}") @@ -813,8 +802,7 @@ def main(): # Compute default storage path from output_db if not specified storage_path = args.storage if storage_path is None: - # Use output_db path without .duckdb extension + csr_table_name - storage_path = f"./{Path(args.output_db).stem}/{args.csr_table}" + storage_path = f"./{Path(args.output_db).stem}" print(f"Storage path: {storage_path}") if args.node_table: diff --git a/icebug_format/graphar.py b/icebug_format/graphar.py index 42e177a..33316df 100644 --- a/icebug_format/graphar.py +++ b/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_.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_.parquet, indptr_.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" diff --git a/icebug_format/test_csr_duckdb.py b/icebug_format/test_csr_duckdb.py new file mode 100644 index 0000000..41b5e52 --- /dev/null +++ b/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() diff --git a/karate/karate_csr.duckdb b/karate/karate_csr.duckdb deleted file mode 100644 index 7b66127..0000000 Binary files a/karate/karate_csr.duckdb and /dev/null differ diff --git a/karate/karate_csr/karate_indices_edges.parquet b/karate/karate_csr/karate_indices_edges.parquet deleted file mode 100644 index d310cce..0000000 Binary files a/karate/karate_csr/karate_indices_edges.parquet and /dev/null differ diff --git a/karate/karate_csr/karate_indptr_edges.parquet b/karate/karate_csr/karate_indptr_edges.parquet deleted file mode 100644 index 407aae1..0000000 Binary files a/karate/karate_csr/karate_indptr_edges.parquet and /dev/null differ diff --git a/karate/karate_csr/karate_mapping_nodes.parquet b/karate/karate_csr/karate_mapping_nodes.parquet deleted file mode 100644 index 2e2be19..0000000 Binary files a/karate/karate_csr/karate_mapping_nodes.parquet and /dev/null differ diff --git a/karate/karate_csr/karate_metadata.parquet b/karate/karate_csr/karate_metadata.parquet deleted file mode 100644 index 41d0e5b..0000000 Binary files a/karate/karate_csr/karate_metadata.parquet and /dev/null differ diff --git a/karate/karate_csr/karate_nodes.parquet b/karate/karate_csr/karate_nodes.parquet deleted file mode 100644 index 9574539..0000000 Binary files a/karate/karate_csr/karate_nodes.parquet and /dev/null differ diff --git a/karate/karate_csr/schema.cypher b/karate/karate_csr/schema.cypher deleted file mode 100644 index 124aaaf..0000000 --- a/karate/karate_csr/schema.cypher +++ /dev/null @@ -1,2 +0,0 @@ -CREATE NODE TABLE nodes(id INT64, club STRING, PRIMARY KEY(id)) WITH (storage = './karate_csr/karate'); -CREATE REL TABLE edges(FROM nodes TO nodes) WITH (storage = './karate_csr/karate'); diff --git a/pyproject.toml b/pyproject.toml index 1cbc177..d9c6661 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,13 +6,10 @@ readme = "README.md" requires-python = ">=3.13" dependencies = [ "duckdb>=1.3.2", + "pyarrow>=21.0.0", ] [project.optional-dependencies] -full = [ - "ladybug>=0.17.0", - "pyarrow>=21.0.0", -] graphar = [ "graphar", ] @@ -36,9 +33,8 @@ testpaths = ["tests"] [dependency-groups] dev = [ - "pytest>=8", + "pytest>=7.2", ] [tool.uv] package = true -dev-dependencies = [] diff --git a/scan.py b/scan.py deleted file mode 100644 index 90c2e8c..0000000 --- a/scan.py +++ /dev/null @@ -1,218 +0,0 @@ -#!/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 --prefix demo -""" - -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_graph_std(input_dir: Path, prefix: str, schema_path: Path | None = None): - """ - Scan the graph data in icebug-disk format from parquet files and print metadata, nodes, and edges. - """ - con = duckdb.connect() # In-memory connection - - try: - # Use provided prefix - metadata_parquet = input_dir / f"{prefix}_metadata.parquet" - - if not metadata_parquet.exists(): - print(f"Metadata parquet {metadata_parquet} not found") - return - - # Get metadata - metadata = con.execute(f"SELECT * FROM '{metadata_parquet}'").fetchone() - if metadata: - n_nodes, n_edges, directed = metadata - print(f"Metadata: {n_nodes} nodes, {n_edges} edges, directed={directed}") - else: - print("No metadata found") - return - - # Node tables - node_parquets = list(input_dir.glob(f"{prefix}_nodes*.parquet")) - print("\nNode Tables:") - for np in node_parquets: - nt = np.stem # remove .parquet - print(f"\nTable: {nt}") - rows = con.execute(f"SELECT * FROM '{np}'").fetchall() - for row in rows: - print(row) - - # Verify CSR index == node offset invariant - print("\nVerifying CSR index == node offset invariant:") - for np in node_parquets: - nt = np.stem - # Extract node type: prefix_nodes[_type] - if nt.startswith(f"{prefix}_nodes"): - suffix = nt[len(f"{prefix}_nodes") :] - if suffix == "": - node_type = "nodes" - else: - node_type = suffix[1:] # remove leading _ - mapping_p = input_dir / f"{prefix}_mapping_{node_type}.parquet" - if mapping_p.exists(): - # Load node rows - node_rows = con.execute(f"SELECT * FROM '{np}'").fetchall() - # Load mapping ordered by csr_index - mapping = con.execute( - f"SELECT original_node_id FROM '{mapping_p}' ORDER BY csr_index" - ).fetchall() - mapping_ids = [row[0] for row in mapping] - # Assume pk is first column - pk_values = [row[0] for row in node_rows] - if len(pk_values) != len(mapping_ids): - print( - f" {nt}: Length mismatch: {len(pk_values)} vs {len(mapping_ids)}" - ) - continue - violations = [ - i - for i in range(len(pk_values)) - if pk_values[i] != mapping_ids[i] - ] - if violations: - print( - f" {nt}: Invariant violated at rows {violations[:5]}{'...' if len(violations) > 5 else ''}" - ) - else: - print(f" {nt}: Invariant holds ✓") - else: - print(f" {nt}: No mapping file found") - - # Edge tables - reconstruct from CSR - print("\nEdge Tables (reconstructed from CSR):") - - # Parse schema for edge relationships - edge_relationships = {} - if schema_path: - edge_relationships = parse_schema_cypher(schema_path) - - indptr_parquets = list(input_dir.glob(f"{prefix}_indptr_*.parquet")) - for indptr_p in indptr_parquets: - indptr_t = indptr_p.stem - edge_name = indptr_t[len(f"{prefix}_indptr_") :] - indices_p = input_dir / f"{prefix}_indices_{edge_name}.parquet" - - if not indices_p.exists(): - print(f"\nSkipping {edge_name}: indices parquet not found") - continue - - # Get source and target node types - 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") - continue - - source_mapping_p = input_dir / f"{prefix}_mapping_{from_node}.parquet" - target_mapping_p = input_dir / f"{prefix}_mapping_{to_node}.parquet" - - if not source_mapping_p.exists(): - print(f"\nSkipping {edge_name}: source mapping parquet not found") - continue - if not target_mapping_p.exists(): - print(f"\nSkipping {edge_name}: target mapping parquet not found") - continue - - print(f"\nTable: {edge_name} (FROM {from_node} TO {to_node})") - - # Fetch data - 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() - source_map = [ - row[0] - for row in con.execute( - f"SELECT original_node_id FROM '{source_mapping_p}' ORDER BY csr_index" - ).fetchall() - ] - target_map = [ - row[0] - for row in con.execute( - f"SELECT original_node_id FROM '{target_mapping_p}' ORDER BY csr_index" - ).fetchall() - ] - - # Reconstruct edges - for i in range(len(indptr) - 1): - start = indptr[i] - end = indptr[i + 1] - source_orig = source_map[i] - for j in range(start, end): - row = indices_result[j] - target_csr = row[0] # target is first column - target_orig = target_map[target_csr] - # Print source, target, and any additional properties - 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" - ) - parser.add_argument("--prefix", help="Table prefix (default: input)") - - args = parser.parse_args() - - input_dir = Path(args.input) - if not input_dir.is_dir(): - print(f"Directory {input_dir} not found") - return - - prefix = args.prefix if args.prefix else args.input - - schema_path = input_dir / "schema.cypher" - if not schema_path.exists(): - schema_path = None - - scan_graph_std(input_dir, prefix, schema_path) - - -if __name__ == "__main__": - main()