# Snowflake Snowpipe Streaming

Stream records into Snowflake in near real time via the [Snowpipe Streaming high-performance architecture](https://docs.snowflake.com/en/user-guide/snowpipe-streaming/snowpipe-streaming-high-performance-overview). Records become queryable in the target table within seconds of being written, with no intermediate files and no warehouse required for ingest.

## Prerequisites

- A Snowflake account on a supported region (high-performance architecture is GA on AWS, Azure, and GCP commercial regions).
- A Snowflake **user** dedicated to Monad ingest, with key-pair auth enabled.
- A **role** with `USAGE` on the database/schema, `INSERT` on the target table, and `OPERATE`/`MONITOR` on the pipe.
- A **target table** with columns matching the JSON keys you stream.
- A **STREAMING pipe** (`CREATE PIPE ... DATA_SOURCE = (TYPE = 'STREAMING')`) pointing at that table.
- The Monad node must be able to reach `*.snowflakecomputing.com` over HTTPS.

## Setup Instructions

### Step 1: Generate an RSA key pair

Generate an unencrypted PKCS#8 RSA key pair locally:

```bash
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out monad_key.pem -nocrypt
openssl rsa -in monad_key.pem -pubout -out monad_key.pub
```

Keep `monad_key.pem` private — you'll paste it into Monad as a secret. The contents of `monad_key.pub` get registered on the Snowflake user.

See [Key-pair authentication and key-pair rotation](https://docs.snowflake.com/en/user-guide/key-pair-auth) for details, including rotation procedure.

### Step 2: Create the Snowflake user and role

Run these as a role with `SECURITYADMIN` or higher:

```sql
CREATE ROLE monad_streaming_role;

CREATE USER monad_streaming_user
  TYPE = SERVICE
  DEFAULT_ROLE = monad_streaming_role
  RSA_PUBLIC_KEY = 'MIIBIj...your public key body, no header/footer/newlines...';

GRANT ROLE monad_streaming_role TO USER monad_streaming_user;
```

> The connector signs JWTs that **do not carry a role claim**. The user's `DEFAULT_ROLE` is what Snowflake uses to authorize the request, so it must point at the role that owns the grants below.

### Step 3: Create the target table

The table schema must match the JSON shape you're streaming. For semi-structured ingest, a single `VARIANT` column works well:

```sql
USE ROLE sysadmin;
CREATE DATABASE IF NOT EXISTS monad_db;
CREATE SCHEMA   IF NOT EXISTS monad_db.monad_schema;

CREATE TABLE monad_db.monad_schema.cloudtrail_events (
  record VARIANT
);
```

For typed columns, define them explicitly and ensure your upstream pipeline emits matching keys.

### Step 4: Create the STREAMING pipe

The pipe is the ingest target the connector writes to. It must be created with `DATA_SOURCE = (TYPE = 'STREAMING')`:

```sql
CREATE PIPE monad_db.monad_schema.cloudtrail_pipe
  AS COPY INTO monad_db.monad_schema.cloudtrail_events
  FROM TABLE (
    DATA_SOURCE => (TYPE => 'STREAMING')
  )
  MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;
```

For a typed schema, replace `MATCH_BY_COLUMN_NAME` with an explicit column list. See [CREATE PIPE](https://docs.snowflake.com/en/sql-reference/sql/create-pipe).

### Step 5: Grant the role access

```sql
USE ROLE sysadmin;

GRANT USAGE   ON DATABASE monad_db                                TO ROLE monad_streaming_role;
GRANT USAGE   ON SCHEMA   monad_db.monad_schema                   TO ROLE monad_streaming_role;
GRANT INSERT  ON TABLE    monad_db.monad_schema.cloudtrail_events TO ROLE monad_streaming_role;
GRANT OPERATE, MONITOR ON PIPE monad_db.monad_schema.cloudtrail_pipe TO ROLE monad_streaming_role;
```

### Step 6: Configure the output in Monad

In the pipeline editor, add the **Snowflake Snowpipe Streaming** output and fill in the settings below.

## Configuration

### Settings

| Setting | Type | Required | Description |
| ------- | ---- | -------- | ----------- |
| Account | string | Yes | Snowflake account identifier, in the form `orgname-account_name` (e.g. `acme-prod1`). This is the same identifier you use in the Snowflake URL. |
| User | string | Yes | The Snowflake user for key-pair auth. Its `DEFAULT_ROLE` must be a role with access to the pipe. |
| Database | string | Yes | Database containing the pipe. Alphanumeric and underscores only. |
| Schema | string | Yes | Schema within the database that contains the pipe. Alphanumeric and underscores only. |
| Pipe | string | Yes | Name of the pre-existing `STREAMING` pipe. Alphanumeric and underscores only. |
| Channel Prefix | string | No | Prefix used for channel names. Final channel names are `{prefix}_{uuid}_{i}`, where `uuid` is a fresh random identifier. And `i` ranges over the channel pool. Defaults to `monad`. |
| Private Key | secret | Yes | RSA private key (PEM) for the user. Either raw PEM or base64-encoded PEM is accepted. |
| Batch Config | object | No | Controls when a batch is flushed to Snowflake. See below. |

### Batch tuning

| Field | Default | Min | Max | Notes |
| ----- | ------- | --- | --- | ----- |
| Records per batch | 10,000 | 1 | 50,000 | Soft cap; flush also fires on size/time. |
| Batch size (bytes) | 2 MiB | 1 MiB | 3.5 MiB | Snowflake caps each `appendRows` body at 4 MiB; we stay below that to leave headroom for JSON framing. |
| Publish rate (seconds) | 10 | 1 | 60 | Maximum time a partial batch waits before being flushed. |

The connector picks the parallel channel count from the configured batch size: smaller batches get more parallel channels (up to 32) to keep per-second throughput up; larger batches use fewer channels since each one already moves more bytes per request. You don't tune this directly — choose the batch size that matches your freshness/throughput goals and the channel pool sizes itself.

## Debugging

### Status checks in Snowflake

> Do **not** use `SYSTEM$PIPE_STATUS` — that's the classic Snowpipe function and its fields (`pendingFileCount` etc.) don't apply to streaming pipes.

For the high-performance architecture, use:

```sql
-- List all open channels on the pipe and their commit progress.
-- This is the most useful diagnostic: client_sequencer / row_sequencer /
-- last_committed_offset_token / last_inserted_timestamp tell you whether
-- each channel is advancing under load.
SHOW CHANNELS IN PIPE MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE;

-- Pipe definition and current status
DESCRIBE PIPE MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE;

-- Per-pipe ingest volume and credit usage (real-time, INFORMATION_SCHEMA)
SELECT *
FROM TABLE(INFORMATION_SCHEMA.PIPE_USAGE_HISTORY(
  DATE_RANGE_START => DATEADD('hour', -1, CURRENT_TIMESTAMP()),
  PIPE_NAME => 'MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE'
));

-- Same data, longer retention, ~45 min lag (ACCOUNT_USAGE)
SELECT *
FROM SNOWFLAKE.ACCOUNT_USAGE.PIPE_USAGE_HISTORY
WHERE PIPE_NAME = 'MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE'
ORDER BY START_TIME DESC
LIMIT 100;
```

`COPY_HISTORY` does **not** record successful Snowpipe Streaming row inserts — they commit directly into the table. It only surfaces parse/load errors. An empty `COPY_HISTORY` result means no errors, not no traffic.

See [INFORMATION_SCHEMA functions](https://docs.snowflake.com/en/sql-reference/info-schema) for the full set.

### Confirming rows landed

```sql
-- 1. Confirm the column name and type
DESC TABLE MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_EVENTS;

-- 2. Plain row count — works regardless of schema
SELECT COUNT(*) FROM MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_EVENTS;

-- 3. If the target is a single VARIANT column (e.g. named "record"),
-- you can pull the latest event timestamp:
SELECT COUNT(*), MAX(record:eventTime::TIMESTAMP_NTZ)
FROM MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_EVENTS;
```

If Monad reports `append rows ok` but the row count isn't climbing, run `SHOW CHANNELS IN PIPE …` and check `last_committed_offset_token` — if it's advancing but the table is empty, the schema mismatch is silently dropping rows.

## Troubleshooting

### Common Issues

1. **`snowflake 408: stream timeout`** — Snowflake received the bytes but couldn't commit them inside its server-side ack window. Usually the pipe is congested. Retries are safe (continuation tokens make appends idempotent). If sustained, drop concurrent writers against this pipe or shard to a second pipe.
2. **`Client.Timeout exceeded while awaiting headers`** — the HTTP client gave up before Snowflake responded. Check network latency from the Monad cluster to the ingest host; firewalls or NAT gateways are common culprits.
3. **`JWT token is invalid` / `390144`** — the public key on the Snowflake user doesn't match the private key in Monad. Re-run `DESC USER` and verify the `RSA_PUBLIC_KEY` fingerprint matches.
4. **`Insufficient privileges to operate on pipe`** — the user's `DEFAULT_ROLE` doesn't have `OPERATE` on the pipe. The JWT carries no role claim, so workspace/session role overrides won't help — fix `DEFAULT_ROLE`.
5. **No rows in the target table despite successful appends** — usually a schema mismatch. Check `INFORMATION_SCHEMA.COPY_HISTORY` for `FIRST_ERROR_MESSAGE`.
6. **`continuation token mismatch` followed by reopen** — expected behavior when a request times out and the connector recovers. If frequent, treat as a symptom of pipe-side congestion.
7. **Region mismatch** — high-performance architecture is not yet available in every region. Confirm your account's region from the [Snowpipe Streaming high-performance overview](https://docs.snowflake.com/en/user-guide/snowpipe-streaming/snowpipe-streaming-high-performance-overview).

## Best Practices

- **One pipe per data shape**, not one pipe per pipeline. Reuse pipes across pipelines whenever the target table is the same.
- **Use semi-structured (`VARIANT`) columns** for schema-evolving sources like Cloudtrail. You can flatten downstream with views or dynamic tables.
- **Rotate keys** at least annually. Snowflake supports two active public keys per user (`RSA_PUBLIC_KEY` and `RSA_PUBLIC_KEY_2`) so you can rotate without downtime — see [Key-pair rotation](https://docs.snowflake.com/en/user-guide/key-pair-auth).
- **Monitor `PIPE_USAGE_HISTORY`** weekly to track credit consumption and ingestion latency trends.

## Cost

Snowpipe Streaming high-performance ingest is billed on a serverless compute model — there is no warehouse to size. Costs vary by region, edition, and ingest volume. The current per-credit and per-feature rates are published in Snowflake's [Credit Consumption Table (PDF)](https://www.snowflake.com/legal-files/CreditConsumptionTable.pdf). That document is the source of truth and is updated by Snowflake periodically; we do not mirror its rates here.

To see your actual usage, query:

```sql
SELECT *
FROM SNOWFLAKE.ACCOUNT_USAGE.PIPE_USAGE_HISTORY
WHERE PIPE_NAME = 'MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE'
ORDER BY START_TIME DESC
LIMIT 100;
```

## Related Articles

- [Snowpipe Streaming high-performance architecture overview](https://docs.snowflake.com/en/user-guide/snowpipe-streaming/snowpipe-streaming-high-performance-overview)
- [Snowpipe Streaming high-performance REST API reference](https://docs.snowflake.com/en/user-guide/snowpipe-streaming/snowpipe-streaming-high-performance-rest-api)
- [CREATE PIPE](https://docs.snowflake.com/en/sql-reference/sql/create-pipe)
- [Key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth)
- [INFORMATION_SCHEMA functions](https://docs.snowflake.com/en/sql-reference/info-schema)
- [Credit Consumption Table (PDF)](https://www.snowflake.com/legal-files/CreditConsumptionTable.pdf)