Skip to content

send_snapshot and reset API - #6042

Merged
gz merged 15 commits into
mainfrom
delta-flushing
May 14, 2026
Merged

gz merged 15 commits into
mainfrom
delta-flushing

Conversation

@gz

@gz gz commented Apr 14, 2026 •

Copy link
Copy Markdown
Contributor

Support for snapshot and follow #5980

This PR adds two features

send_snapshot: A output-connector boolean (default false). When set, the connector delivers the full current state of a materialized view once before it starts streaming incremental updates. Delivery is persisted as part of the checkpoint, so resuming a pipeline does not re-send it; modifying the connector across a restart triggers a fresh snapshot.

HTTP egress snapshot mode: /egress/...?send_snapshot=true makes the stream begin with a full snapshot of the materialized view before switching to incremental updates. Each response chunk carries a snapshot flag so consumers can distinguish snapshot rows from subsequent deltas.

How it works

send_snapshot

when false:
does nothing.

when true:
sets a Pending bit in OutputEndpointControl at connector startup, the circuit thread's push_output checks it each step and, when pending, grabs the latest cached view snapshot from trace_snapshots, and enqueues it as a OutputBatchType::Snapshot batch and marks the snapshot as Sent in OutputEndpointControl.

A separate initial_snapshot_sent bool is checkpointed (in CheckpointOutputEndpointMetrics) so resuming from a checkpoint skips the replay, while pipeline_diff clears it across restarts whenever the connector config changes that's how a modified connector re-fires the snapshot.

HTTP egress

send_snapshot=true is a per-request query parameter, since the endpoint only exists for the lifetime of the HTTP connection. When set, the handler seeds the stream by pulling the current materialized-view snapshot from trace_snapshots, emits it as a single chunk annotated with snapshot: true, then falls back to the normal delta-streaming.

Checklist

  • Validation/test it works correctly with completion tokens (hard to test)
  • Validation/test it works correctly with fault tolerance (if it can)
  • Make sure we implement reset properly for every connector where we can (currently supports postgres+delta) maybe we should log a warning/error for things that can’t be reset (redis)
  • If reset takes a long time are we able to surface this to users
  • Are we handling stats correctly?
  • Unit tests added/updated
  • Integration tests added/updated
  • Documentation updated

Breaking Changes?

Ideally, none.

@gz
gz requested a review from ryzhyk April 14, 2026 22:14
@gz

gz commented Apr 14, 2026

Copy link
Copy Markdown
Contributor Author

I fixed some unrelated clippy warnings along the way in avro, ignore them

Comment thread crates/adapterlib/src/transport.rs Resolved
@gz
gz force-pushed the delta-flushing branch from 1bf7937 to f6bd731 Compare April 14, 2026 22:33

@mythical-fred mythical-fred left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One blocker.

Comment thread crates/adapterlib/src/transport.rs Resolved

@ryzhyk ryzhyk left a comment •

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

additional comments:

  • I think this won't work correctly when restarting from a checkpoint, since we'll end up sending a snapshot on each restart. We can discuss offline how to handle this correctly.
  • I don't see any reason to only allow reset for snapshot-and-follow connectors. I suggest that instead of mode we have a Boolean attribute like initial_snapshot, which only controls behavior on startup. Independent of the value of this attribute, we allow reset for all materialized views.
  • Does this require any support in the coordinator? (cc @blp)

I didn't review Python and doc parts, since they will likely change if we address the above issues

Comment thread crates/adapterlib/src/format.rs Resolved
Comment thread crates/adapterlib/src/format.rs Resolved
Comment thread crates/fda/src/main.rs Resolved
Comment thread crates/adapters/src/controller.rs Outdated
} else {
for batch in snapshot_batches {
self.status.enqueue_batch(endpoint_id, batch.len());
queue.push(BatchQueueEntry {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You should be able to push the entire snapshot as one batch. Otherwise, it can produce mutually canceling updates as it processes it batch by batch.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please check; I didn't know how to do this

@gz

gz commented Apr 16, 2026 •

Copy link
Copy Markdown
Contributor Author

additional comments:

I think this won't work correctly when restarting from a checkpoint, since we'll end up sending a snapshot on each restart. We can discuss offline how to handle this correctly.

I'll add a test for now

I don't see any reason to only allow reset for snapshot-and-follow connectors. I suggest that instead of mode we have a Boolean attribute like initial_snapshot, which only controls behavior on startup. Independent of the value of this attribute, we allow reset for all materialized views.

the idea was that by marking a connector as snapshot_and_follow the user gains the ability to reset the table which also means that feldera is essentially "managing" the table (sending snapshots etc).
e.g., as you point out it's problematic to just truncate a postgresql table but the user would explicitly opt-in by setting this mode for the output connector.. IMO most connectors we have today support some sensible reset semantics as long as the user understands / buys into it

@gz

gz commented Apr 16, 2026

Copy link
Copy Markdown
Contributor Author

The goal with the output connector mode snapshot_and_follow was that

a) feldera takes ownership of the destination's contents: it can write a materialized view to the sink before switching to streaming incremental updates.. the only time to do this right now should be after reset
b) when you call the connector's reset endpoint it wipes the destination and replay a fresh snapshot of the current view state, it's only possible for snapshot_and_follow connectors

@ryzhyk

ryzhyk commented Apr 16, 2026

Copy link
Copy Markdown
Contributor

I don't see any reason to only allow reset for snapshot-and-follow connectors. I suggest that instead of mode we have a Boolean attribute like initial_snapshot, which only controls behavior on startup. Independent of the value of this attribute, we allow reset for all materialized views.

the idea was that by marking a connector as snapshot_and_follow the user gains the ability to reset the table which also means that feldera is essentially "managing" the table (sending snapshots etc). e.g., as you point out it's problematic to just truncate a postgresql table but the user would explicitly opt-in by setting this mode for the output connector.. IMO most connectors we have today support some sensible reset semantics as long as the user understands / buys into it

This seems a bit arbitrary. A concrete concern I have is that most connectors are created with the pipeline, when the pipeline is empty. The snapshot_and_follow setting doesn't make any difference at that point, and most users won't consider specifying it... until something goes wrong and they decide to reset the connector, which we arbitrarily reject, even though the engine is capable of doing it.

I don't think it's really intuitive that snapshot_and_follow means it's a Feldera-managed table and that it automatically changes the behavior when the connector is instantiated or reset. I understand the challenges of having too many configuration options, but at least they let you explicitly specify how the connector behaves.

@gz

gz commented Apr 16, 2026

Copy link
Copy Markdown
Contributor Author

ok lets discuss API tomorrow offline

@gz
gz force-pushed the delta-flushing branch 4 times, most recently from d8d3df8 to 24c41f8 Compare April 22, 2026 09:39

@mythical-fred mythical-fred left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@gz
gz force-pushed the delta-flushing branch 4 times, most recently from 2380b8c to c154dae Compare April 23, 2026 05:39
@gz gz changed the title snapshot_and_follow mode for connectors send_snapshot and reset API Apr 23, 2026
@gz
gz requested a review from ryzhyk April 23, 2026 06:22
@gz

gz commented Apr 23, 2026 •

Copy link
Copy Markdown
Contributor Author

@ryzhyk please re-review, update PR description with latest design as discussed

@gz
gz force-pushed the delta-flushing branch 2 times, most recently from a9dab7f to 7b7f3d1 Compare April 24, 2026 06:05

@mythical-fred mythical-fred left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Re-reviewing because the new [adapters] Materialized indexes commit (847+/235-, 23 files) is a substantial new feature added to a previously-approved PR. Two blockers and several smaller concerns.

Blockers

  1. Breaking signature change without SQL-compiler update. register_index and register_index_persistent changed key_fields from &[&SqlIdentifier] to &[String]. The Rust test scaffolding is updated, but sql-to-dbsp-compiler/.../ToRustVisitor.java still emits &vec!(&SqlIdentifier::from("field"), ...) (around line 1083) which is Vec<&SqlIdentifier>. After this PR lands, any SQL program with CREATE INDEX will fail to compile the generated Rust. CI will catch this once it runs, but it has to be fixed before merge.

  2. The headline optimization isn't wired up to the SQL compiler. Commit message says register_materialized_output_map simultaneously registers a materialized view and an index sharing one integral, and register_materialized_index registers additional indexes. Neither symbol appears anywhere under sql-to-dbsp-compiler/. SQL-emitted programs still call register_materialized_output_zset + register_index separately, which means the integral-sharing optimization doesn't actually kick in for the primary code path. Either wire it up or split this into a follow-up and clearly say so in the commit message.

Smaller concerns (non-blocking)

  • SerCursorFlattened (in crates/adapterlib/src/catalog.rs): nine of its SerCursor methods panic. CSV/JSON/Parquet only call the iteration subset, but anyone adding a new output format or reusing the wrapper later will get a runtime panic instead of a compile error. A focused trait (SerCursorIterate?) capturing only the methods these consumers need would make the API safer; alternatively, add unimplemented!() with an explanation rather than panic! so the failure mode is clearer.
  • SerCursorFlattened doc: "Doesn't return values in order" should make explicit that output ordering changes from key-sorted to (key-major, val within key) — relevant for any consumer that depended on snapshot determinism.
  • The new feature lacks dedicated tests beyond plumbing parameters through existing scaffolds. The integral-sharing optimization in particular should have a test that asserts the same integral is reused across view + index registrations (e.g. by counting integrate operators in the circuit).
  • Commit message subject typo: [adapers] in d058d2bb ("Remove reset_status placeholder.").

) where
index_name: &SqlIdentifier,
view_name: &SqlIdentifier,
key_fields: &[String],

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Breaking change: was &[&SqlIdentifier] on main. The SQL compiler emits &vec!(&SqlIdentifier::from("field"), ...) (see ToRustVisitor.java around line 1083) which won't compile against &[String]. Either revert to taking &[&SqlIdentifier], accept impl IntoIterator<Item = impl AsRef<str>>, or update the SQL compiler emission in this PR.

/// * Doesn't return values in order.
/// * Panics when any operations over values are attempted.
/// * Panics on seek_key_exact and seek_key.
pub struct SerCursorFlattened<'a> {

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nine of the implemented methods panic at runtime (serialize_val_*, val_to_*, seek_key*). That's fine for today's three callers (CSV/JSON/Parquet bulk encoders) but bakes a footgun in for future consumers. Two options:

  1. Define a narrower trait (SerCursorIterate?) with just key_valid/val_valid/step_*/rewind_*/serialize_key*/weight/*_to_avro, and have the encoders accept that. Compile-time guarantee, no panics.
  2. Keep the current shape but use unreachable!("flattened cursors are key-iterated only") so the failure mode is greppable and clearly intentional.

Also: the doc string says "Doesn't return values in order" — please make it explicit that this is a deliberate ordering change (key-major iteration) so future readers know snapshot output order is no longer key-sorted.

stream: Stream<RootCircuit, OrdIndexedZSet<K, V>>,
alias_as_index: Option<SqlIdentifier>,
schema: &str,
key_fields: &[String],

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

register_materialized_output_map and register_materialized_index are introduced here as the headline of the commit, but I can't find any caller for them under sql-to-dbsp-compiler/. Are they exercised by anything beyond the manually-written postgres/delta tests? If the SQL compiler emission isn't being updated in this PR, the integral-sharing optimization that motivates this commit doesn't actually take effect for SQL programs — please clarify the rollout in the commit message and/or add a follow-up issue link.

@mythical-fred mythical-fred left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Prior REQUEST_CHANGES blockers addressed in 6d27256:

  • ToRustVisitor.java now emits the new &[String] key format (appendIndexKeys -> "field".to_string()) instead of the stale &vec!(&SqlIdentifier::from(...)).
  • Integral-sharing optimization is wired up: register_materialized_output_map_persistent for the first index of a MATERIALIZED VIEW (sharing the integral with the view), register_materialized_index_persistent for additional indexes, and the materialized view's own register_materialized_output_zset_persistent is correctly suppressed in that case.
  • CatalogTests.indexTest/indexTest2 exercise both the materialized and non-materialized paths with the expected register_* call counts.

Soft items from prior review still stand (SerCursorFlattened panics on 9 methods, [adapers] subject typo); already noted, not blocking.

Comment thread crates/adapters/src/controller.rs Resolved
@gz
gz force-pushed the delta-flushing branch from 35bcd28 to eedf057 Compare May 14, 2026 04:11
gz and others added 12 commits May 14, 2026 00:42
This adds a per-output-connector `send_snapshot` flag. When set to
`true`, the connector emits a full snapshot of its (materialized) view
the first time it runs, before streaming incremental updates. The
snapshot is delivered exactly once per connector lifetime: a
`snapshot_sent` flag is checkpointed, so resuming from a checkpoint
does not re-send. Modifying the connector across a restart drops its
stats so the next start delivers a fresh snapshot.

Signed-off-by: Gerd Zellweger <[email protected]>
Add `test_egress_send_snapshot` exercising the send_snapshot path on
the streaming egress API, and re-export shared test helpers
(KAFKA_BOOTSTRAP, MINIO_*, env_truthy, required_env, runs_in_ci) from
`tests/__init__.py` so connector status and other tests share one
source of truth.

Signed-off-by: Gerd Zellweger <[email protected]>
Document the `send_snapshot` connector flag in the connectors index
and per-sink pages: the connector emits a full materialized-view
snapshot before streaming incremental updates, persists "already
delivered" across checkpoints, and re-fires after a connector edit.

Signed-off-by: Gerd Zellweger <[email protected]>
`num_parsers` was listed as a string with no default; it is an integer
and defaults to 4.
Previously, output snapshots were only used with ad hoc queries. We now also
use them to send view snapshots to the connector. Since a view can have multiple
indexes, we need to materialize every index, so we can snapshot them. As an
important optimization, connectors that attach to the view without an index
can use the integral associated with any of the indexes to produce both
incremental and snapshot outputs by iterating over values ignoring keys.

This commit adds support for materialized indexes, including the above
optimization.

Updated catalog API:

- `register_materialized_zset` (existing method) - to be used to register
  materialized views _without_ indexes only.
- `register_materialized_output_map` - used to simultaneously register a
  materialized view and an index, which will share a common integral.
- `register_materialized_index` - used with views that have multiple indexes
  to register the second, third, etc. index.

Signed-off-by: Leonid Ryzhyk <[email protected]>
When materializing output streams, we always created the integral
on the host assigned to run all connectors for this output stream.
This is the right thing to do if we need to send a snapshot of the
stream to a connector.

However, it creates imbalance in the pipeline, which needs to
materialize integrals on a single host. For tables with primary keys,
this also causes state duplication as we cannot reuse the PK index
as the snapshot of the collection.

This commit eliminates this duplication: since we don't support
snapshotting input tables, we can spread their integrals across all
hosts and reuse existing integrals when possible.

In addition we cleanup the catalog logic a little:
- Remove unused functions
- Add missing regions

Signed-off-by: Leonid Ryzhyk <[email protected]>
When we resend a snapshot we want to keep statistics of the previous
incarantion (I believe). Also rename these boolean flag to something
I can understand.

Signed-off-by: Gerd Zellweger <[email protected]>
We didn't force a step when the pipeline was paused, this
caused the http egress not to send a snapshot if we connected
to it while a pipeline was paused.

Fixes it and adds a test.

We also added a test confirming the pause-sends-snapshot
behavior for delta-lake. Here it already works correctly
because modifying a connector causes some crap in the
circuit to force a step while paused.

Signed-off-by: Gerd Zellweger <[email protected]>
@gz
gz force-pushed the delta-flushing branch from 73315a0 to f0dffdc Compare May 14, 2026 15:25
@gz
gz enabled auto-merge May 14, 2026 15:26
@gz
gz added this pull request to the merge queue May 14, 2026
@gz
gz removed this pull request from the merge queue due to a manual request May 14, 2026
@gz
gz force-pushed the delta-flushing branch from f0dffdc to 8710cfd Compare May 14, 2026 15:44
@gz
gz enabled auto-merge May 14, 2026 15:44
Signed-off-by: feldera-bot <[email protected]>
@gz
gz added this pull request to the merge queue May 14, 2026
@github-merge-queue
github-merge-queue Bot removed this pull request from the merge queue due to failed status checks May 14, 2026
@gz
gz enabled auto-merge May 14, 2026 18:30
@gz
gz added this pull request to the merge queue May 14, 2026
Merged via the queue into main with commit 14c84f4 May 14, 2026
1 check passed
@gz
gz deleted the delta-flushing branch May 14, 2026 21:02
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

6 participants