feldera · xiasongh · Jul 8, 2026 · Jul 8, 2026 · Jul 13, 2026 · Jul 14, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -25,6 +25,7 @@ members = [
   "sql-to-dbsp-compiler/lib/readers",
   "crates/datagen",
   "crates/iceberg",
+  "crates/snowflake",
   "crates/storage",
   "crates/rest-api",
   "crates/ir",
@@ -61,6 +62,7 @@ arrow-json = "58"
 arrow-schema = "58"
 ascii_table = "=4.0.2"
 async-channel = "2.3.1"
+async-compression = { version = "0.4.42", features = ["gzip", "tokio"] }
 async-nats = "0.47.0"
 async-std = "1.12.0"
 async-stream = "0.3.5"
@@ -143,6 +145,7 @@ feldera-cloud1-client = "0.1.2"
 feldera-datagen = { path = "crates/datagen" }
 feldera-fxp = { version = "0.324.0", path = "crates/fxp", features = ["dbsp"] }
 feldera-iceberg = { path = "crates/iceberg" }
+feldera-snowflake = { path = "crates/snowflake" }
 feldera-observability = { version = "0.324.0", path = "crates/feldera-observability" }
 feldera-macros = { version = "0.324.0", path = "crates/feldera-macros" }
 feldera-sqllib = { version = "0.324.0", path = "crates/sqllib" }
@@ -213,6 +216,7 @@ paste = "1.0.12"
 petgraph = "0.6.0"
 pg-client-config = "0.1.2"
 pin-project-lite = "0.2.16"
+pkcs8 = { version = "0.10.2", features = ["3des"] }
 postgres = "0.19.10"
 postgres-openssl = "0.5.1"
 postgresql_embedded = { version = "0.20.0", features = ["bundled"] }
@@ -244,6 +248,7 @@ rmpv = "1.3.0"
 # Pin `roaring` exactly because the crate only supports the current stable
 # Rust and may start requiring newer compiler features in patch releases.
 roaring = "=0.11.3"
+rsa = { version = "0.9.10", features = ["pkcs5"] }
 rstest = "0.15"
 # Make sure this is the same rustls version used by the `tonic` crate.
 # See the `ensure_default_crypto_provider` function.

diff --git a/crates/adapters/Cargo.toml b/crates/adapters/Cargo.toml
@@ -19,6 +19,7 @@ default = [
     "with-kafka",
     "with-deltalake",
     "with-iceberg",
+    "with-snowflake",
     "with-avro",
     "with-nexmark",
     "with-pubsub",
@@ -39,6 +40,7 @@ with-deltalake = [
     "parquet/object_store",
 ]
 with-iceberg = ["feldera-iceberg"]
+with-snowflake = ["feldera-snowflake"]
 with-pubsub = ["google-cloud-pubsub", "google-cloud-gax"]
 with-avro = [
     "apache-avro",
@@ -82,6 +84,7 @@ feldera-observability = { workspace = true }
 feldera-samply = { workspace = true }
 feldera-storage = { workspace = true }
 feldera-iceberg = { workspace = true, optional = true }
+feldera-snowflake = { workspace = true, optional = true }
 awc = { workspace = true, features = [
     "compress-gzip",
     "compress-brotli",

diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -8415,7 +8415,7 @@ impl ControllerInner {
                 }
                 TransactionState::Started {
                     tid,
-                    start,
+                    start: _,
                     processed_records,
                 } if next.is_none() || next != open => {
                     // The next step belongs to a different transaction, or the

diff --git a/crates/adapters/src/integrated.rs b/crates/adapters/src/integrated.rs
@@ -134,6 +134,10 @@ pub fn create_integrated_input_endpoint(
                 consumer,
             ))
         }
+        #[cfg(feature = "with-snowflake")]
+        TransportConfig::SnowflakeInput(config) => Box::new(
+            feldera_snowflake::SnowflakeInputEndpoint::new(endpoint_name, config, consumer),
+        ),
         TransportConfig::PostgresInput(config) => {
             Box::new(PostgresInputEndpoint::new(endpoint_name, config, consumer))
         }

diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs
@@ -128,6 +128,7 @@ pub fn input_transport_config_to_endpoint(
         | TransportConfig::HttpOutput(_)
         | TransportConfig::RedisOutput(_)
         | TransportConfig::IcebergInput(_)
+        | TransportConfig::SnowflakeInput(_)
         | TransportConfig::NullOutput => return Ok(None),
     };
     Ok(Some(endpoint))

diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs
@@ -25,6 +25,7 @@ use crate::transport::postgres::{
 use crate::transport::pubsub::PubSubInputConfig;
 use crate::transport::redis::RedisOutputConfig;
 use crate::transport::s3::S3InputConfig;
+use crate::transport::snowflake::SnowflakeReaderConfig;
 use crate::transport::url::UrlInputConfig;
 use core::fmt;
 use feldera_ir::{MirNode, MirNodeId};
@@ -1879,6 +1880,7 @@ pub enum TransportConfig {
     RedisOutput(RedisOutputConfig),
     // Prevent rust from complaining about large size difference between enum variants.
     IcebergInput(Box<IcebergReaderConfig>),
+    SnowflakeInput(Box<SnowflakeReaderConfig>),
     PostgresInput(PostgresReaderConfig),
     PostgresCdcInput(PostgresCdcReaderConfig),
     PostgresOutput(PostgresWriterConfig),
@@ -1912,6 +1914,7 @@ impl TransportConfig {
             TransportConfig::DeltaTableOutput(_) => "delta_table_output".to_string(),
             TransportConfig::DynamoDBOutput(_) => "dynamodb_output".to_string(),
             TransportConfig::IcebergInput(_) => "iceberg_input".to_string(),
+            TransportConfig::SnowflakeInput(_) => "snowflake_input".to_string(),
             TransportConfig::PostgresInput(_) => "postgres_input".to_string(),
             TransportConfig::PostgresCdcInput(_) => "postgres_cdc_input".to_string(),
             TransportConfig::PostgresOutput(_) => "postgres_output".to_string(),

diff --git a/crates/feldera-types/src/transport.rs b/crates/feldera-types/src/transport.rs
@@ -13,4 +13,5 @@ pub mod postgres;
 pub mod pubsub;
 pub mod redis;
 pub mod s3;
+pub mod snowflake;
 pub mod url;