Repository navigation
send_snapshot and reset API - #6042
Conversation
|
I fixed some unrelated clippy warnings along the way in avro, ignore them |
There was a problem hiding this comment.
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
resetfor snapshot-and-follow connectors. I suggest that instead ofmodewe have a Boolean attribute likeinitial_snapshot, which only controls behavior on startup. Independent of the value of this attribute, we allowresetfor 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
| } else { | ||
| for batch in snapshot_batches { | ||
| self.status.enqueue_batch(endpoint_id, batch.len()); | ||
| queue.push(BatchQueueEntry { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Please check; I didn't know how to do this
|
additional comments:
I'll add a test for now
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). |
|
The goal with the output connector mode 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 |
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. |
|
ok lets discuss API tomorrow offline |
d8d3df8 to
24c41f8
Compare
2380b8c to
c154dae
Compare
|
@ryzhyk please re-review, update PR description with latest design as discussed |
a9dab7f to
7b7f3d1
Compare
mythical-fred
left a comment
There was a problem hiding this comment.
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
-
Breaking signature change without SQL-compiler update.
register_indexandregister_index_persistentchangedkey_fieldsfrom&[&SqlIdentifier]to&[String]. The Rust test scaffolding is updated, butsql-to-dbsp-compiler/.../ToRustVisitor.javastill emits&vec!(&SqlIdentifier::from("field"), ...)(around line 1083) which isVec<&SqlIdentifier>. After this PR lands, any SQL program withCREATE INDEXwill fail to compile the generated Rust. CI will catch this once it runs, but it has to be fixed before merge. -
The headline optimization isn't wired up to the SQL compiler. Commit message says
register_materialized_output_mapsimultaneously registers a materialized view and an index sharing one integral, andregister_materialized_indexregisters additional indexes. Neither symbol appears anywhere undersql-to-dbsp-compiler/. SQL-emitted programs still callregister_materialized_output_zset+register_indexseparately, 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(incrates/adapterlib/src/catalog.rs): nine of itsSerCursormethods 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, addunimplemented!()with an explanation rather thanpanic!so the failure mode is clearer.SerCursorFlatteneddoc: "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]ind058d2bb("Remove reset_status placeholder.").
| ) where | ||
| index_name: &SqlIdentifier, | ||
| view_name: &SqlIdentifier, | ||
| key_fields: &[String], |
There was a problem hiding this comment.
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> { |
There was a problem hiding this comment.
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:
- Define a narrower trait (
SerCursorIterate?) with justkey_valid/val_valid/step_*/rewind_*/serialize_key*/weight/*_to_avro, and have the encoders accept that. Compile-time guarantee, no panics. - 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], |
There was a problem hiding this comment.
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
left a comment
There was a problem hiding this comment.
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_persistentfor the first index of a MATERIALIZED VIEW (sharing the integral with the view),register_materialized_index_persistentfor additional indexes, and the materialized view's ownregister_materialized_output_zset_persistentis correctly suppressed in that case. CatalogTests.indexTest/indexTest2exercise 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.
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.
Signed-off-by: Gerd Zellweger <[email protected]>
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]>
Signed-off-by: Mihai Budiu <[email protected]>
Signed-off-by: Gerd Zellweger <[email protected]>
Signed-off-by: Gerd Zellweger <[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]>
…Padded Signed-off-by: Gerd Zellweger <[email protected]>
Signed-off-by: feldera-bot <[email protected]>
Signed-off-by: Gerd Zellweger <[email protected]>
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=truemakes 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_snapshotwhen false:
does nothing.
when true:
sets a
Pendingbit inOutputEndpointControlat connector startup, the circuit thread'spush_outputchecks 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 asSentinOutputEndpointControl.A separate initial_snapshot_sent bool is checkpointed (in CheckpointOutputEndpointMetrics) so resuming from a checkpoint skips the replay, while
pipeline_diffclears it across restarts whenever the connector config changes that's how a modified connector re-fires the snapshot.HTTP egress
send_snapshot=trueis 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 withsnapshot: true, then falls back to the normal delta-streaming.Checklist
Breaking Changes?
Ideally, none.