diff --git a/bench/benchmark_ldbc.py b/bench/benchmark_ldbc.py index 5b7fda1..7bc9433 100755 --- a/bench/benchmark_ldbc.py +++ b/bench/benchmark_ldbc.py @@ -32,8 +32,8 @@ def run_one( str(REPO_ROOT / "bench" / "run_one.py"), "--source-dir", str(source_dir), - "--output-db", - str(out_dir / f"{backend}.duckdb"), + "--output-dir", + str(out_dir / f"{backend}-csr"), "--backend", backend, *(["--memory-limit", memory_limit] if memory_limit else []), diff --git a/bench/compare_memgraph.py b/bench/compare_memgraph.py index 9050442..c3b38ec 100755 --- a/bench/compare_memgraph.py +++ b/bench/compare_memgraph.py @@ -60,8 +60,8 @@ def main() -> None: str(REPO_ROOT / "bench" / "run_one.py"), "--source-dir", args.source_dir, - "--output-db", - str(run_dir / "out.duckdb"), + "--output-dir", + str(run_dir / "out-csr"), "--backend", "pyarrow", *(["--memory-limit", args.memory_limit] if args.memory_limit else []), diff --git a/bench/run_one.py b/bench/run_one.py index c06ae22..9398697 100755 --- a/bench/run_one.py +++ b/bench/run_one.py @@ -29,7 +29,7 @@ def _max_rss_mib() -> float: def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--source-dir", required=True) - ap.add_argument("--output-db", required=True) + ap.add_argument("--output-dir", required=True) ap.add_argument( "--backend", required=True, choices=["pyarrow", "duckdb", "datafusion"] ) @@ -39,7 +39,7 @@ def main() -> None: start = time.perf_counter() results = convert_parquet_dir_to_csr( source_dir=args.source_dir, - output_db=args.output_db, + output_dir=args.output_dir, backend=args.backend, memory_limit=args.memory_limit, ) diff --git a/icebug_format/_convert_datafusion.py b/icebug_format/_convert_datafusion.py index b1a27a7..568f2a9 100644 --- a/icebug_format/_convert_datafusion.py +++ b/icebug_format/_convert_datafusion.py @@ -33,6 +33,7 @@ from icebug_format.convert_parquet import ( ICEBUG_DISK_VERSION, _write_icebug_parquet, + csr_rel_name, parse_size_to_bytes, resolve_rel_column_names, ) @@ -342,7 +343,9 @@ def convert_graph( f"FROM relations ORDER BY csr_source, csr_target" ) _stream_to_parquet( - ctx.sql(idx_sql), out_dir / f"indices_{name}.parquet", target_to_uint64=True + ctx.sql(idx_sql), + out_dir / f"indices_{csr_rel_name(name)}.parquet", + target_to_uint64=True, ) # indptr: degrees from a streaming GROUP BY, then histogram + prefix sum. @@ -357,5 +360,7 @@ def convert_graph( out_dir / f"nodes_{name}.parquet", ) _write_icebug_parquet( - indptr, out_dir / f"indptr_{name}.parquet", compression=_COMPRESSION + indptr, + out_dir / f"indptr_{csr_rel_name(name)}.parquet", + compression=_COMPRESSION, ) diff --git a/icebug_format/_convert_duckdb.py b/icebug_format/_convert_duckdb.py index ed9e024..dd49dfa 100644 --- a/icebug_format/_convert_duckdb.py +++ b/icebug_format/_convert_duckdb.py @@ -21,7 +21,7 @@ import pyarrow as pa import pyarrow.parquet as pq -from icebug_format.convert_parquet import resolve_rel_column_names +from icebug_format.convert_parquet import csr_rel_name, resolve_rel_column_names def _q(name: str) -> str: @@ -139,10 +139,10 @@ def convert_graph( con, "vertices", out_dir / f"nodes_{name}.parquet" ) _write_parquet_with_icebug_metadata( - con, "indices", out_dir / f"indices_{name}.parquet" + con, "indices", out_dir / f"indices_{csr_rel_name(name)}.parquet" ) _write_parquet_with_icebug_metadata( - con, "indptr", out_dir / f"indptr_{name}.parquet" + con, "indptr", out_dir / f"indptr_{csr_rel_name(name)}.parquet" ) finally: con.close() diff --git a/icebug_format/_convert_pyarrow.py b/icebug_format/_convert_pyarrow.py index 2c90f47..2b33c0b 100644 --- a/icebug_format/_convert_pyarrow.py +++ b/icebug_format/_convert_pyarrow.py @@ -30,6 +30,7 @@ from icebug_format.convert_parquet import ( _build_indptr, _write_icebug_parquet, + csr_rel_name, resolve_rel_column_names, ) @@ -344,5 +345,5 @@ def convert_graph( indices, indptr = _finalize_csr(rel, prop_cols, add_reverse_edges, n_nodes) _write_icebug_parquet(vtable, out_dir / f"nodes_{name}.parquet") - _write_icebug_parquet(indices, out_dir / f"indices_{name}.parquet") - _write_icebug_parquet(indptr, out_dir / f"indptr_{name}.parquet") + _write_icebug_parquet(indices, out_dir / f"indices_{csr_rel_name(name)}.parquet") + _write_icebug_parquet(indptr, out_dir / f"indptr_{csr_rel_name(name)}.parquet") diff --git a/icebug_format/cli.py b/icebug_format/cli.py index 04c9172..ed77f7b 100644 --- a/icebug_format/cli.py +++ b/icebug_format/cli.py @@ -186,6 +186,30 @@ def duckdb_type_to_cypher_type(duckdb_type: str) -> str: return type_map.get(base_type, "STRING") +def get_node_display_name(table_name: str) -> str: + """Derive the NODE table name used in schema.cypher from a DuckDB table.""" + if table_name == "nodes": + return "nodes" + elif table_name.startswith("nodes_"): + return table_name[6:].lower() # Remove "nodes_" prefix and lowercase + return table_name.lower() + + +def get_edge_display_name(table_name: str) -> str: + """Derive the REL table name used in schema.cypher from a DuckDB table. + + The ``_rel`` suffix keeps the REL table distinct from a same-named NODE + table. LadybugDB derives the ``indices_.parquet`` / + ``indptr_.parquet`` filenames from this name, so the parquet + export must use the same name. + """ + if table_name == "edges": + return "edges" + elif table_name.startswith("edges_"): + return table_name[6:].lower() # Remove "edges_" prefix and lowercase + return table_name.lower() + "_rel" + + def generate_schema_cypher( con, csr_table_name: str, @@ -212,22 +236,6 @@ def generate_schema_cypher( """ lines = [] - # Helper to derive display name from table name (lowercase) - # nodes => nodes, nodes_person => person, nodes_foo => foo - def get_node_display_name(table_name: str) -> str: - if table_name == "nodes": - return "nodes" - elif table_name.startswith("nodes_"): - return table_name[6:].lower() # Remove "nodes_" prefix and lowercase - return table_name.lower() - - def get_edge_display_name(table_name: str) -> str: - if table_name == "edges": - return "edges" - elif table_name.startswith("edges_"): - return table_name[6:].lower() # Remove "edges_" prefix and lowercase - return table_name.lower() - # Build mapping of original table names to display names node_display_names = {nt: get_node_display_name(nt) for nt in node_tables} @@ -352,20 +360,23 @@ def get_display_name(table_name: str, prefix: str) -> str: _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 + # Export edge tables: indices_.parquet, indptr_.parquet + # (rel_name == the REL table name emitted in schema.cypher, so LadybugDB can + # map the schema TABLE back to its parquet file) for edge_table in edge_tables: edge_name = ( edge_table[6:].lower() if edge_table.startswith("edges_") else edge_table.lower() ) + rel_name = get_edge_display_name(edge_table) indices_table = f"{csr_table_name}_indices_{edge_name}" - indices_file = parquet_dir / f"indices_{edge_name}.parquet" + indices_file = parquet_dir / f"indices_{rel_name}.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_name}" - indptr_file = parquet_dir / f"indptr_{edge_name}.parquet" + indptr_file = parquet_dir / f"indptr_{rel_name}.parquet" _write_parquet_with_icebug_metadata(con, indptr_table, indptr_file) print(f" Exported: {indptr_table} -> {indptr_file.name}") @@ -377,7 +388,7 @@ def get_display_name(table_name: str, prefix: str) -> str: edge_tables, parquet_dir, edge_relationships, - node_type_to_table + node_type_to_table, ) schema_file = parquet_dir / "schema.cypher" schema_file.write_text(schema_cypher) @@ -731,6 +742,12 @@ def main(): help="Conversion backend for --source-dir input " "(default: auto-detect an installed SQL engine, else pyarrow)", ) + parser.add_argument( + "--output-dir", + type=str, + help="Output directory for icebug-disk Parquet files (--source-dir only; " + "default: -csr)", + ) parser.add_argument( "--output-db", type=str, @@ -772,7 +789,7 @@ def main(): "--storage", type=str, default=None, - help="Storage path for schema.cypher (default: output_db path without .duckdb extension)", + help="Storage path recorded in schema.cypher (default: derived from the output path)", ) parser.add_argument( "--schema", @@ -807,15 +824,15 @@ def main(): print(f"Add reverse edges: {args.add_reverse_edges}") print(f"DuckDB/DataFusion memory limit: {args.memory_limit}") - output_db = args.output_db or str( - Path(args.source_dir).parent / f"{Path(args.source_dir).name}_csr.duckdb" + output_dir = args.output_dir or str( + Path(args.source_dir).parent / f"{Path(args.source_dir).name}-csr" ) - storage_path = args.storage or f"./{Path(output_db).stem}" - print(f"Output directory: {Path(output_db).parent / Path(output_db).stem}") + storage_path = args.storage or f"./{Path(output_dir).name}" + print(f"Output directory: {output_dir}") results = convert_parquet_dir_to_csr( source_dir=args.source_dir, - output_db=output_db, + output_dir=output_dir, backend=args.backend, add_reverse_edges=args.add_reverse_edges, memory_limit=args.memory_limit, diff --git a/icebug_format/convert_parquet.py b/icebug_format/convert_parquet.py index 021955b..be2e1de 100644 --- a/icebug_format/convert_parquet.py +++ b/icebug_format/convert_parquet.py @@ -8,8 +8,8 @@ as :mod:`icebug_format.cli` for DuckDB sources: - ``nodes_.parquet`` vertex table sorted by primary key -- ``indices_.parquet`` dense target id per edge, sorted by (source, target) -- ``indptr_.parquet`` CSR row pointers (``N + 1`` entries) +- ``indices__rel.parquet`` dense target id per edge, sorted by (source, target) +- ``indptr__rel.parquet`` CSR row pointers (``N + 1`` entries) - ``schema.cypher`` LadybugDB mount schema Three interchangeable backends are provided: @@ -258,6 +258,18 @@ def _arrow_type_to_cypher(t: pa.DataType) -> str: return "STRING" +def csr_rel_name(graph_name: str) -> str: + """Return the REL table name for a homogeneous CSR graph. + + LadybugDB derives the parquet filenames from the schema table names, so the + REL table (``_rel``) must match the ``indices__rel.parquet`` / + ``indptr__rel.parquet`` files written by the backends. The suffix + keeps the REL table distinct from the NODE table, whose name is the graph + name itself. + """ + return f"{graph_name}_rel" + + def _generate_schema_cypher(graph_name: str, output_dir: Path, storage: str) -> str: """ Build schema.cypher for one graph from the parquet files just written. @@ -265,7 +277,8 @@ def _generate_schema_cypher(graph_name: str, output_dir: Path, storage: str) -> The graph is homogeneous: both edge endpoints use the single vertex table. """ node_pq = output_dir / f"nodes_{graph_name}.parquet" - indices_pq = output_dir / f"indices_{graph_name}.parquet" + rel_name = csr_rel_name(graph_name) + indices_pq = output_dir / f"indices_{rel_name}.parquet" node_schema = pq.ParquetFile(node_pq).schema_arrow indices_schema = pq.ParquetFile(indices_pq).schema_arrow @@ -284,7 +297,7 @@ def _generate_schema_cypher(graph_name: str, output_dir: Path, storage: str) -> ] props_str = (", " + ", ".join(props)) if props else "" lines.append( - f"CREATE REL TABLE {graph_name}(FROM {graph_name} TO {graph_name}{props_str}) " + f"CREATE REL TABLE {rel_name}(FROM {graph_name} TO {graph_name}{props_str}) " f"WITH (storage = '{storage}', format = 'icebug-disk');" ) return "\n".join(lines) + "\n" @@ -354,7 +367,7 @@ def _select_backend(backend: str, memory_limit: str | None = None) -> Callable: def convert_parquet_dir_to_csr( source_dir: str | Path, - output_db: str | Path | None = None, + output_dir: str | Path | None = None, graph_name: str | None = None, backend: str = "auto", add_reverse_edges: bool = False, @@ -367,9 +380,9 @@ def convert_parquet_dir_to_csr( Args: source_dir: Directory containing ``-v.parquet``/``-e.parquet`` pairs (and/or ``vertex.parquet``/``edge.parquet``). - output_db: Output database path used only as a naming convention; the - Parquet files are written to a sibling directory named after its - stem. Defaults to ``/_csr.duckdb``. + output_dir: Output directory for the icebug-disk Parquet files. When + multiple graphs are discovered each one is written to a subdirectory + ``/``. Defaults to ``-csr``. graph_name: Optional filter; when omitted, every discovered graph is converted. backend: One of ``auto``, ``pyarrow``, ``duckdb``, ``datafusion``. @@ -398,18 +411,17 @@ def convert_parquet_dir_to_csr( f"{', '.join(g['name'] for g in graphs)}" ) - output_db = ( - Path(output_db) - if output_db - else Path(source_dir).parent / f"{Path(source_dir).name}_csr.duckdb" + output_dir = ( + Path(output_dir) + if output_dir + else Path(source_dir).parent / f"{Path(source_dir).name}-csr" ) - output_base = Path(output_db).parent / Path(output_db).stem if len(graphs) == 1: - out_dirs = [output_base] + out_dirs = [output_dir] else: - out_dirs = [output_base / g["name"] for g in graphs] + out_dirs = [output_dir / g["name"] for g in graphs] - storage_path = storage if storage is not None else f"./{Path(output_db).stem}" + storage_path = storage if storage is not None else f"./{output_dir.name}" convert = _select_backend(backend, memory_limit) results: list[dict] = [] diff --git a/tests/test_parquet_convert.py b/tests/test_parquet_convert.py index 8836188..0d67cad 100644 --- a/tests/test_parquet_convert.py +++ b/tests/test_parquet_convert.py @@ -33,8 +33,8 @@ def _write_graph(dir_: Path, name: str, vertex_ids, edges, prop=None): def _read_csr(out_dir: Path, name: str): - indices = pq.read_table(out_dir / f"indices_{name}.parquet") - indptr = pq.read_table(out_dir / f"indptr_{name}.parquet") + indices = pq.read_table(out_dir / f"indices_{name}_rel.parquet") + indptr = pq.read_table(out_dir / f"indptr_{name}_rel.parquet") nodes = pq.read_table(out_dir / f"nodes_{name}.parquet") return nodes, indices, indptr @@ -51,7 +51,7 @@ def test_directed_sparse_ids(backend): ) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend=backend + src, output_dir=Path(tmp) / "out", backend=backend ) assert [r["name"] for r in res] == ["g"] out_dir = Path(res[0]["output_dir"]) @@ -73,7 +73,7 @@ def test_dense_contiguous_ids(backend): _write_graph(src, "g", [2, 0, 1], [(0, 1), (2, 0), (1, 1)]) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend=backend + src, output_dir=Path(tmp) / "out", backend=backend ) out_dir = Path(res[0]["output_dir"]) _, indices, indptr = _read_csr(out_dir, "g") @@ -90,7 +90,7 @@ def test_reverse_edges(backend): res = convert_parquet_dir_to_csr( src, - output_db=Path(tmp) / "out.duckdb", + output_dir=Path(tmp) / "out", backend=backend, add_reverse_edges=True, ) @@ -110,7 +110,7 @@ def test_self_loops_appear_once_with_reverse_edges(backend): res = convert_parquet_dir_to_csr( src, - output_db=Path(tmp) / "out.duckdb", + output_dir=Path(tmp) / "out", backend=backend, add_reverse_edges=True, ) @@ -127,7 +127,7 @@ def test_self_loops_preserved_directed(backend): _write_graph(src, "g", [0, 1], [(0, 0), (0, 1)]) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend=backend + src, output_dir=Path(tmp) / "out", backend=backend ) out_dir = Path(res[0]["output_dir"]) _, indices, _ = _read_csr(out_dir, "g") @@ -142,7 +142,7 @@ def test_edge_properties_preserved(backend): _write_graph(src, "g", [0, 1], [(0, 1)], prop=("weight", [2.5])) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend=backend + src, output_dir=Path(tmp) / "out", backend=backend ) out_dir = Path(res[0]["output_dir"]) _, indices, _ = _read_csr(out_dir, "g") @@ -158,7 +158,7 @@ def test_empty_edges(backend): _write_graph(src, "g", [0, 1, 2], []) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend=backend + src, output_dir=Path(tmp) / "out", backend=backend ) out_dir = Path(res[0]["output_dir"]) _, indices, indptr = _read_csr(out_dir, "g") @@ -174,11 +174,15 @@ def test_icebug_disk_metadata_written(backend): _write_graph(src, "g", [0, 1], [(0, 1)]) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend=backend + src, output_dir=Path(tmp) / "out", backend=backend ) out_dir = Path(res[0]["output_dir"]) - for f in ("nodes_g.parquet", "indices_g.parquet", "indptr_g.parquet"): + for f in ( + "nodes_g.parquet", + "indices_g_rel.parquet", + "indptr_g_rel.parquet", + ): meta = pq.ParquetFile(out_dir / f).metadata.metadata or {} assert meta.get(b"icebug_disk_version") == b"v1" @@ -190,16 +194,36 @@ def test_schema_cypher_generated(backend): _write_graph(src, "g", [0, 1], [(0, 1)]) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend=backend + src, output_dir=Path(tmp) / "out", backend=backend ) out_dir = Path(res[0]["output_dir"]) schema = (out_dir / "schema.cypher").read_text() assert "CREATE NODE TABLE g(id INT64, PRIMARY KEY(id))" in schema - assert "CREATE REL TABLE g(FROM g TO g)" in schema + assert "CREATE REL TABLE g_rel(FROM g TO g)" in schema assert "icebug-disk" in schema +def test_default_output_dir_csr_suffix_and_schema_names(): + """Default output dir is -csr; rel table name differs from node.""" + with tempfile.TemporaryDirectory() as tmp: + src = Path(tmp) / "graph500-24" + src.mkdir() + _write_graph(src, "graph500-24", [0, 1], [(0, 1)]) + + res = convert_parquet_dir_to_csr(src, backend="pyarrow") + assert Path(res[0]["output_dir"]) == Path(tmp) / "graph500-24-csr" + + schema = (Path(res[0]["output_dir"]) / "schema.cypher").read_text() + assert "CREATE NODE TABLE graph500_24(id INT64, PRIMARY KEY(id))" in schema + assert ( + "CREATE REL TABLE graph500_24_rel(FROM graph500_24 TO graph500_24)" + in schema + ) + # NODE and REL table names must not clash + assert "CREATE REL TABLE graph500_24(" not in schema + + @pytest.mark.parametrize("backend", ["duckdb", "datafusion"]) def test_memory_limit_accepted(backend): """SQL backends honor memory_limit; output is unchanged.""" @@ -209,7 +233,7 @@ def test_memory_limit_accepted(backend): res = convert_parquet_dir_to_csr( src, - output_db=Path(tmp) / "out.duckdb", + output_dir=Path(tmp) / "out", backend=backend, memory_limit="128MB", ) @@ -303,7 +327,7 @@ def test_graph_name_filter(): _write_graph(src, "two", [0, 1], [(1, 0)]) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", graph_name="two", backend="pyarrow" + src, output_dir=Path(tmp) / "out", graph_name="two", backend="pyarrow" ) assert [r["name"] for r in res] == ["two"] @@ -315,7 +339,7 @@ def test_multi_graph_output_dirs(): _write_graph(src, "two", [0, 1], [(1, 0)]) res = convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "out.duckdb", backend="pyarrow" + src, output_dir=Path(tmp) / "out", backend="pyarrow" ) assert {r["name"] for r in res} == {"one", "two"} dirs = {Path(r["output_dir"]).name for r in res} @@ -337,16 +361,14 @@ def test_backends_agree(backend): _write_graph(src, "g", ids, edges) convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "base.duckdb", backend="pyarrow" - ) - convert_parquet_dir_to_csr( - src, output_db=Path(tmp) / "other.duckdb", backend=backend + src, output_dir=Path(tmp) / "base", backend="pyarrow" ) + convert_parquet_dir_to_csr(src, output_dir=Path(tmp) / "other", backend=backend) - base = pq.read_table(Path(tmp) / "base" / "indices_g.parquet") - base_ptr = pq.read_table(Path(tmp) / "base" / "indptr_g.parquet") - other = pq.read_table(Path(tmp) / "other" / "indices_g.parquet") - other_ptr = pq.read_table(Path(tmp) / "other" / "indptr_g.parquet") + base = pq.read_table(Path(tmp) / "base" / "indices_g_rel.parquet") + base_ptr = pq.read_table(Path(tmp) / "base" / "indptr_g_rel.parquet") + other = pq.read_table(Path(tmp) / "other" / "indices_g_rel.parquet") + other_ptr = pq.read_table(Path(tmp) / "other" / "indptr_g_rel.parquet") assert other.equals(base) assert other_ptr.equals(base_ptr)