Move `PostgresSourceReader` to `SourceReader` over `SimpleSource` #11946

guswynn · 2022-04-21T00:17:40Z

Ultimately, I think we should not have multiple Source traits. This is the first such change for this: switching Postgres off of a SimpleSource to a full SourceReader. Note that this gives us timestamp bindings for pg for free!

This pr has several subtle parts:

The first commit resolves an outstanding bug: SourceReader assumes that a message at some offset the last such message. Pg has transactions that happen at the same timestamp, so we need to support holding back a cursor to support this. This is quite subtle, and @petrosagg may be interested in looking closely
The second commit adds support for sources that retract (that is, dd diffs that ARENT 1) values. This is to support both replication failure (causing a revert), and pg deletes.
- This pr uses new associated types to avoid unwraps as much as possible, but to keep render_source readable, forces a diff in the value in DecodeResult.
- We add unreachable! for not-possible diff values in envelopes that don't support esoteric diffs (namely, things other than None. Its not 100% clear to me if KeyEnvelope::None is the only supported SourceEnvelope::None, but the coden as-written does not deal with
The third commit actually moves pg to a SourceReader. As much code as possible is preserved, and I opted for the get_next_message route over next, as we can transform all sources at the same time
The final commit ensures that pg sources are not considered monotonic. This appears to have already been an outstanding bug

Motivation

This PR adds a known-desirable feature.

#11899

Tips for reviewer

I recommend reviewing the commits in this pr separately.

Testing

This PR has adequate test coverage / QA involvement has been duly considered.

Release notes

None

src/dataflow-types/src/types.rs

cjubb39

gave a quick look-through (didn't get to the actual logic of the postgres source reader impl so it's possible that some suggestions are redundant. In that case sorry!)

src/storage/src/source/mod.rs

src/storage/src/render/debezium.rs

src/storage/src/render/sources.rs

cjubb39 · 2022-05-18T20:56:41Z

src/storage/src/source/mod.rs

@@ -478,9 +527,15 @@ pub struct SourceMessage<Key, Value> {
    /// Headers, if the source is configured to pass them along. If it is, but there are none, it
    /// passes `Some([])`
    pub headers: Option<Vec<(String, Option<Vec<u8>>)>>,
+
+    /// Allow sources to optionally output a specific differential
+    /// `diff` value. Defaults to `+1`.


I'm curious why you've structured it as an arbitrary Diff type rather than either:

Option<mz_repr::Diff>

mz_repr::Diff and having everything other than postgres set it to 1 (since they can authoritatively say that)

I don't follow (perhaps out of ignorance!) when we'd need to support a non-mz_repr::Diff in this field?

I had it Option<mz_repr::Diff> or something similar but I found that it made it unclear in render_source what was happening. with the associated type we can ensure that ONLY
SourceType::Row ends up having type Diff = Diff, where as other types always have type Diff = () .

in the future this also enables us to generic-ize DecodeResult and cleanup render_source even more

so in short: yes, we want other diff types, namely () to clearly communicate when a source wants to use the default behavior, as opposed to all those sources needing to implement that default themselves

with the associated type we can ensure that ONLY SourceType::Row ends up having type Diff = Diff, where as other types always have type Diff = ()

sorry -- I don't think I follow what value this provides

all those sources needing to implement that default themselves

but the default is just putting a 1 instead of a () yeah? And currently it's a () that somewhere else is mapped to a 1?

I think my concern is that it seems like extra complexity "what is a specific_diff as opposed to just a regular diff?" that doesn't practically buy us much -- but I am happy to say that it's possible I'm just not following the benefits

with a fixed type, either diff: i64 or diff: Option<i64>, the SourceReader impl is required to provide the "default" to us. If kafka's SourceReader gives us 2 over None or 1, we can't really tell if thats wrong or not, we have to assume its right

With this, we declare, at the top of the trait impl, that the kafka source reader can produce only 1 value as a diff (in this case (), but we could make an empty struct call Default or whatever, preventing any complexity or thought in the decision when producing the SourceMessage inside the next method.

I found this to be much cleaner and easier to reason about, and less fragile. You are right that it doesn't buy us much, but I am inclined to encode invariants in the type system as much as possible, without making stuff unreadable. I found this to be more readable than the previous version (you may be able to find a commit on this pr from earlier) that didn't do it this way

If kafka's SourceReader gives us 2 over None or 1, we can't really tell if thats wrong or not, we have to assume its right

Isn't KafkaSourceReader the authoritative source for what is correct for a kakfa source though? Just like the PostgresSourceReader is the authoritative source for the diff field for a postgres source. We're not supposing what we should protect against postgres being wrong. I think that's the thing I'm not understanding.

Hmm I guess this setup is analogous to parametrizing over a trait implemented by two enums: enum AppendOnly { Append } and enum AppendOrDelete { Append, Delete } -- which I do very clearly see the value in soperhaps it's the naming tripping me up? I think writing those enums out is more explicit (and maybe even more clear but I was the one who came up with it so grain of salt...) but if you don't want to do that, spitballing here: what about just renaming specific_diff -> update_type?

@cjubb39 KafkaSourceReader isnt really the authoritative source: it can be composed with envelopes that may or may not support non-1 diffs; so in some sense, the caller in render::sources::render_source is authoritative (as it should be, it is the place where SourceReaders are correctly instantiated

we can definitely change the name! update_type is fine, what about type MessageDiff and message_diff?

src/storage/src/source/postgres.rs

cjubb39 · 2022-05-23T20:08:24Z

src/storage/src/source/mod.rs

@@ -478,9 +527,15 @@ pub struct SourceMessage<Key, Value> {
    /// Headers, if the source is configured to pass them along. If it is, but there are none, it
    /// passes `Some([])`
    pub headers: Option<Vec<(String, Option<Vec<u8>>)>>,
+
+    /// Allow sources to optionally output a specific differential
+    /// `diff` value. Defaults to `+1`.


with the associated type we can ensure that ONLY SourceType::Row ends up having type Diff = Diff, where as other types always have type Diff = ()

sorry -- I don't think I follow what value this provides

all those sources needing to implement that default themselves

but the default is just putting a 1 instead of a () yeah? And currently it's a () that somewhere else is mapped to a 1?

I think my concern is that it seems like extra complexity "what is a specific_diff as opposed to just a regular diff?" that doesn't practically buy us much -- but I am happy to say that it's possible I'm just not following the benefits

src/storage/src/source/mod.rs

src/storage/src/source/postgres.rs

src/dataflow-types/src/types.rs

guswynn · 2022-05-24T22:00:18Z

@cjubb39 I believe I resolved all your comments!

cjubb39

please update the comment on the MzOffset impl below before merging! will leave the other one up to you

cjubb39 · 2022-05-25T15:21:50Z

src/dataflow-types/src/types.rs

@@ -1483,6 +1483,16 @@ pub mod sources {
        }
    }

+    /// Convert from KafkaOffset to MzOffset (1-indexed)


update comment kafka -> pg

src/storage/src/source/mod.rs

cjubb39 · 2022-05-25T15:29:38Z

Thanks for doing this change! Happy to moving to have one impl for sources!

offset

guswynn force-pushed the pgsr branch 10 times, most recently from 79f16a5 to efaa025 Compare April 28, 2022 16:34

guswynn marked this pull request as ready for review April 28, 2022 16:48

guswynn changed the title ~~pg sourcereader~~ Move PostgresSourceReader to SourceReader over SimpleSource Apr 28, 2022

guswynn requested review from petrosagg and a team April 28, 2022 16:55

frankmcsherry reviewed May 3, 2022

View reviewed changes

src/dataflow-types/src/types.rs Show resolved Hide resolved

guswynn force-pushed the pgsr branch 3 times, most recently from 2905cd2 to 174c8c6 Compare May 10, 2022 23:19

guswynn mentioned this pull request May 11, 2022

Disambiguate user-provided ENVELOPEs and internal SourceEnvelopes #12413

Closed

guswynn force-pushed the pgsr branch 3 times, most recently from d144230 to 2eeb57b Compare May 18, 2022 18:56

cjubb39 reviewed May 18, 2022

View reviewed changes

guswynn force-pushed the pgsr branch from 2eeb57b to 3a974eb Compare May 18, 2022 21:19

guswynn mentioned this pull request May 19, 2022

Don't buffer postgres sources messages as aggressively #12557

Closed

guswynn force-pushed the pgsr branch from 3a974eb to f8456bd Compare May 23, 2022 18:36

cjubb39 reviewed May 23, 2022

View reviewed changes

guswynn force-pushed the pgsr branch 3 times, most recently from b6da316 to 8624835 Compare May 24, 2022 22:00

cjubb39 approved these changes May 25, 2022

View reviewed changes

guswynn added 6 commits May 25, 2022 13:23

add ability to delay timestamping to collect multiple messages at same

04c0309

offset

add support for explicit diff's from sources in SourceEnvelope::None

2022736

move postgres SimpleSource to SourceReader

f1f0969

do not consider pg sources monotonic

9f53c5e

reduce some indentation in render_source

c028bd6

add pglsn->mzoffset from impl

5715fb3

guswynn force-pushed the pgsr branch from 8624835 to 5715fb3 Compare May 25, 2022 20:25

guswynn enabled auto-merge (rebase) May 25, 2022 20:25

guswynn merged commit 224de26 into MaterializeInc:main May 25, 2022

guswynn deleted the pgsr branch May 25, 2022 22:19

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Move `PostgresSourceReader` to `SourceReader` over `SimpleSource` #11946

Move `PostgresSourceReader` to `SourceReader` over `SimpleSource` #11946

guswynn commented Apr 21, 2022 •

edited

cjubb39 left a comment

cjubb39 May 18, 2022

guswynn May 18, 2022

guswynn May 18, 2022

cjubb39 May 23, 2022

guswynn May 23, 2022

cjubb39 May 23, 2022

guswynn May 24, 2022

cjubb39 May 23, 2022

guswynn commented May 24, 2022

cjubb39 left a comment

cjubb39 May 25, 2022

cjubb39 commented May 25, 2022

Move PostgresSourceReader to SourceReader over SimpleSource #11946

Move PostgresSourceReader to SourceReader over SimpleSource #11946

Conversation

guswynn commented Apr 21, 2022 • edited

Motivation

Tips for reviewer

Testing

Release notes

cjubb39 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guswynn commented May 24, 2022

cjubb39 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cjubb39 commented May 25, 2022

Move `PostgresSourceReader` to `SourceReader` over `SimpleSource` #11946

Move `PostgresSourceReader` to `SourceReader` over `SimpleSource` #11946

guswynn commented Apr 21, 2022 •

edited