Add Snowflake snapshot input connector by xiasongh · Pull Request #6601 · feldera/feldera

xiasongh · 2026-07-08T14:42:41Z

Add an integrated Snowflake input connector backed by Arrow REST results. Support JWT key-pair authentication, projection pushdown, connector-managed snapshot transactions, concurrent chunk reads, streaming decode, and parallel parsing.

Describe Manual Test Plan

Checklist

Unit tests added/updated
Integration tests added/updated
Documentation updated
Changelog updated

Breaking Changes?

Mark if you think the answer is yes for any of these components:

OpenAPI / REST HTTP API / feldera-types / manager (What is a breaking change?)
Feldera SQL (Syntax, Semantics)
feldera-sqllib (incl. dependencies fxp, etc.) (What is a breaking change?)
Python SDK (What is a breaking change?)
fda (CLI arguments)
Adapters (including configuration)
Storage Format / Checkpoints
Others (specify)

Describe Incompatible Changes

mythical-fred

Draft, so high-level only. Architecturally clean: new feldera-snowflake crate under with-snowflake in the default feature set, SnowflakeReaderConfig mirrored in Rust and Java with parallel validators and matching test coverage, integrated-connector plumbing wired through ConnectorValidator / integrated.rs / TransportConfig::SnowflakeInput. Unit tests for query construction, config validation, Arrow decoding, gzip streaming, URL redaction, and an #[ignore]d live smoke test. Nice shape overall.

Higher-level things worth deciding before this comes out of draft:

Snapshot-only, but framed as extensible. SnowflakeIngestMode and SnowflakeTransactionMode are single-variant enums (Snapshot / None+Snapshot). That is fine as forward-compatible wire config, but the roadmap should be explicit: is CDC via CHANGES / streams / Snowpipe the intended follow-up, or is this permanently a bulk-snapshot connector? The fault_tolerance() -> None and Replay { .. } => panic! decisions hinge on that answer — for a snapshot-only connector they are reasonable; for a future CDC mode they will need to be revisited before merge, not after.
commit_transaction_on_error semantics. On mid-stream failure with transaction_mode: snapshot the connector currently publishes a commit_transaction for whatever it already queued. The inline comment acknowledges the constraint ("Feldera does not currently expose transaction rollback to connectors"), but the resulting behaviour — surface a partial snapshot as a committed Feldera transaction — is the opposite of what snapshot transaction mode should promise. Worth discussing whether the correct default is to signal an unhealthy connector + refuse to commit, or to wait for a rollback primitive in the pipeline before shipping the snapshot variant of transaction_mode at all.
max_concurrent_readers has no upper bound. #[schema(minimum = 1)] + runtime check for > 0, but a user can request 10_000. flatten_unordered(max_concurrent_readers) will happily open that many concurrent HTTP streams against Snowflake's chunk storage. Cap it (say, 64) or document a hard limit; do not leave it as a foot-gun.
Two crypto stacks. jsonwebtoken in the workspace uses aws-lc-rs, but the new crate pulls in openssl just to load the RSA private key and compute a public-key SHA-256 fingerprint. That works, but it doubles the crypto surface in a connector that only signs a JWT. Consider using rsa + sha2 (pure Rust) or aws-lc-rs's own key-parsing APIs so the whole connector rides on one crypto backend.
client.rs is 650 lines and covers auth, JWT signing, session management, statement execution, chunk streaming, retry policy, and URL redaction. Not a blocker, but splitting into auth.rs (JWT + login) and query.rs (statement + chunks) before it grows a CDC path will keep review-ability sane.
Replay panics. Even for snapshot-only that is a hard crash on a legitimate control-plane message. Prefer returning an error(true, ...) on the consumer and letting the controller unwind rather than tearing the process down.

For ready-to-review: add docs.feldera.com/docs/connectors/sources/snowflake.md, changelog entry, and a manual test plan describing the credentials/warehouse used for the #[ignore] smoke test.

mihaibudiu · 2026-07-08T18:19:52Z

+        return ok;
+    }
+
+    private boolean checkOptionalNonEmpty(ConfigReporter reporter, @Nullable String value, String name) {


I would move this function into the base interface, could be useful for other users.

mihaibudiu · 2026-07-08T18:23:42Z

+use anyhow::{bail, Result as AnyResult};
+use feldera_types::program_schema::Relation;
+
+pub(crate) fn build_snapshot_query(


Is this function identical to the one for Delta?
If so, we should pull it into a common library build_sql_snapshot_query

mihaibudiu · 2026-07-08T18:25:17Z

+    fn skips_unused_columns_when_requested_by_table() {
+        let mut relation = relation(&["ID", "UNUSED"]);
+        relation.fields[1].unused = true;
+        let position = SourcePosition {


don't we have a default() method on SourcePosition? Or perhaps an INVALID singleton?
We should use that.

mihaibudiu · 2026-07-08T18:26:07Z

+    /// Key-pair authentication using a JWT signed with an RSA private key.
+    #[default]
+    #[serde(rename = "SNOWFLAKE_JWT")]
+    SnowflakeJwt,


I assume this is designed to grow

mihaibudiu · 2026-07-08T18:27:20Z

+    ///
+    /// The value can be a fully-qualified Snowflake table name, for example
+    /// `"MY_DATABASE.MY_SCHEMA.MY_TABLE"`, or an unqualified table name when
+    /// `database` and `schema` are configured on the Snowflake session.


What if both specify a schema?

mihaibudiu · 2026-07-08T20:46:16Z

+pub(crate) struct SnowflakeClient {
+    pub(super) http: Client,
+    pub(super) base_url: Url,
+    session_token: String,


I wonder why some are pub and this one isn't

mihaibudiu · 2026-07-08T20:52:37Z

+    for (field, column) in schema.fields().iter().zip(batch.columns()) {
+        let logical_type = field.metadata().get("logicalType").map(String::as_str);
+
+        // Snowflake encodes scaled FIXED values as unscaled signed integers. Its C/C++ client


I would move this comment to the normalize_fixed function

mihaibudiu · 2026-07-08T20:55:04Z

+    // Snowflake's reference implementation documents both the current structured timestamp
+    // encoding and the legacy TIMESTAMP_TZ layout handled below:
+    // https://github.com/snowflakedb/libsnowflakeclient/blob/master/cpp/lib/ArrowChunkIterator.cpp
+    if let Some(values) = column.as_any().downcast_ref::<StructArray>() {


I hope this is right, I haven't checked, having some tests with real data will confirm

mihaibudiu · 2026-07-08T20:57:56Z

+    anyhow!("Snowflake timestamp value in column '{field_name}' at row {row} is out of range")
+}
+
+fn struct_child<'a, T: Array + 'static>(


This function is only used for timestamp decoding, the name should reflect that, and perhaps some documentation too.
The errors produced are otherwise very mysterious.

mihaibudiu · 2026-07-08T21:04:14Z

+    pub(crate) batches: BoxStream<'a, AnyResult<RecordBatch>>,
+}
+
+pub(crate) fn build_snapshot_query(


Also, pub functions should be documented.

mihaibudiu · 2026-07-08T21:12:01Z

To accept this we probably would need some end-to-end integration tests that generate data and load it. I wonder whether you need a real snowflake installation, or there's a tool to mock one.

swanandx · 2026-07-09T10:59:56Z

while reviewing, i see lot of this is something that can be abstracted away if we use some already tested library like https://github.com/andrusha/snowflake-rs ( or some other libraries? )

have we tried using it? if yes, why we decided to roll our own implementation? any downside to snowflake-rs? can we fork & solve it instead if it exists?

xiasongh · 2026-07-09T16:41:34Z

while reviewing, i see lot of this is something that can be abstracted away if we use some already tested library like https://github.com/andrusha/snowflake-rs ( or some other libraries? )

have we tried using it? if yes, why we decided to roll our own implementation? any downside to snowflake-rs? can we fork & solve it instead if it exists?

i did take a look at the existing libraries, but it didn't support streaming semantics for its query api. since the only thing i cared about was a performant query api and the surface area was small, i ended up implementing from scratch. to be honest, i didn't think too hard about this and wasn't sure if we wanted to maintain a fork. i'm totally fine with forking and adapting it to support

(thinking about it more, it would have been nice to atleast reuse the auth)

xiasongh · 2026-07-09T16:58:40Z

To accept this we probably would need some end-to-end integration tests that generate data and load it. I wonder whether you need a real snowflake installation, or there's a tool to mock one.

agreed. i know of this tool, but it's pretty basic and doesn't emulate snowflake's query api exactly. we will probably want to run tests against actual snowflake

ryzhyk · 2026-07-09T17:03:06Z

I don't think we can avoid having a real snowflake account and testing against that. I am ok with merging it though and adding tests later. Otherwise, this PR will bit rot. We just shouldn't list Snowflake among supported connector types.

ryzhyk · 2026-07-09T17:16:42Z

Merging this will also make sure that we have a release where the manager understands snowflake connector config. Let's create a meta-issue with a list of todos needed to productize this and merge once we're happy with what's already there.

Fred's two crypto stack concern may be a real blocker. It will increase the amount of security patching we'll need going forward. @gz , any opinion on this?
Re commit semantics: the approach we took in the delta connector is:
- For retryable errors, have a configurable retry count. It can be infinite by default. Setting a finite retry count allows transactions to commit despite errors (set connector status to unhealthy while retrying).
- For non-retryable errors, report connector errors and commit. We may add a mode to stop the pipeline then this happens in the future.

swanandx · 2026-07-09T20:06:35Z

(thinking about it more, it would have been nice to atleast reuse the auth)

maybe! I don't think current impl have session-token refresh, it would be good to have for client.

Add an integrated Snowflake input connector backed by Arrow REST results. Support JWT key-pair authentication, projection pushdown, connector-managed snapshot transactions, concurrent chunk reads, streaming decode, and parallel parsing.

mythical-fred reviewed Jul 8, 2026

View reviewed changes

ryzhyk temporarily deployed to ci July 8, 2026 21:00 — with GitHub Actions Inactive

mihaibudiu reviewed Jul 8, 2026

View reviewed changes

xiasongh added 7 commits July 21, 2026 18:28

Add Snowflake snapshot input connector

b1da973

Add an integrated Snowflake input connector backed by Arrow REST results. Support JWT key-pair authentication, projection pushdown, connector-managed snapshot transactions, concurrent chunk reads, streaming decode, and parallel parsing.

Organize and tune Snowflake connector

9e23405

Support Snowflake column mapping

ee86fec

Support configurable Snowflake number conversion

e6a5d25

Use table property for skipping unused Snowflake columns

b2a1f58

Add full snapshot Snowflake checkpoint support

07a0949

Add jittered retries for Snowflake chunk downloads

337271d

xiasongh force-pushed the xiasongh/snowflake-input-connector branch from de807df to 337271d Compare July 22, 2026 01:59

ryzhyk temporarily deployed to ci July 22, 2026 04:06 — with GitHub Actions Inactive

Uh oh!

Conversation

xiasongh commented Jul 8, 2026

Describe Manual Test Plan

Checklist

Breaking Changes?

Describe Incompatible Changes

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mihaibudiu commented Jul 8, 2026

Uh oh!

swanandx commented Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

xiasongh commented Jul 9, 2026

Uh oh!

xiasongh commented Jul 9, 2026

Uh oh!

ryzhyk commented Jul 9, 2026

Uh oh!

ryzhyk commented Jul 9, 2026

Uh oh!

swanandx commented Jul 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

swanandx commented Jul 9, 2026 •

edited

Loading