Attempting to push batches into TAIL. by cirego · Pull Request #1 · frankmcsherry/materialize

cirego · 2020-07-31T23:51:55Z

Attempting to push batches into TAIL.

Currently fails to compile due to my not quite understanding how types
are converted across Timely boundaries. Just moving the devolution code
into sink::tail gives the following error:

[E0599]: no method named `flat_map` found for enum `timely_communication::message::RefOrMut<'_, std::vec::Vec<std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>>` in the current scope
--> src/dataflow/src/sink/tail.rs:33:32
|
33  |               let rows = batches.flat_map(move |batch| {
|                                  ^^^^^^^^ method not found in `timely_communication::message::RefOrMut<'_, std::vec::Vec<std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>>`
|
::: /home/chris/.cargo/git/checkouts/timely-dataflow-4c0cc365061cd263/76e6049/communication/src/message.rs:9:1
|
9   |   pub enum RefOrMut<'a, T> where T: 'a {
|   ------------------------------------ doesn't satisfy `_: std::iter::Iterator`
|
= note: the method `flat_map` exists but the following trait bounds were not satisfied:
`timely_communication::message::RefOrMut<'_, std::vec::Vec<std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>>: std::iter::Iterator`
which is required by `&mut timely_communication::message::RefOrMut<'_, std::vec::Vec<std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>>: std::iter::Iterator`
`std::vec::Vec<std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>: std::iter::Iterator`
which is required by `&mut std::vec::Vec<std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>: std::iter::Iterator`
`[std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>]: std::iter::Iterator`
which is required by `&mut [std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>]: std::iter::Iterator`

I have tried a few iterations on this. Namely:

Diff 1: Call iter() on batches. Compiler says:

error[E0599]: no method named `iter` found for struct `std::iter::FlatMap<std::slice::Iter<'_, std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>, std::iter::FlatMap<std::vec::IntoIter<((repr::row::Row, ()), std::vec::Vec<(u64, isize)>)>, std::iter::Map<std::vec::IntoIter<(u64, isize)>, [closure@src/dataflow/src/sink/tail.rs:39:42: 39:75 row:_]>, [closure@src/dataflow/src/sink/tail.rs:38:45: 40:18]>, [closure@src/dataflow/src/sink/tail.rs:35:48: 41:14]>` in the current scope
--> src/dataflow/src/sink/tail.rs:44:43
|
44 |             for (row, time, diff) in rows.iter() {
|                                           ^^^^ method not found in `std::iter::FlatMap<std::slice::Iter<'_, std::rc::Rc<differential_dataflow::trace::implementations::ord::OrdKeyBatch<repr::row::Row, u64, isize>>>, std::iter::FlatMap<std::vec::IntoIter<((repr::row::Row, ()), std::vec::Vec<(u64, isize)>)>, std::iter::Map<std::vec::IntoIter<(u64, isize)>, [closure@src/dataflow/src/sink/tail.rs:39:42: 39:75 row:_]>, [closure@src/dataflow/src/sink/tail.rs:38:45: 40:18]>, [closure@src/dataflow/src/sink/tail.rs:35:48: 41:14]>`

Diff 2: Remove the call to iter() on rows. Compiler says:

error[E0308]: mismatched types
--> src/dataflow/src/sink/tail.rs:46:50
|
46 |                     connector.frontier.less_than(time)
|                                                  ^^^^
|                                                  |
|                                                  expected `&u64`, found `u64`
|                                                  help: consider borrowing here: `&time`

error[E0308]: mismatched types
--> src/dataflow/src/sink/tail.rs:48:51
|
48 |                     connector.frontier.less_equal(time)
|                                                   ^^^^
|                                                   |
|                                                   expected `&u64`, found `u64`
|                                                   help: consider borrowing here: `&time`

error[E0614]: type `u64` cannot be dereferenced
--> src/dataflow/src/sink/tail.rs:53:36
|
53 |                         timestamp: *time,
|                                    ^^^^^

error[E0614]: type `isize` cannot be dereferenced
--> src/dataflow/src/sink/tail.rs:54:31
|
54 |                         diff: *diff,
|                               ^^^^^

Diff 3: "Fix" those errors by adding "&" and removing "*". Compiler says:

error[E0507]: cannot move out of `connector`, a captured variable in an `FnMut` closure
--> src/dataflow/src/sink/tail.rs:34:24
|
28 |     connector: TailSinkConnector,
|     --------- captured outer variable
...
34 |         input.for_each(move |_, batches| {
|                        ^^^^^^^^^^^^^^^^^ move out of `connector` occurs here
...
45 |                 let should_emit = if connector.strict {
|                                      ---------
|                                      |
|                                      move occurs because `connector` has type `dataflow_types::types::TailSinkConnector`, which does not implement the `Copy` trait
|                                      move occurs due to use in closure

error[E0507]: cannot move out of `tx`, a captured variable in an `FnMut` closure
--> src/dataflow/src/sink/tail.rs:34:24
|
32 |     let mut tx = block_on(connector.tx.connect()).expect("tail transmitter failed");
|         ------ captured outer variable
33 |     stream.sink(Pipeline, &format!("tail-{}", id), move |input| {
34 |         input.for_each(move |_, batches| {
|                        ^^^^^^^^^^^^^^^^^ move out of `tx` occurs here
...
63 |             block_on(tx.send(results)).expect("tail send failed");
|                      --
|                      |
|                      move occurs because `tx` has type `std::pin::Pin<std::boxed::Box<dyn futures_sink::Sink<std::vec::Vec<dataflow_types::types::Update>, Error = comm::error::Error> + std::marker::Send>>`, which does not implement the `Copy` trait
|                      move occurs due to use in closure

And now I'm pretty sure I'm just going down the wrong path, not
understanding something simple from the first part of the patch.

cirego · 2020-07-31T23:57:05Z

@frankmcsherry, I'd appreciate you taking a look and letting me know where I went wrong!

frankmcsherry · 2020-08-01T00:44:42Z

What I think Rust is complaining about is the move keyword you have in the for_each call. This keyword tells Rust that you want the closure to take ownership of the instances it closes over (specifically, connector and tx). However, it can't do this because the outer closure needs to be in a legit state the next time it is called too, which Rust tries to communicate with

cannot move out of `tx`, a captured variable in an `FnMut` closure

I think you might be fine removing the move on line 34, as I think the things you use in the closure can be used by reference, and then the for_each returns; it doesn't need ownership that I can tell. A different way to think about it is that you are happy to provide ownership of &connector and &mut tx, which is roughly what leaving off the move would attempt. If that fails, you can create actual let connector_ref = &connector; bindings and move those in, but .. I'd start with ditching the move and see what happens.

This is the first step towards having TAIL output updates order by timestamp.

cirego · 2020-08-01T05:00:38Z

What I think Rust is complaining about is the move keyword you have in the for_each call. This keyword tells Rust that you want the closure to take ownership of the instances it closes over (specifically, connector and tx). However, it can't do this because the outer closure needs to be in a legit state the next time it is called too, which Rust tries to communicate with
cannot move out of `tx`, a captured variable in an `FnMut` closure
I think you might be fine removing the move on line 34, as I think the things you use in the closure can be used by reference, and then the for_each returns; it doesn't need ownership that I can tell. A different way to think about it is that you are happy to provide ownership of &connector and &mut tx, which is roughly what leaving off the move would attempt. If that fails, you can create actual let connector_ref = &connector; bindings and move those in, but .. I'd start with ditching the move and see what happens.

Yep! Looks like the input.for_each(move |_, batches| line was the culprit. Looking at the patch, it would appear that I added the move too... not sure why I did that. Thanks for the quick diagnosis Frank!

Still looks like there is room for improvement in the loop construction via removing call to `to_vec` and using another iteration method.

cirego · 2020-11-25T19:50:12Z

I'm going to close this in favor of the work being done by Matt.

Example of crashing query is: select (select a) from (select 1 as a); where we end up with self-referencing Map like the following: Raw Plan --------------------- %0 = + | Constant () + | Map 1, select(%1)+ | | + | | %1 = + | | | Constant () + | | | Map #^0 + | | + | Project (#1) +

…aterializeInc#9197) Fix an error in `splice_parameters` in handling nested queries. Before this fix to the depth, the following query plan was generated: ``` | | | | | | %3 = + | | | | | | | Get mz_catalog.mz_types (s4027) + | | | | | | | Filter (#^3 = "bool") + | | | | | | | Map oidtoregtype(#^1) + | | | | | | | Project (#4) ``` whereby the correct plan is: ``` materialize=> explain raw plan for select 'bool'::regtype::text; Raw Plan ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- %0 = + | Constant () + | Map select(%1) + | | + | | %1 = + | | | Get mz_catalog.mz_types (s4027) + | | | Filter (#1 = regtypetooid(select(%2))) + | | | | + | | | | %2 = + | | | | | Constant () + | | | | | Map if isnull("bool") then {null} els {if ("bool" ~ "^\d+$") then {i32toregtype(strtoi32("bool"))} els {error_if_null(select(%3), (("type \"" || "bool") || "\" does not exist"))}}+ | | | | | | + | | | | | | %3 = + | | | | | | | Get mz_catalog.mz_types (s4027) + | | | | | | | Filter (#3 = "bool") + | | | | | | | Map oidtoregtype(#1) + | | | | | | | Project (#4) + | | | | | | + | | | | + | | | Project (#3) + | | ```

Improve key information for Filters. Use equality predicates in the filter to refine and expand the keys reported by the input operator. If `[0 1]` is an input key and there `#0 = #1` exists as a predicate, we should present both `[0]` and `[1]` as keys of the output. Also, if there is a key involving X column and an equality between X and another column Y, a variant of that key with Y instead of X should be presented as a key of the output.

The QGM tests were panicking on the following query select case when x.a > x.b then x.c else lpad(x.d, 4, 'ab') end from (select 10 as a, 5 as b, 'world' as c, 'realm' as d) x with the error "unreachable: column not found." The plan in question is | Constant () | Map null | Project (#0, #0, #0) | Map if isnull(#0) then {coalesce(#1, #2, "hello")} els {substr(#1, i32toi64(3))}, mz_logical_timestamp() | Project (#3, #4) where it previously was: | Constant () | Map null, null, null | Map if isnull(#0) then {coalesce(#1, #2, "hello")} els {substr(#1, i32toi64(3))}, mz_logical_timestamp() | Project (#3, #4) Handle this situation by always wrapping the input to a map in a fresh Select box, with the same justification as applies to Filter nodes.

This commit adds a design doc for transactional consistency. It is in service of correctness property #1, which is copied below: Transactions in Materialize are strictly serializable with respect to the operations that occur inside of Materialize. Operations include: - SELECT, INSERT, UPDATE, and DELETE statements (but not TAIL) - Materialize initiated acknowledgements of upstream sources (e.g., the commit of an offset by a Kafka source, the acknowledge of an LSN by a PostgresSQL source, etcetera) However, this design document only focuses on the first bullet point, and leaves the second bullet point for a follow up design document. Works towards resolved #11631

Extract inner function for build_bucketed_stage

…lizeInc#25999) This PR changes the Fivetran Destination to mark primary key columns as `NOT NULL`. Primary keys are not allowed to be `NULL` and adding this constraint allows us to plan queries that are currently a three-way join as a two-way join. ``` CREATE TABLE logs (id text, ts int); CREATE TABLE logs_temp (id text, ts int); EXPLAIN SELECT FROM logs WHERE (id, ts) IN (SELECT id, ts FROM logs_temp); Explained Query: Project () // { arity: 0 } Join on=(#0{id} = #2{id} = #4{id} AND #1{ts} = #3{ts} = #5{ts}) type=delta // { arity: 6 } ArrangeBy keys=[[#0{id}, #1{ts}]] // { arity: 2 } ReadStorage materialize.parker_timmerman.logs // { arity: 2 } ArrangeBy keys=[[#0{id}, #1{ts}]] // { arity: 2 } Distinct project=[#0{id}, #1{ts}] // { arity: 2 } Filter (#0{id}) IS NOT NULL AND (#1{ts}) IS NOT NULL // { arity: 2 } ReadStorage materialize.parker_timmerman.logs // { arity: 2 } ArrangeBy keys=[[#0{id}, #1{ts}]] // { arity: 2 } Distinct project=[#0{id}, #1{ts}] // { arity: 2 } ReadStorage materialize.parker_timmerman.logs_temp // { arity: 2 } ``` Now with the `NOT NULL` constraints ``` CREATE TABLE logs (id text NOT NULL, ts int NOT NULL); CREATE TABLE logs_temp (id text NOT NULL, ts int NOT NULL); EXPLAIN SELECT FROM logs WHERE (id, ts) IN (SELECT id, ts FROM logs_temp); Explained Query: Project () // { arity: 0 } Join on=(#0{id} = #2{id} AND #1{ts} = #3{ts}) type=differential // { arity: 4 } ArrangeBy keys=[[#0{id}, #1{ts}]] // { arity: 2 } ReadStorage materialize.parker_timmerman.logs // { arity: 2 } ArrangeBy keys=[[#0{id}, #1{ts}]] // { arity: 2 } Distinct project=[#0{id}, #1{ts}] // { arity: 2 } ReadStorage materialize.parker_timmerman.logs_temp // { arity: 2 } ``` Note: Frank currently has a draft PR that allows the optimizer to move the `Filter` in the first plan so even without these constraints we can plan the query as a two-way join. I put up this PR because making the change for the Fivetran Destination was faster. ### Motivation Improve the performance of the destination, which should reduce the amount of time the global write lock gets held ### Checklist - [ ] This PR has adequate test coverage / QA involvement has been duly considered. - [ ] This PR has an associated up-to-date [design doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md), is a design doc ([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)), or is sufficiently small to not require a design.  - [ ] If this PR evolves [an existing `$T ⇔ Proto$T` mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md) (possibly in a backwards-incompatible way), then it is tagged with a `T-proto` label. - [ ] If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label ([example](https://github.com/MaterializeInc/cloud/pull/5021)).  - [x] This PR includes the following [user-facing behavior changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note): - Improve the performance of the Fivetran Destination

…zeInc#28333) The action of `EquivalencePropagation` on `LetRec` seems to be wrong, descending into `values` (which is fine) presenting the `outer_equivalences` that will be applied to the `LetRec` (which is not fine). These outer equivalences apply to the `LetRec`, and can only be propagated to the `body` of the `LetRec`. For example, if `#1 > 3` is a thing that must be true of the `LetRec`, it must be true of `body` but needn't be true of any binding, because .. they have totally unrelated columns. A more advanced analysis could present more information about what equivalences will be applied, but it seems like it would need to be a top-down analogue of the bottom-up `Analysis`. I.e. it would need to iterate around `LetRec` until fixed point reached, all that stuff. Not in the cards at the moment, so just need to fix the current transform. ### Motivation  ### Tips for reviewer  ### Checklist - [ ] This PR has adequate test coverage / QA involvement has been duly considered. ([trigger-ci for additional test/nightly runs](https://trigger-ci.dev.materialize.com/)) - [ ] This PR has an associated up-to-date [design doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md), is a design doc ([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)), or is sufficiently small to not require a design.  - [ ] If this PR evolves [an existing `$T ⇔ Proto$T` mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md) (possibly in a backwards-incompatible way), then it is tagged with a `T-proto` label. - [ ] If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label ([example](MaterializeInc/cloud#5021)).  - [ ] This PR includes the following [user-facing behavior changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note): -

…Inc#30134) This PR improves the performance of `Equivalences::union`, and potentially alters its outputs (on non-canonical inputs). Superficially, there is now a method `union_many` that accepts multiple equivalences to merge, and which `union` just calls. This allows our variadic `Union` stages to call this once, rather than a sequence of rebuilds. This also reduces a fair bit of cloning. I did not measure how impactful this step is in isolation. More importantly, the implementation is changed from quadratic all-pairs set intersection to a map based approach that intersects the keys of two maps, and builds equivalence classes from the results. If an expression maps to `foo` and `bar` under the two maps, then a class `(foo, bar)` is created, and assigned a new class identifier. This new identifier is used as `foo` in the next iteration, with the next input equivalence class, repeating through each of the classes. This ends up being linear time, which is great. This does not improve the inherent limitation that we are looking for syntactic equality, so `f(#0)` and `f(#1)` will not match, even if in the first input `#0 == #1`. We might need something more like e-graphs (or our `project` implementation, which is a simplified form of intersection) to get this. This PR has the potential to *change* the output when the input is not an equivalence relation. It would previous search through non-equivalence class `self.classes`, firing on hits in those lists. When `classes` describes an equivalence relation we'll get the same output, but when they are e.g. not closed under transitivity, .. I don't know what output we are producing. ### Motivation  ### Tips for reviewer  ### Checklist - [ ] This PR has adequate test coverage / QA involvement has been duly considered. ([trigger-ci for additional test/nightly runs](https://trigger-ci.dev.materialize.com/)) - [ ] This PR has an associated up-to-date [design doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md), is a design doc ([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)), or is sufficiently small to not require a design.  - [ ] If this PR evolves [an existing `$T ⇔ Proto$T` mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md) (possibly in a backwards-incompatible way), then it is tagged with a `T-proto` label. - [ ] If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label ([example](MaterializeInc/cloud#5021)).  - [ ] If this PR includes major [user-facing behavior changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note), I have pinged the relevant PM to schedule a changelog post.

Currently, within `LetRec` blocks we only perform relation CSE within each binding. We do not optimize across bindings, nor do we take advantage of bindings outside the `LetRec` region. This PR addresses this with extended code for the `LetRec` case that both references `LetRec`-external bindings, and shares across `LetRec`-internal bindings. There are some potential gotchas, documented in the code: 1. References to `LetRec` bindings should not be equated across the moment the binding is refreshed. Each binding has a "before" identifier and an "after" identifier, and this is sufficient to avoid unintentionally equating the two. 2. The `body` should not borrow much from within the scope. Although all terms are available for use, the current suffix of the optimizer pipeline goes sideways if this happens. Concretely, `JoinImplementation` ends up surprised about the location of arrangements it thought would certainly exist, but do not. There are some unspecified requirements of `RelationCSE`, and the careful treatment of `body` seems to be necessary (unknown if sufficient). 3. There was some initial concern that various iteration limits would complicate the applicability of CSE, but it seems that all new terms can be treated as if having no iteration limits, and things "just work out" because two terms are equated only if they are exact syntactic matches, which happens only when they depend on the same terms, in which case those terms will have the same limits. As a demonstration of what can change, these two WMR blocks are merged, and the necessary arrangement of `t` by `b` is shared between the two terms. On `main` that sharing cannot happen, and two identical copies are made. ```sql create table t(a int, b int); create view foo as with mutually recursive foo0 (a int, b int) as ( select t.a, foo0.b from t, foo0 where t.b = foo0.a union select * from t where t.a > 3 ) select * from foo0; create view bar as with mutually recursive bar0 (a int, b int) as ( select t.a, bar0.b from t, bar0 where t.b = bar0.a union select * from t where t.a > 2 ) select * from bar0; -- demonstrates re-use of `t` arranged by `b`. explain select * from foo union select * from bar; ``` this results in a shared term `l0` seen below ``` materialize=> explain select * from foo union select * from bar; Optimized Plan ------------------------------------------------- Explained Query: + Return + Distinct project=[#0, #1] + Union + Get l1 + Get l2 + With Mutually Recursive + cte l2 = + Distinct project=[#0, #1] + Union + Project (#0, #3) + Join on=(#1 = #2) type=differential+ Get l0 + ArrangeBy keys=[[#0]] + Filter (#0) IS NOT NULL + Get l2 + Filter (#0 > 2) + ReadStorage materialize.public.t + cte l1 = + Distinct project=[#0, #1] + Union + Project (#0, #3) + Join on=(#1 = #2) type=differential+ Get l0 + ArrangeBy keys=[[#0]] + Filter (#0) IS NOT NULL + Get l1 + Filter (#0 > 3) + ReadStorage materialize.public.t + cte l0 = + ArrangeBy keys=[[#1]] + Filter (#1) IS NOT NULL + ReadStorage materialize.public.t + + Source materialize.public.t + + Target cluster: quickstart + (1 row) materialize=> ``` There is a missed opportunity for further sharing, where loop-invariant structure is common to both `values` and `body`. We could perform more work to hoist all bindings that do not depend on `values` out of `body`, and install those bindings before considering `values`. In some sense, the CSE must be done "outside the scope" in order to avoid the peril of `body` depending on terms that will later be made unavailable. ### Motivation  ### Tips for reviewer  ### Checklist - [ ] This PR has adequate test coverage / QA involvement has been duly considered. ([trigger-ci for additional test/nightly runs](https://trigger-ci.dev.materialize.com/)) - [ ] This PR has an associated up-to-date [design doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md), is a design doc ([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)), or is sufficiently small to not require a design.  - [ ] If this PR evolves [an existing `$T ⇔ Proto$T` mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md) (possibly in a backwards-incompatible way), then it is tagged with a `T-proto` label. - [ ] If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label ([example](MaterializeInc/cloud#5021)).  - [ ] If this PR includes major [user-facing behavior changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note), I have pinged the relevant PM to schedule a changelog post.

This adds some of the work from MaterializeInc#28874 on identifying equivalences that can move through `Reduce` aggregates. Specifically, for aggregates like `Min` and `Max` (and ilk), equivalences involving the input expressions may remain true about the output. However, some care is required because even though it may be that `#0 = #1` we should not conclude that `Min(#0) = Max(#1)`. The code details the reasoning, and how one should re-imagine multi-aggregate reduces as joins of single-aggregate reduces. ### Motivation  ### Tips for reviewer  ### Checklist - [ ] This PR has adequate test coverage / QA involvement has been duly considered. ([trigger-ci for additional test/nightly runs](https://trigger-ci.dev.materialize.com/)) - [ ] This PR has an associated up-to-date [design doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md), is a design doc ([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)), or is sufficiently small to not require a design.  - [ ] If this PR evolves [an existing `$T ⇔ Proto$T` mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md) (possibly in a backwards-incompatible way), then it is tagged with a `T-proto` label. - [ ] If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label ([example](MaterializeInc/cloud#5021)).  - [ ] If this PR includes major [user-facing behavior changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note), I have pinged the relevant PM to schedule a changelog post.

…terializeInc#36739) ### Motivation Closes MaterializeInc/database-issues#9376 ### Description Makes default `EXPLAIN` give substantially cleaner output, more like `EXPLAIN OPTIMIZED PLAN`, with a particular eye to identifying cross joins. Old janky: ``` →Differential Join %0 » %1 Join stage 0 in %1 project=(#2) map=((#0{v} + #1{v})) ``` New hotness: ``` →Differential Cross Join %0[×] » %1[×] after %1: Project: #2 Map: (#0{v} + #1{v}) →Arranged l0 →Arranged l0 ``` We call a join a cross join if _any_ part of it involves a cross join: ``` query T multiline EXPLAIN WITH(humanized expressions) SELECT a, b, c, d, e, f FROM t, u, v WHERE a = c ---- Explained Query: →Differential Cross Join %0[#0{a}] » %1[#0{c}] » %2[×] after %1: Filter: (#0{a}) IS NOT NULL →Arranged materialize.public.t →Arranged materialize.public.u →Arrange (empty key) →Arranged materialize.public.v ``` ### Verification Rewrote `default.slt` and added cross join cases.

projection_thinning (transform #6): operators produce only columns consumers demand. Consumers process before producers — the root with all columns demanded, then bindings last-to-first against the union of use-site demands. thin() rewrites in place, returns the produced original-column list, and callers remap through it; exact narrowing is forced only where layouts must agree (binding bodies, Union branches). Use sites become Project-over-Get markers fixed up once layouts are final. The only deletions are Map scalars and Reduce aggregates no consumer observes (only error possibilities can disappear, never rows) and Constant columns. LetRec stays opaque: anything used inside one is demanded in full. The pipeline re-runs linear_fuse afterwards to restore the M-F-P normal form. Two refinements earned from the LEFT JOIN probe: - join_flatten emission rewrites region outputs to each equivalence class's least column member (datum-equal on every surviving row), so duplicated join outputs become undemanded and thin away; - a binding whose body ends in a Project with duplicate outputs stores the deduplicated layout, and use sites re-duplicate (markers compose the duplication; bare full-demand Gets get wrapped during fixup). The LEFT JOIN cte now stores 3 columns with Project (#0,#1,#0,#2) at the use site — main's shape exactly. Sweep-010 (12 files): pass=4 plandiff=8 resultdiff=0 panic=0; joins.slt failing EXPLAINs 6 -> 3. Co-Authored-By: Claude Fable 5 <[email protected]>

cirego force-pushed the sink_batches2 branch 2 times, most recently from f53e49c to 7fd64a2 Compare July 31, 2020 23:56

Push batches into TAIL

9c3b457

This is the first step towards having TAIL output updates order by timestamp.

cirego force-pushed the sink_batches2 branch from 7fd64a2 to 9c3b457 Compare August 1, 2020 04:59

Produce results directly from batches

e5cd272

Still looks like there is room for improvement in the loop construction via removing call to `to_vec` and using another iteration method.

cirego closed this Nov 25, 2020

cirego deleted the sink_batches2 branch March 10, 2021 20:30

benesch restored the sink_batches2 branch November 23, 2021 13:36

benesch deleted the sink_batches2 branch December 7, 2021 16:34

vmarcos added a commit that referenced this pull request Mar 21, 2023

Merge pull request #1 from antiguru/validating_minmaxes_inner

5a8ea16

Extract inner function for build_bucketed_stage

Navigation Menu

Repository navigation

Attempting to push batches into TAIL. - #1

Attempting to push batches into TAIL.#1
cirego wants to merge 2 commits into
frankmcsherry:sink_batchesfrom
MaterializeInc:sink_batches2

cirego commented Jul 31, 2020 •

edited

Loading

Uh oh!

cirego commented Jul 31, 2020

Uh oh!

frankmcsherry commented Aug 1, 2020 •

edited

Loading

Uh oh!

cirego commented Aug 1, 2020

Uh oh!

cirego commented Nov 25, 2020

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

cirego commented Jul 31, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cirego commented Jul 31, 2020

Uh oh!

frankmcsherry commented Aug 1, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cirego commented Aug 1, 2020

Uh oh!

cirego commented Nov 25, 2020

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

cirego commented Jul 31, 2020 •

edited

Loading

frankmcsherry commented Aug 1, 2020 •

edited

Loading