impl: snowflake integration

Author: itsezc

Cargo.lock

@@ -1,5 +1,5 @@
 [workspace]
-members = ["crates/binlog-protocol", "crates/mysql-binlog-source", "crates/kafka-source", "crates/kafka-producer", "crates/kafka-types", "crates/postgresql", "crates/postgresql-wal2json-source", "crates/postgresql-pgoutput-source", "crates/pgoutput-protocol", "crates/postgresql-trigger-source", "crates/csv-source", "crates/jsonl-source", "crates/file", "crates/sync-core", "crates/loadtest-generator", "crates/loadtest-populate", "crates/mysql-types", "crates/postgresql-types", "crates/mongodb-types", "crates/surreal2-types", "crates/surreal3-types", "crates/json-types", "crates/csv-types", "crates/neo4j-types", "crates/loadtest-populate-mysql", "crates/loadtest-populate-postgresql", "crates/loadtest-populate-mongodb", "crates/loadtest-populate-csv", "crates/loadtest-populate-jsonl", "crates/loadtest-populate-neo4j", "crates/loadtest-populate-kafka", "crates/loadtest-verify-surreal2", "crates/loadtest-verify-surreal3", "crates/surreal-sink", "crates/surreal2-sink", "crates/surreal3-sink", "crates/surreal-client", "crates/surreal2-client", "crates/surreal3-client", "crates/surreal-version", "crates/mongodb-changestream-source", "crates/neo4j-source", "crates/checkpoint", "crates/checkpoint-surreal3", "crates/mysql-trigger-source", "crates/loadtest-distributed", "crates/surrealdb-multi-sdk-demo", "crates/interleaved-snapshot"]
+members = ["crates/binlog-protocol", "crates/mysql-binlog-source", "crates/kafka-source", "crates/kafka-producer", "crates/kafka-types", "crates/postgresql", "crates/postgresql-wal2json-source", "crates/postgresql-pgoutput-source", "crates/pgoutput-protocol", "crates/postgresql-trigger-source", "crates/csv-source", "crates/jsonl-source", "crates/file", "crates/sync-core", "crates/loadtest-generator", "crates/loadtest-populate", "crates/mysql-types", "crates/postgresql-types", "crates/mongodb-types", "crates/surreal2-types", "crates/surreal3-types", "crates/json-types", "crates/csv-types", "crates/neo4j-types", "crates/loadtest-populate-mysql", "crates/loadtest-populate-postgresql", "crates/loadtest-populate-mongodb", "crates/loadtest-populate-csv", "crates/loadtest-populate-jsonl", "crates/loadtest-populate-neo4j", "crates/loadtest-populate-kafka", "crates/loadtest-verify-surreal2", "crates/loadtest-verify-surreal3", "crates/surreal-sink", "crates/surreal2-sink", "crates/surreal3-sink", "crates/surreal-client", "crates/surreal2-client", "crates/surreal3-client", "crates/surreal-version", "crates/mongodb-changestream-source", "crates/neo4j-source", "crates/checkpoint", "crates/checkpoint-surreal3", "crates/mysql-trigger-source", "crates/loadtest-distributed", "crates/surrealdb-multi-sdk-demo", "crates/interleaved-snapshot", "crates/snowflake-types", "crates/snowflake-source"]
 resolver = "2"
 
 [package]
@@ -142,6 +142,10 @@ surreal-sync-postgresql = { path = "crates/postgresql" }
 # Watermark-based interleaved snapshot full-sync framework
 surreal-sync-interleaved-snapshot = { path = "crates/interleaved-snapshot" }
 
+# Snowflake ingestion source (SQL REST API v2, key-pair JWT auth)
+surreal-sync-snowflake-source = { path = "crates/snowflake-source" }
+snowflake-types = { path = "crates/snowflake-types" }
+
 # Binlog protocol (Flavor for binlog test containers)
 binlog-protocol = { path = "crates/binlog-protocol" }
 

Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "surreal-sync-snowflake-source"
+version = "0.1.0"
+edition = "2021"
+description = "Snowflake ingestion source for surreal-sync (SQL REST API v2, key-pair JWT auth)"
+authors = ["surreal-sync contributors"]
+
+[dependencies]
+# Core data model + sink trait
+sync-core = { path = "../sync-core" }
+surreal-sink = { path = "../surreal-sink" }
+snowflake-types = { path = "../snowflake-types" }
+
+# HTTP client (rustls to match the process-wide aws-lc-rs provider; no OpenSSL)
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+
+# Snowflake key-pair JWT generation
+snowflake-jwt = "0.3"
+
+# Async runtime (timers for async statement polling)
+tokio = { version = "1.49", features = ["time"] }
+
+# Serialization
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+
+# Error handling / logging
+anyhow = "1.0"
+tracing = "0.1"

crates/snowflake-source/Cargo.toml

@@ -0,0 +1,33 @@
+//! Table discovery for a Snowflake schema.
+
+use anyhow::Result;
+
+use crate::client::SnowflakeClient;
+
+/// List base-table names in `database.schema` via `INFORMATION_SCHEMA.TABLES`.
+///
+/// Snowflake stores unquoted identifiers upper-cased, so `schema` is upper-cased
+/// before matching. Views and temporary tables are excluded (`TABLE_TYPE = 'BASE TABLE'`).
+pub async fn list_tables(
+    client: &SnowflakeClient,
+    database: &str,
+    schema: &str,
+) -> Result<Vec<String>> {
+    let db = database.to_ascii_uppercase();
+    let schema_upper = schema.to_ascii_uppercase();
+    let sql = format!(
+        "SELECT TABLE_NAME FROM \"{db}\".INFORMATION_SCHEMA.TABLES \
+         WHERE TABLE_SCHEMA = '{schema_upper}' AND TABLE_TYPE = 'BASE TABLE' \
+         ORDER BY TABLE_NAME"
+    );
+
+    let result = client.execute_query(&sql).await?;
+    let tables = result
+        .rows
+        .iter()
+        .filter_map(|row| row.first())
+        .filter_map(|cell| cell.as_str())
+        .map(|s| s.to_string())
+        .collect();
+    Ok(tables)
+}