Determine whether we require snapshots from our durable source imports. (For example, these can often be skipped for simple subscribe queries.)
(dataflow: &mut DataflowDesc)
| 456 | /// Determine whether we require snapshots from our durable source imports. |
| 457 | /// (For example, these can often be skipped for simple subscribe queries.) |
| 458 | pub fn optimize_dataflow_snapshot(dataflow: &mut DataflowDesc) -> Result<(), TransformError> { |
| 459 | // For every global id, true iff we need a snapshot for that global ID. |
| 460 | // This is computed bottom-up: subscribes may or may not require a snapshot from their inputs, |
| 461 | // index exports definitely do, and objects-to-build require a snapshot from their inputs if |
| 462 | // either they need to provide a snapshot as output or they may need snapshots internally, eg. to |
| 463 | // compute a join. |
| 464 | let mut downstream_requires_snapshot = BTreeMap::new(); |
| 465 | |
| 466 | for (_id, export) in &dataflow.sink_exports { |
| 467 | *downstream_requires_snapshot |
| 468 | .entry(Id::Global(export.from)) |
| 469 | .or_default() |= export.with_snapshot; |
| 470 | } |
| 471 | for (_id, (export, _typ)) in &dataflow.index_exports { |
| 472 | *downstream_requires_snapshot |
| 473 | .entry(Id::Global(export.on_id)) |
| 474 | .or_default() |= true; |
| 475 | } |
| 476 | for BuildDesc { id: _, plan } in dataflow.objects_to_build.iter().rev() { |
| 477 | // For now, we treat all intermediate nodes as potentially requiring a snapshot. |
| 478 | // Walk the AST, marking anything depended on by a compute object as snapshot-required. |
| 479 | let mut todo = vec![(true, &plan.0)]; |
| 480 | while let Some((requires_snapshot, expr)) = todo.pop() { |
| 481 | match expr { |
| 482 | MirRelationExpr::Get { id, .. } => { |
| 483 | *downstream_requires_snapshot.entry(*id).or_default() |= requires_snapshot; |
| 484 | } |
| 485 | other => { |
| 486 | todo.extend(other.children().rev().map(|c| (true, c))); |
| 487 | } |
| 488 | } |
| 489 | } |
| 490 | } |
| 491 | for (id, import) in &mut dataflow.source_imports { |
| 492 | let with_snapshot = downstream_requires_snapshot |
| 493 | .entry(Id::Global(*id)) |
| 494 | .or_default(); |
| 495 | |
| 496 | // As above, fetch the snapshot if there are any transformations on the raw source data. |
| 497 | // (And we'll always need to check for things like temporal filters, since those allow |
| 498 | // snapshot data to affect diffs at times past the as-of.) |
| 499 | *with_snapshot |= import.desc.arguments.operators.is_some(); |
| 500 | |
| 501 | import.with_snapshot = *with_snapshot; |
| 502 | } |
| 503 | for (_id, import) in &mut dataflow.index_imports { |
| 504 | let with_snapshot = downstream_requires_snapshot |
| 505 | .entry(Id::Global(import.desc.on_id)) |
| 506 | .or_default(); |
| 507 | |
| 508 | import.with_snapshot = *with_snapshot; |
| 509 | } |
| 510 | |
| 511 | Ok(()) |
| 512 | } |
| 513 | |
| 514 | /// Restricts the indexes imported by `dataflow` to only the ones it needs. |
| 515 | /// It also adds to the `DataflowMetainfo` how each index will be used. |