Arrow Data Plane

The Arrow data plane is a high-performance ingest path for Kanoniv Cloud that replaces row-by-row JSON uploads with columnar Arrow reads, DuckDB staging, and Parquet bulk ingest. It is designed for warehouse-to-cloud pipelines where data originates in Snowflake, BigQuery, or Postgres.

Why Arrow?

The default cloud pipeline reads warehouse tables row by row, converts to JSON, and uploads in batches of 500 via POST /v1/ingest/batch. This works but becomes slow at scale.

The Arrow data plane changes every step:

Step	JSON Path	Arrow Path
Read	Row-by-row cursor fetch	Columnar Arrow fetch (fetch_arrow_all())
Stage	pandas concat in memory	DuckDB UNION ALL (zero-copy)
Upload	500-record JSON batches	Single Parquet file per source
Server ingest	Row-by-row INSERT	COPY FROM Parquet

Result: ~6x faster end-to-end for a 6,400-record, 5-source pipeline (120s vs ~780s).

Prerequisites

Install the SDK with the dataplane extra:

pip install kanoniv[cloud,dataplane]

This adds three dependencies:

• pyarrow - Arrow columnar format and Parquet I/O
• duckdb - In-process SQL engine for staging
• snowflake-connector-python - Native Snowflake connector with Arrow support

How It Works

Snowflake          DuckDB             Parquet          Kanoniv API
   |                 |                   |                 |
   |  Arrow fetch    |                   |                 |
   |---------------->|  UNION ALL        |                 |
   |                 |  + column map     |                 |
   |                 |------------------>|  POST /v1/      |
   |                 |    export         |  ingest/parquet |
   |                 |                   |---------------->|
   |                 |                   |                 |  reconcile
   |                 |                   |                 |----------->
   |                 |                   |    bulk         |
   |                 |                   |    crosswalk    |
   |                 |                   |<----------------|

1. Arrow fetch - Each warehouse source is read as a native Arrow table via fetch_arrow_all(). This avoids Python object overhead and returns columnar data directly.
2. DuckDB staging - All source Arrow tables are registered in an in-memory DuckDB instance. A UNION ALL query maps source columns to canonical field names (using your spec's attribute mappings), adds source_name and external_id columns, and produces a single unified Arrow table.
3. Parquet export - The unified table is split back into per-source Parquet files with zstd compression.
4. Bulk ingest - Each Parquet file is uploaded via POST /v1/ingest/parquet (multipart). The server loads it via COPY FROM instead of row-by-row INSERT. Content hashing provides idempotent ingest - unchanged records are skipped.
5. Reconcile - Standard cloud reconciliation job runs on the ingested data.
6. Bulk crosswalk - Results are fetched via POST /v1/entities/linked/bulk (up to 1000 IDs per call) instead of individual entity lookups.

Auto-Detection

When you call kanoniv.cloud.reconcile(), the Arrow path is selected automatically if:

1. The dataplane extras are installed (pyarrow, duckdb, snowflake-connector-python)
2. At least one source has a connection_string (i.e., it was created with Source.from_warehouse() or Source.from_dbt())

No code changes are required. The same cloud.reconcile() call that works with CSV or pandas sources works with warehouse sources - it just runs faster.

If the dataplane extras are not installed, the function falls back to the JSON path silently.

import kanoniv
from kanoniv import Source, Spec

spec = Spec.from_file("specs/customer.yaml")

# These warehouse sources trigger the Arrow path automatically
sources = [
    Source.from_warehouse("crm", "ANALYTICS.IDENTITY.STG_CRM_CONTACTS",
                          connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY"),
    Source.from_warehouse("billing", "ANALYTICS.IDENTITY.STG_BILLING_CUSTOMERS",
                          connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY"),
]

# Arrow path is used automatically - same API as always
with kanoniv.cloud.reconcile(sources, spec, api_key="kn_...") as result:
    print(result.summary())

Quick Start

import kanoniv
from kanoniv import Source, Spec

spec = Spec.from_file("specs/customer.yaml")

# 1. Define warehouse sources
crm = Source.from_warehouse(
    "crm",
    "ANALYTICS.IDENTITY.STG_CRM_CONTACTS",
    connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY",
)

billing = Source.from_warehouse(
    "billing",
    "ANALYTICS.IDENTITY.STG_BILLING_CUSTOMERS",
    connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY",
)

support = Source.from_warehouse(
    "support",
    "ANALYTICS.IDENTITY.STG_SUPPORT_TICKETS",
    connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY",
)

# 2. Reconcile on the cloud (Arrow path auto-detected)
with kanoniv.cloud.reconcile(
    [crm, billing, support],
    spec,
    api_key="kn_...",
    timeout=300.0,
) as result:
    print(result.summary())
    df = result.to_pandas()
    print(f"{len(df)} canonical entities")

Manual Pipeline

For advanced use cases, you can use the Arrow components directly.

Stage sources into a unified Arrow table

from kanoniv import Source, Spec
from kanoniv.staging import stage_sources, export_parquet
from kanoniv.cloud_io import read_arrow

spec = Spec.from_file("specs/customer.yaml")
conn_str = "snowflake://user:pass@acct/ANALYTICS/IDENTITY"

sources = [
    Source.from_warehouse("crm", "STG_CRM_CONTACTS", connection_string=conn_str),
    Source.from_warehouse("billing", "STG_BILLING_CUSTOMERS", connection_string=conn_str),
]

# Stage: Arrow fetch + DuckDB UNION ALL + column mapping
unified = stage_sources(sources, spec, conn_str)
print(f"Staged {unified.num_rows} rows, {unified.num_columns} columns")

# Export to Parquet
path = export_parquet(unified, "/tmp/staged.parquet")
print(f"Parquet: {path}")

Upload Parquet and reconcile

from kanoniv import Client

with Client(api_key="kn_...") as client:
    # Upload spec
    client.specs.ingest(spec.raw, compile=True)

    # Upload Parquet
    result = client.ingest_parquet("crm", "/tmp/crm.parquet", entity_type="customer")
    print(f"New: {result['new']}, Updated: {result['updated']}, Unchanged: {result['unchanged']}")

    # Run reconciliation
    job = client.jobs.run("reconciliation")
    # ... poll until done

Read Arrow tables directly

from kanoniv.cloud_io import read_arrow

# Read a Snowflake table as an Arrow table
table = read_arrow(
    "ANALYTICS.IDENTITY.STG_CRM_CONTACTS",
    "snowflake://user:pass@acct/ANALYTICS/IDENTITY",
)
print(f"{table.num_rows} rows, columns: {table.column_names}")

Write results back to Snowflake

from kanoniv.cloud_io import write_parquet_to_warehouse

# Write Arrow tables back to Snowflake via Parquet PUT + COPY INTO
counts = write_parquet_to_warehouse(
    {"golden_customers": golden_table, "crosswalk": crosswalk_table},
    "snowflake://user:pass@acct/ANALYTICS/IDENTITY",
    schema="KANONIV_RESOLVED",
)
for table_name, n in counts.items():
    print(f"  {table_name}: {n} rows")

API Endpoints

POST /v1/ingest/parquet

Upload a Parquet file for bulk ingestion. Uses content hashing for idempotent ingest - records with unchanged content are skipped.

Request: Multipart form data with fields:

Field	Type	Required	Description
file	binary	yes	Parquet file
source_name	string	yes	Source system name
entity_type	string	no	Entity type (default: "entity")

Response:

{
  "new": 1234,
  "updated": 56,
  "unchanged": 5153
}

SDK:

result = client.ingest_parquet("crm", "data/crm.parquet", entity_type="customer")

POST /v1/entities/linked/bulk

Fetch linked entities for multiple canonical IDs in a single request. Used by the Arrow path to build the crosswalk table efficiently.

Request:

{
  "entity_ids": ["uuid-1", "uuid-2", "uuid-3"]
}

Maximum 1000 IDs per request.

Response:

{
  "results": {
    "uuid-1": [{"source_name": "crm", "external_id": "sf_001"}, ...],
    "uuid-2": [{"source_name": "billing", "external_id": "cus_002"}, ...],
    ...
  }
}

SDK:

result = client.entities.get_linked_bulk(["uuid-1", "uuid-2", "uuid-3"])
for entity_id, linked in result["results"].items():
    print(f"{entity_id}: {len(linked)} linked records")

Performance

Benchmarked with a 5-source dataset (6,443 records total, ~2,100 canonical entities):

Stage	JSON Path	Arrow Path
Read from Snowflake	~30s	~8s
Staging / transform	~5s (pandas)	~5s (DuckDB)
Upload to API	~600s (13 batches x 500)	~15s (5 Parquet files)
Reconciliation job	~120s	~120s
Fetch results	~25s	~10s (bulk)
Total	~780s	~120s (6.5x)

The biggest win is upload: Parquet bulk ingest replaces hundreds of small JSON batch requests with a handful of multipart file uploads.

Gotchas

Snowflake columns are UPPERCASE. Arrow tables fetched from Snowflake have uppercase column names (EMAIL, PHONE). The staging layer automatically lowercases them before DuckDB processing. If you use read_arrow() directly, call table.rename_columns([c.lower() for c in table.column_names]).

URL percent-encoding in connection strings. Snowflake passwords with special characters must be percent-encoded in SQLAlchemy URLs (e.g., p%40ss for p@ss). The cloud_io module decodes these automatically via urllib.parse.unquote().

Parquet files are temporary. The auto-detected Arrow path writes Parquet to temp files and cleans them up after upload. If you use the manual pipeline, manage your own file lifecycle.

Content hashing is idempotent. Re-uploading the same Parquet file is safe - the server computes SHA-256 content hashes and skips unchanged records. The response shows new, updated, and unchanged counts.

The local SDK is unaffected. The Arrow data plane only applies to kanoniv.cloud.reconcile(). Local reconciliation via kanoniv.reconcile() continues to use in-process Rust and is not changed.