r2g — Relational-to-Graph for ArangoDB

Purpose. This repository is a working reference application for relational-to-graph mapping with ArangoDB: it implements the patterns, trade-offs, and techniques for projecting relational and structured data into a graph model, and serves as the reference consumer of the relational-schema-analyzer core. That core (RSA) is actively adopted by other projects in ArangoDB Solutions (e.g. the contextual-data-fabric building blocks) — so it is an active dependency, not a throwaway demo.

Project relational and structured data sources as a graph in ArangoDB — materialize via batch ETL, sync via CDC/Kafka, or query interactively through the mapping studio. Or federate: emit CSI v1 + R2RML mappings so a query engine reaches your data where it lives — across PostgreSQL, ClickHouse, and ArangoDB, joined on declared cross-source keys — with no data movement.

r2g turns relational schemas into ArangoDB graph schemas mechanically: tables become document collections, foreign keys become edges, join tables become edges, and types are coerced from PostgreSQL representations into proper JSON types. PostgreSQL, MySQL / MariaDB, SQL Server, Snowflake, and ClickHouse are supported as relational sources today (plus CSV directories and Kafka topics); the connector layer is designed for additional structured and semi-structured sources over time. Beyond loading data into ArangoDB, r2g can also emit federation mappings (CSI v1 + R2RML) so a query engine reaches your data where it lives — see Federation.

Status — actively used; the production core is RSA. r2g is the reference application for relational-to-graph migration with ArangoDB, and its reusable introspection/analysis core has been extracted into relational-schema-analyzer (RSA). This core is actively adopted by other projects in ArangoDB Solutions — the contextual-data-fabric building blocks depend on it — so RSA is the production-grade dependency and carries the production bar. Downstream systems depend on RSA (and named, tested r2g modules), not on r2g as a whole. r2g itself is well-tested (CI: ruff + mypy + a large unit suite + Dockerized integration); its operational envelope is still single-node with no scale/HA guarantees — tracked in Known limitations.

See docs/PRD.md for the full product requirements document and roadmap.

Concepts

Relational databases model relationships implicitly through foreign keys and resolve them at query time via joins. Graph databases model relationships explicitly as first-class edges, enabling direct traversal without joins.

The R2G pipeline applies a mechanical mapping:

• Each table becomes an ArangoDB document collection (vertices). The table's primary key becomes the document _key.
• Each foreign key becomes an edge collection. For every row in the source table, an edge is created from the source vertex to the target vertex, using the FK value to resolve the _to endpoint.
• Join tables (many-to-many) become edges rather than vertices -- the two FK columns point to the two vertex collections the edge connects.
• Data types are coerced from PostgreSQL representations to proper JSON types: integers, floats, booleans, nested JSON for jsonb columns, and arrays.

flowchart LR
    PG[(PostgreSQL)] -->|ingest-schema| SJ[schema.json]
    SJ -->|generate-config| MY[mapping.yaml]
    PG -->|"dump-tables"| CSV[".csv per table"]
    CSV --> CSVPATH["generate-csv-import<br/>(file-based)"]
    MY --> CSVPATH
    SJ --> CSVPATH
    CSVPATH --> SH1["import_csv.sh"]
    SH1 -->|"arangoimport --type csv"| ADB[(ArangoDB)]
    PG -->|"stream (direct)"| STREAM["streaming pipeline<br/>server-side cursors"]
    MY --> STREAM
    SJ --> STREAM
    STREAM -->|"HTTP API bulk import"| ADB

Federation: query your data where it lives (CSI v1 + R2RML)

Beyond moving data into ArangoDB, r2g can emit the mappings that let a query engine reach your data where it already lives — unmoved, across several engines at once. This is r2g's role as the forward mapping producer for the Contextual Data Fabric federated-query engine.

Two exports, both load-independent (no migration required):

• export-csi emits a CSI v1 (Conceptual Schema Interchange) document pairing the conceptual model — CC-12 OWL naming: singular PascalCase entities, lowerCamel properties — with its ArangoDB physical mapping. It drives the fabric's Arango leg (CSI → MappingBundle → arango-sparql-py).
• export-r2rml emits R2RML for the relational leg (R2RML → Ontop), so a Virtual Knowledge Graph engine can answer SPARQL over the live relational database with no data movement. Both legs default to the same concept namespace, so they share one vocabulary.

Cross-source join keys (P6.7). Two sources often share a business key — say account_id in a ClickHouse analytics table and in a PostgreSQL CRM Account. r2g proposes these as shared keys (confirm-to-accept, never invented) and emits them as conceptualModel.joinKeys in the CSI export: declared cross-source join keys the federated executor bind-joins on, with no materialized edge.

The payoff is a single conceptual query spanning PostgreSQL (via Ontop) ⋈ ClickHouse (via a native executor) ⋈ ArangoDB (via arango-sparql-py), joined on a declared key and grounded with citations — three engines, no data movement.

r2g export-csi   --config mapping.yaml --schema schema.json --output analytics.csi.json
r2g export-r2rml --config mapping.yaml --schema schema.json --output analytics.r2rml.ttl

Federation is read-only by construction and reuses r2g's mapping model, so the same introspect → shape → validate workflow yields both a migration and a federation mapping from one source.

Prerequisites

• Python 3.10+
• PostgreSQL with data you want to migrate (any version with pg_catalog support)
• ArangoDB instance (tested with 3.11+) with arangoimport on your PATH
• psql or another tool to export CSV dumps from PostgreSQL

Features

• Schema introspection -- connects to PostgreSQL and extracts tables, columns, primary keys, and foreign keys
• Mechanical mapping -- tables become document collections, foreign keys become edge collections, join tables become edges
• Type coercion -- PostgreSQL types (integer, boolean, jsonb, arrays, etc.) are converted to proper JSON types
• YAML-driven configuration -- auto-generate a default mapping or hand-tune collection names, field renames, include/exclude lists
• Polars-powered file processing -- CSV/TSV/GZ dump files processed via Polars for high throughput
• arangoimport script generation -- produces executable bash scripts that load documents first, then edges, with configurable connection parameters
• Studio batch-script viewer -- previews the current draft mapping as a secret-safe JSONL or CSV-direct arangoimport bundle, or as one focused vertex/edge command, with copy and client-side download; this is an alternative artifact, while Studio Load continues to stream directly over HTTP
• Named graph creation -- generates arangosh JavaScript to create ArangoDB named graph definitions from edge mappings
• Structured logging -- human-readable dev output or JSON for production via structlog
• CSV-direct import -- generate arangoimport --type csv scripts that import PG CSV dumps directly with --translate for key remapping, --datatype for type coercion, and collection prefixes for edge _from/_to construction; no intermediate JSONL needed
• Mapping visualizer -- interactive HTML visualization (D3.js force-directed graph) of the relational-to-graph mapping with four views: graph schema, relational schema cards, edge mapping detail, and a mapping editor with YAML export
• Direct PG streaming -- stream data directly from PostgreSQL to ArangoDB via the HTTP API with server-side cursors, configurable batch sizes, and REPEATABLE READ snapshot isolation -- no intermediate files; supports parallel streaming with --workers
• Dry-run mode -- stream --dry-run validates connectivity to both PostgreSQL and ArangoDB, reads and transforms all data, but skips writes and graph creation -- reports row counts and sample documents per collection for pre-flight validation
• Progress bars and throughput -- Rich progress bars during streaming with real-time row counts; elapsed time and rows/second throughput displayed on completion
• Retry with backoff -- transient ArangoDB write failures (connection errors, server overload) are retried with exponential backoff
• Collection management -- --drop-collections flag drops and recreates target collections before import for idempotent re-runs
• Table filtering -- --include-tables and --exclude-tables on the stream command for selective import of large schemas
• Import error reporting -- document-level errors from ArangoDB bulk imports are captured, logged, and displayed in the summary table instead of silently dropped
• Comprehensive type mapping -- 50+ PostgreSQL types explicitly mapped to JSON types: integer variants, float variants, boolean, JSON/JSONB, UUID, timestamps, intervals, network types, geometric types, and text search types
• Schema diff -- diff-schema command compares two schema snapshots and reports added/removed tables, column type changes, nullable changes, primary key changes, and foreign key changes; supports --json output for scripting
• Config migration -- migrate-config command auto-updates a mapping YAML when the PostgreSQL schema evolves: adds collections for new tables, adds edges for new FKs, removes edges for dropped FKs, flags orphaned collections, and cleans stale field references and type overrides -- all while preserving user customizations (renames, field mappings, include/exclude lists)
• Data validation -- validate-data command checks referential integrity of dump files before import: builds PK lookup sets per table and verifies every FK value references an existing PK; reports orphaned references that would produce broken edges
• Topological import ordering -- document collections are imported in dependency order (FK targets before FK sources) so that referenced vertices exist before edges are created; circular FK dependencies are detected and warned
• Environment variable support -- connection parameters (PG_CONN, ARANGO_ENDPOINT, ARANGO_DB, ARANGO_USER, ARANGO_PASSWORD) can be set via environment variables or a .env file; CLI flags override env vars when both are provided
• Skip existing -- stream --skip-existing skips collections that already contain data, enabling resumption of partial streaming runs without re-importing completed collections
• Incremental streaming -- stream --since 2026-04-01T00:00:00 filters rows by a timestamp column (auto-detects updated_at/created_at or use --since-column); combine with --on-duplicate=replace for basic incremental updates
• PK-less table safety -- tables without a primary key are warned during validation and streaming; documents receive auto-generated keys and edges referencing such tables are flagged
• CDC (Change Data Capture) -- near real-time PostgreSQL→ArangoDB sync via logical replication. PGReplicationListener manages replication slots, polls pg_logical_slot_get_changes, parses output via test_decoding (built-in) or wal2json plugins. DeltaTransformer converts row-level changes to graph mutations (document upserts/deletes + edge recalculation). CDCHandler orchestrates event processing with transaction grouping and stats tracking. Configurable conflict resolution policies: source_wins (default), last_write_wins, log_and_skip, fail. CLI commands: cdc-setup, cdc-teardown, cdc-status, cdc-start
• Composite foreign key support -- multi-column foreign keys are correctly introspected from pg_catalog, represented in mappings, and transformed into composite _key / _from / _to values using a configurable separator
• Multi-schema support -- --pg-schema option on ingest-schema, dump-tables, and stream commands allows introspection and import from any PostgreSQL schema, not just public
• Automated table dumping -- dump-tables command connects to PostgreSQL and exports each table as a CSV file in one pass
• Join table auto-detection -- generate-config heuristically identifies junction tables (exactly 2 FKs, no non-structural data columns) and flags them as join tables

Project structure

src/r2g/
├── main.py                     # Typer CLI (top-level commands + `source`/`project`/`secrets` groups)
├── config_migrate.py           # Config migration when schema evolves
├── data_validator.py           # Referential integrity checker for dump data
├── schema_diff.py              # Schema comparison / structural diff
├── topo_sort.py                # Topological sort for import ordering, cycle detection
├── cdc/                        # Change Data Capture (Phase 3)
│   ├── conflict.py             # ConflictPolicy, ConflictResolver, ConflictLog
│   ├── models.py               # ChangeEvent, ArangoDelta, TransactionBatch
│   ├── parsers.py              # Output plugin parsers (test_decoding, wal2json)
│   ├── pg_listener.py          # PGReplicationListener: slot mgmt, polling loop
│   ├── delta_transformer.py    # Convert CDC events → ArangoDB mutations
│   ├── handler.py              # CDCHandler: orchestrate event processing with stats
│   ├── kafka_parser.py         # Debezium and flat JSON message parsers
│   └── kafka_consumer.py        # Kafka consumer with confluent-kafka
├── types.py                    # Pydantic models (Schema, Table, MappingConfig, EdgeDefinition, ...)
├── config.py                   # ConfigManager, YAML load/save, PG→JSON type map, join detection
├── snapshot.py                 # Shared introspection + reviewed-overlay snapshot pipeline
├── log.py                      # structlog setup
├── connectors/
│   ├── base.py                 # SourceConnector registry/factory
│   ├── session.py              # SourceSession bulk-read protocol
│   ├── postgres.py             # PostgreSQL connector/session via psycopg
│   ├── mysql.py                # MySQL/MariaDB connector/session
│   ├── mssql.py                # SQL Server connector/session
│   ├── snowflake.py            # Snowflake connector/session + key-pair auth
│   ├── clickhouse.py           # ClickHouse connector/session
│   └── arango_writer.py        # ArangoDB HTTP API writer via python-arango
├── input/
│   └── dump_reader.py          # Polars-based CSV/TSV/GZ reader
├── transformers/
│   ├── node_transformer.py     # Row → ArangoDB document (with type coercion)
│   ├── edge_transformer.py     # Row → ArangoDB edge (FK and join-table modes)
│   └── converter.py            # Re-exports NodeTransformer, EdgeTransformer
├── generators/
│   ├── arangoimport.py         # Bash script generator (JSONL and CSV-direct)
│   └── visualizer.py           # Interactive HTML mapping visualizer + editor
└── streaming/
    └── pipeline.py             # SourceSession → ArangoDB direct streaming pipeline

Installation

r2g is a CLI plus an optional FastAPI mapping studio. It ships from PyPI as r2g-arango and is designed around opt-in extras so you only pull in the connectors / UI you actually need.

End users — install with pipx

We recommend pipx so the CLI lives in its own isolated venv and does not pollute your project environments:

pipx install 'r2g-arango[postgres,ui]'   # PG source + interactive mapping studio
r2g --help
r2g ui                                   # opens http://localhost:8501

Plain pip install works too; just be aware that the connector dependency trees (psycopg, snowflake-connector-python, confluent-kafka) are substantial and you probably don't want them in an unrelated app venv.

Extras matrix

Pick the extras that match your use case. Combine with commas.

Common recipes:

pipx install 'r2g-arango[postgres,ui]'              # mapping studio against PG
pipx install 'r2g-arango[mysql,ui]'                 # mapping studio against MySQL/MariaDB
pipx install 'r2g-arango[sqlserver,ui]'             # mapping studio against SQL Server
pipx install 'r2g-arango[postgres,kafka]'           # batch load + Kafka CDC worker
pipx install 'r2g-arango[snowflake,ui]'             # Snowflake source via the studio
pipx install 'r2g-arango[clickhouse,ui]'            # ClickHouse source via the studio
pipx install 'r2g-arango[postgres,llm]'             # PG + AI ontology suggestions (Phase 10)
pipx install 'r2g-arango[postgres,ontology]'        # PG + deterministic ontology engine (Phase 10)
pip   install 'r2g-arango[all]'                     # kitchen sink

Contributors — editable install

git clone https://github.com/ArthurKeen/r2g-arango
cd r2g-arango
pip install -e ".[all,test,dev]"   # all extras + pytest + ruff
ruff check src/ tests/
pytest tests/ -m "not integration"

Quick start

0. Configure credentials (optional)

Instead of passing connection strings on the command line, create a .env file (see .env.example):

cp .env.example .env
# Edit .env with your PostgreSQL and ArangoDB credentials

The CLI auto-loads .env from the working directory. All connection flags (--conn, --pg-conn, --endpoint, --database, --username, --password) can be set via environment variables (PG_CONN, ARANGO_ENDPOINT, ARANGO_DB, ARANGO_USER, ARANGO_PASSWORD). CLI flags override env vars.

Snowflake accepts password URLs and Contextual Data Fabric-style key-pair URLs:

snowflake://user:password@account/DATABASE/SCHEMA?warehouse=WH&role=R
snowflake://$SNOWFLAKE_USER:@$SNOWFLAKE_ACCOUNT/DATABASE/SCHEMA?warehouse=$SNOWFLAKE_WAREHOUSE&role=$SNOWFLAKE_ROLE&private_key_file=$SNOWFLAKE_PRIVATE_KEY_FILE&private_key_file_pwd=$SNOWFLAKE_PRIVATE_KEY_FILE_PWD

Password and key-pair credentials are mutually exclusive. Key-pair connections use SNOWFLAKE_JWT; missing/unreadable key files fail before the driver is called, and credentials are scrubbed from API errors. The private key must exist on the machine or container running r2g at the path resolved from SNOWFLAKE_PRIVATE_KEY_FILE; an environment reference does not upload or copy the key.

Snowflake quick start

Install the connector, register the source, and capture its schema:

pip install 'r2g-arango[snowflake,ui]'
r2g source add \
  --name customer360 \
  --type snowflake \
  --conn 'snowflake://$SNOWFLAKE_USER:@$SNOWFLAKE_ACCOUNT/DATABASE/SCHEMA?warehouse=$SNOWFLAKE_WAREHOUSE&role=$SNOWFLAKE_ROLE&private_key_file=$SNOWFLAKE_PRIVATE_KEY_FILE' \
  --key-overlay ./keys.overlay.json
r2g source snapshot customer360 --pg-schema SCHEMA

--key-overlay is optional, but recommended for Snowflake schemas whose PK/FK constraints are absent or informational. It accepts a reviewed RSA JSON/YAML overlay; r2g stores the parsed overlay and records its source and fingerprint on each snapshot. r2g never invents those keys. Name-based FK suggestions work for Snowflake, but value-overlap sampling is not yet implemented for Snowflake and falls back to name-only evidence.

Export CSV files or stream directly from the registered source:

r2g source dump customer360 --pg-schema SCHEMA --output-dir ./dumps
r2g stream --source customer360 --pg-schema SCHEMA \
  --schema schema.json --config mapping.yaml \
  --database customer360_graph

In Studio, Load uses the direct streaming path from Snowflake to ArangoDB. View arangoimport script bundle only renders an alternative, secret-safe batch artifact; it does not execute the script or create its input files. The bundled live demo and role/setup instructions are in examples/snowflake_customer_360/README.md.

1. Extract schema from PostgreSQL

r2g ingest-schema --conn "postgresql://user:pass@localhost/mydb" --output schema.json

2. Generate a default mapping config

r2g generate-config --schema schema.json --output mapping.yaml

This creates a YAML file with one document collection per table and one edge collection per foreign key. Edit it to rename collections, exclude fields, or mark join tables.

3. Dump tables to CSV

Use the built-in dump-tables command to export all tables at once:

r2g dump-tables --conn "postgresql://user:pass@localhost/mydb" --output-dir ./dumps

Or use psql manually (one file per table, filename must match the table name):

for table in users orders products; do
  psql -d mydb -c "COPY ${table} TO STDOUT WITH CSV HEADER" > dumps/${table}.csv
done

3a. Visualize the mapping (optional)

r2g visualize-mapping --schema schema.json --config mapping.yaml --output mapping.html

Opens an interactive HTML report in your browser showing the PG-to-graph mapping: draggable graph schema, table cards with PK/FK badges, and edge mapping details.

4. Generate CSV-direct import script (preferred)

r2g generate-csv-import \
  --schema schema.json \
  --config mapping.yaml \
  --data-dir ./dumps \
  --output import_csv.sh \
  --endpoint http://localhost:8529 \
  --database mydb \
  --graph-name my_graph

4a. Alternative: JSONL transform path

Transform an entire directory of CSV dumps in one pass:

r2g transform-all \
  --schema schema.json \
  --config mapping.yaml \
  --input-dir ./dumps \
  --output-dir ./output \
  --file-pattern "*.csv"

Or transform a single table's nodes or edges:

r2g transform-nodes --schema schema.json --config mapping.yaml --table users --input dumps/users.csv --output output/users.jsonl
r2g transform-edges --schema schema.json --config mapping.yaml --table orders --input dumps/orders.csv --output output/orders_edges.jsonl

Then generate the arangoimport script:

r2g generate-import \
  --config mapping.yaml \
  --data-dir ./output \
  --output import.sh \
  --endpoint http://localhost:8529 \
  --database mydb \
  --graph-name my_graph

This produces an executable import.sh (documents first, then edges) and an arangosh graph creation script.

The Studio exposes the same renderer under Actions → View arangoimport script bundle and on target collection/edge context menus. The preview defaults to ./output/*.jsonl or ./dumps/*.csv; the artifact directory is editable and shell-quoted in every generated command. It reads ARANGO_PASSWORD only at runtime and never executes or writes the displayed script on the server.

5. Load into ArangoDB

CSV-direct path (step 4):

./import_csv.sh

JSONL path (step 4a):

./import.sh

Override connection details via environment variables (works with the generated scripts):

ARANGO_ENDPOINT=http://prod:8529 ARANGO_DB=prod_db ARANGO_PASSWORD=secret ./import_csv.sh

Pre-flight: validate data integrity (optional)

r2g validate-data --schema schema.json --config mapping.yaml --data-dir ./dumps

Checks that every FK value in your dump files references an existing PK in the target table. Reports orphaned references that would produce broken edges in ArangoDB.

Schema evolution: migrating the mapping config

When your PostgreSQL schema changes (new tables, dropped columns, added/removed FKs), update the mapping config automatically:

# Re-extract the updated schema
r2g ingest-schema --conn "postgresql://user:pass@localhost/mydb" --output schema_v2.json

# Migrate the existing config to match
r2g migrate-config --schema schema_v2.json --config mapping.yaml

This preserves all your customizations (collection renames, field mappings, include/exclude lists, type overrides) while adapting to schema changes. Use --output new_mapping.yaml to write to a different file, or --json-report for machine-readable output.

Alternative: Direct streaming (no intermediate files)

Skip steps 3-5 entirely and stream data directly from PostgreSQL to ArangoDB:

r2g stream \
  --pg-conn "postgresql://user:pass@localhost/mydb" \
  --schema schema.json \
  --config mapping.yaml \
  --endpoint http://localhost:8529 \
  --database mydb \
  --batch-size 10000 \
  --graph-name my_graph

This uses server-side cursors with REPEATABLE READ isolation for consistent snapshots and bulk-imports via the ArangoDB HTTP API.

Options:

• --workers 4 -- parallel streaming with per-worker PG + ArangoDB connections. Note: each worker opens its own REPEATABLE READ transaction; this provides per-worker consistency but not a single global snapshot across all workers. For strict point-in-time consistency, use --workers 1 (default).
• --on-duplicate replace -- ArangoDB duplicate handling strategy (replace, update, ignore, error)
• --include-tables users,orders -- only stream specified tables (and their edges)
• --exclude-tables audit_log -- skip specified tables
• --drop-collections -- drop and recreate target collections before import
• --skip-existing -- skip collections that already have data (for resuming partial runs)

Add --dry-run to preview row counts and sample documents without writing to ArangoDB:

r2g stream --dry-run \
  --pg-conn "postgresql://user:pass@localhost/mydb" \
  --schema schema.json \
  --config mapping.yaml \
  --endpoint http://localhost:8529 \
  --database mydb

CLI reference

r2g source — catalog source management

r2g project — project management

r2g entitlements — governance (Phase 9)

The load API also accepts emit_governance / tier_layout to drop the artifact set on a governed load.

r2g secrets — catalog credential encryption

r2g catalog — external data catalog discovery (Phase 8a)

Connect to an external enterprise data catalog (OpenMetadata today) and use it to discover and import migration sources ("discover-then-connect"). Credentials are not read from the catalog — imported sources use $ENV_VAR placeholders you resolve at connect time.

All commands support --verbose / -v for debug logging and --json-log for structured JSON output. Run r2g --version to print the installed version, and r2g --install-completion to set up shell tab-completion (bash/zsh/fish).

Mapping configuration

The YAML mapping config controls how PostgreSQL tables map to ArangoDB collections. See examples/sample_mapping.yaml for a commented example.

Key sections:

• collections -- per-table settings: target collection name, field renames (field_mappings), exclude_fields, include_fields, is_join_table
• edges -- foreign key relationships: edge collection name, from/to vertex collections, from/to fields
• type_overrides -- force a specific JSON type for a column when auto-detection is wrong
• key_separator -- character used to join composite primary key values (default: _)

CDC (Change Data Capture)

After an initial full load, use CDC to keep ArangoDB in sync with PostgreSQL changes in near real-time:

# 1. Create a logical replication slot on PostgreSQL
r2g cdc-setup --pg-conn "postgresql://user:pass@localhost/mydb"

# 2. Start the CDC listener (blocks, Ctrl+C to stop)
r2g cdc-start --pg-conn "postgresql://user:pass@localhost/mydb" \
  schema.json mapping.yaml \
  --endpoint http://localhost:8529 \
  --database mydb \
  --username root --password secret \
  --poll-interval 1.0 --batch-size 1000

# 3. Check slot status
r2g cdc-status --pg-conn "postgresql://user:pass@localhost/mydb"

# 4. Clean up when done
r2g cdc-teardown --pg-conn "postgresql://user:pass@localhost/mydb"

The listener supports two PostgreSQL output plugins:

• test_decoding (default) -- built-in to PostgreSQL, no extensions needed
• wal2json -- requires the wal2json extension; provides cleaner JSON output

Conflict resolution

Use --conflict-policy to control how write conflicts are handled:

r2g cdc-start --pg-conn "..." schema.json mapping.yaml \
  --conflict-policy last_write_wins

A conflict summary table is printed at the end of each CDC session showing counts by conflict type.

For UPDATE/DELETE events, set REPLICA IDENTITY FULL on source tables to capture old row values:

ALTER TABLE users REPLICA IDENTITY FULL;

Kafka CDC

Consume change events from Kafka instead of PostgreSQL logical replication. Requires the optional dependency:

pip install 'r2g-arango[kafka]'

Debezium messages (default --format debezium):

r2g kafka-start schema.json mapping.yaml \
  --brokers localhost:9092 \
  --topics dbserver1.public.users,dbserver1.public.orders \
  --group-id r2g-cdc \
  --format debezium \
  --offset-reset earliest \
  --endpoint http://localhost:8529 \
  --database mydb \
  --username root --password secret

Flat JSON from custom producers (--format flat):

r2g kafka-start schema.json mapping.yaml \
  --brokers kafka.example.com:9092 \
  --topics my_app.changes \
  --format flat \
  --group-id r2g-flat \
  --conflict-policy source_wins \
  --endpoint http://localhost:8529 \
  --database mydb

Use --offset-reset earliest or latest to control where a new consumer group starts. The same conflict policies as PG CDC apply via --conflict-policy. Press Ctrl+C for graceful shutdown (offsets committed for processed batches). Optional flags include --batch-size for poll batching.

Known limitations

r2g is a reference application (see the Status note near the top): well-tested, but not held to a production operational bar. The following constraints apply:

• Supported sources: PostgreSQL, MySQL / MariaDB, SQL Server, Snowflake, ClickHouse, and CSV directories (Kafka is supported for streaming sync via kafka-start and introspection in the catalog). Schema introspection, FK inference (with value-overlap sampling on PostgreSQL, MySQL, SQL Server, and CSV), dump export (r2g source dump), and streaming into ArangoDB (r2g stream --source …) work across these backends through a common SourceConnector / SourceSession abstraction. MySQL is gated on the optional r2g-arango[mysql] extra (pure-Python pymysql, also covers MariaDB); SQL Server on r2g-arango[sqlserver] (pure-Python pymssql); Snowflake on r2g-arango[snowflake]; ClickHouse on r2g-arango[clickhouse] (pure-Python clickhouse-connect). PostgreSQL, MySQL, and SQL Server are verified end-to-end against live servers in the integration suite; Snowflake password/key-pair snapshot, reviewed-key overlay, mapping, dump, and streaming paths are field-validated against the live Customer 360 demo. No SQLite or Oracle support yet.
• External data catalog discovery (Phase 8a): connect to an OpenMetadata catalog (r2g-arango[openmetadata]) to browse its database/schema/Kafka assets and import a selection as an r2g source — see r2g catalog. Distinct from r2g's internal catalog; read-only; credentials stay with the user (the catalog supplies host/db, not secrets). AWS Glue and Atlan are planned next (see docs/PRD.md Phase 8).
• Ontology derivation (Phase 10): an optional way to propose a richer target graph from the introspected schema — the engine proposes, the deterministic pipeline disposes. Two engines: an LLM engine (r2g-arango[llm]; OpenAI, Anthropic/Claude, or any OpenAI-compatible/local endpoint like Ollama, vLLM, LM Studio), and a deterministic engine (r2g-arango[ontology]) that runs the shared relational-schema-analyzer — the introspection core originally extracted from r2g — to derive a conceptual model (semantic collection names, join-table detection, foreign-key relationships, confidence/provenance) offline, from structure alone (no rows, no network); add --refine to LLM-improve it. The LLM path is metadata-only (no row data; Phase-9 Restricted/PII columns redacted to name-only and never sampled), prompt-injection-hardened; opt-in --sample/--ground add non-sensitive example values and Phase-11 denormalization findings. Both engines' proposals are validated/repaired against the real schema so a proposal can never load a hallucinated table or column, flowing through the same validate_config → mapper-review → loader path as Auto-Map (still the default); nothing is applied without explicit confirmation. Use it from the CLI (r2g ontology suggest --engine {llm,rsa}) or the Mapping Studio: a "Suggest model (AI)" action (Actions menu / canvas right-click / m) opens a floating review panel — pick the engine, accept or reject each suggestion per item, then apply the selection as an editable draft to review and Save.
• Data validation is opt-in -- orphaned foreign key references (FK values pointing to non-existent PKs) will produce edges to vertices that don't exist in ArangoDB. Use validate-data before import to catch these, but it is not enforced automatically.
• Incremental + change capture -- stream --since filters rows by a timestamp column for basic incremental loads; full change capture is available via the CDC (cdc-start) and Kafka (kafka-start) pipelines with configurable conflict resolution, and mapping-diff / selective-reload reconcile mapping changes against an already-loaded target.
• Self-referential FKs -- these work but produce edges within the same collection (e.g., orders.referrer_id -> customers.id creates orders_to_customers_referrer_id). This is correct but may be unexpected.
• ArangoDB write path -- the stream command connects directly to ArangoDB via python-arango, but the file-based paths (generate-csv-import, generate-import) still require arangoimport installed separately.
• Credential handling -- catalog secrets (source connection strings, target passwords) are encrypted at rest with Fernet using a key from R2G_SECRET_KEY or ~/.r2g/secret.key (managed via r2g secrets init|status|migrate); API responses redact them. Generated import scripts still reference connection defaults via environment variables, and there is no integrated remote secrets manager (e.g. Vault).

Testing

pytest tests/ -v

Over 1,000 tests (6 integration tests skipped without Docker) covering CLI commands (via typer.testing.CliRunner, including CDC and Kafka commands), types (including composite FK serialization), schema diff, config migration, data validation (referential integrity with orphan detection), topological sort (dependency ordering, circular FK detection), config validation (including self-referential FKs, duplicate edge naming, PK-less table warnings, and reserved-attribute checks), dump reader, node/edge transformers, import generators (JSONL and CSV-direct), visualizer, ArangoDB writer (retry logic, error surfacing, single-document CDC ops), the full CDC stack (models, output-plugin parsers, delta transformer, handler, conflict resolution, replication listener), Kafka parsers and consumer, the streaming pipeline (sequential, parallel, filtering, skip-existing, topological/since handling), the expression evaluator (expressions.py), the MySQL, SQL Server, Snowflake, ClickHouse, and CSV connectors, the CSI v1 and R2RML federation emitters, FK inference (name heuristics + PostgreSQL/MySQL/SQL Server/CSV value-overlap samplers), naming conventions, mapping diff + selective reload, temporal graph mode (applier, models, queries), credential encryption (security.py), the Mapping Studio API (ui/server.py) and MCP server, plus end-to-end integration tests against live PostgreSQL / MySQL / SQL Server + ArangoDB.

To run unit tests only (no Docker required):

pytest tests/ -m "not integration"

To run all tests including integration (requires Docker PG + ArangoDB):

pytest tests/ -v

Roadmap

Phases 1 through 4 are implemented. See docs/PRD.md for the full phased roadmap:

• Phase 1 -- Table dump file processing (MVP): schema ingestion, JSONL transforms, CSV-direct import, visualizer -- complete
• Phase 2 -- Direct PostgreSQL streaming (server-side cursors, HTTP API bulk import, REPEATABLE READ snapshots) -- complete
• Phase 3 -- CDC integration: event models, delta transformer, handler, replication listener, output plugin parsers (test_decoding/wal2json), continuous polling loop, conflict resolution (source_wins/last_write_wins/log_and_skip/fail), CLI commands -- complete
• Phase 4 -- Kafka integration: Debezium parser, flat JSON parser, confluent-kafka consumer, kafka-start CLI command -- complete
• Phase 5 -- Temporal graph mode: immutable-proxy time travel pattern (ProxyIn/Entity/ProxyOut), soft deletes with created/expired versioning, TTL aging, MDI-prefixed temporal indexes, point-in-time query templates, SmartGraph compatibility -- implemented (r2g.temporal, --temporal/--ttl-seconds/--smart-field on cdc-start/kafka-start; live-warehouse field validation pending)
• Phase 5f -- Naming conventions (PascalCase collections / camelCase properties + edges / snake_case) and rename change-management for already-loaded targets: r2g.naming, identity-based diff in mapping-diff, in-place selective-reload, migration-plan / migrate API endpoints, reserved-attribute protection -- implemented
• Phase 6 -- Snowflake integration -- done. Includes password/key-pair authentication, schema introspection, reviewed RSA key overlays with provenance/fingerprints, type mapping, source-agnostic CSV dumps and direct streaming, and the bundled Customer 360 rehearsal. Deterministic FK inference is available, but Snowflake value-overlap sampling remains name-only. r2g stream --source <name> and r2g source dump <name> dispatch through the registered source connector; legacy PostgreSQL-only flags remain available for compatibility.
• Phase 12 -- Federation: forward CSI v1 and R2RML emitters (export-csi / export-r2rml) so r2g mappings drive the Contextual Data Fabric federated-query engine; ClickHouse source connector (federation mapping + ETL); P6.7 cross-source shared keys emitted as conceptualModel.joinKeys -- shipped (producer side); the fabric-side E1 routing consumer is tracked separately
• Phase 7+ -- Additional sources, LLM-driven ontology derivation, ArangoRDF, bi-directional sync -- exploratory

Contributing

Contributions are welcome. Before opening a pull request please read CONTRIBUTING.md and the Code of Conduct.

Bug reports and feature requests go to the issue tracker. Use the provided templates so reports include the context needed to reproduce.

Security

Please do not report security vulnerabilities through public issues. See SECURITY.md for the responsible disclosure process.

Hardening the Studio UI and MCP server

The Studio UI and MCP servers are built for frictionless local use: bound to loopback they need no token. When exposing either on a shared host, set these environment variables (see .env.example):

The MCP stdio transport (Cursor / Claude Desktop) is local-only and needs no token; only the network-exposed SSE transport is gated.

License

r2g is licensed under the Apache License, Version 2.0. Third-party dependency notices are in NOTICE.

Changelog

See CHANGELOG.md for release notes.