add incremental to jdbc jooq source (and postgres) #1172

cgardens · 2020-12-02T22:23:55Z

What

Adds support for incremental syncs to the ~~jdbc~~ jooq-compatible sources. The goal of this PR is to set the base for all jdbc source and to be able to ship the postgres source using incremental.

How

Add JdbcStateManager to track original and new cursor states as we progress. This should be reusable for all database style source I think.
Add an iterator (StateDecoratingIterator) that handles tracking the max cursor value and emitting the state message at the end of syncing each stream. Note: 1 state message per stream, even if no records are replicated for the stream.
Decided to take a shortcut here and just do jooq databases for this PR.
- Down side of this is that i just copy and pasted the existing AbstractJdbcSource for now.

Decisions to make

Is separating the JdbcSource and JooqSource necessary? The hard thing™ here is constructing WHERE clauses with the correct syntax to do the comparison we need for the cursor based on the type of the column. Jooq seems to make that easier (though as you go through the PR, you'll see it is not perfect). What I plan to do next is see what a just jdbc version of this looks like. If it is good enough, I will just get rid of all jooq from the database sources and they can all use one. That's the dream.
- The problem is made much worse because our data types aren't granular enough. e.g. we treat DateTime as a string. But when we do the query in the db, we need to know that that string is actually a datetime. So the internals of the source can't piggy back off the airbyte data model. This breaks the previously JdbcSourceAbstraction because it trusted that our types were good enough.

Checklist

fix generating json2pojo models
add unit tests for state manager
run it in integration tests
add it officially to the pg source and make sure standard tests pass
manual test of PG since we don't have super comprehensive standard tests for incremental yet. (manual testing unfortunately caught a pretty terrible bug. i think this is signal that we need to up our game on test our standard source test)
document how one is supposed to add a catalog for standard source tests.
bump version

Reading Order

AbstractJdbcSource
JdbcStateManager
StateDecoratingIterator
the rest

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcState1.java

cgardens · 2020-12-03T02:59:42Z

airbyte-commons/src/main/java/io/airbyte/commons/stream/MoreStreams.java

+
+public class MoreStreams {
+
+  public static <T> Stream<T> toStream(Iterator<T> iterator) {


will find replace other instances of this in a separate pr.

...nnectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/IncrementalUtils.java

cgardens · 2020-12-03T03:00:52Z

...nnectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/JdbcStateManager.java

+    return Optional.ofNullable(map.get(streamName));
+  }
+
+  public Optional<String> getOriginalCursorField(String streamName) {


this is a little goofy as opposed to just returning the cursor info. i didn't want to have to worry about defensive copy issues though.

cgardens · 2020-12-03T03:07:34Z

airbyte-integrations/connectors/source-jdbc-models/build.gradle

@@ -0,0 +1,24 @@
+import org.jsonschema2pojo.SourceType


I was running into a weird issue while trying to nest this model inside of airbyte-integrations/connectors/source-jdbc so I pulled it out to be a top level connector while I was developing. I will fix the issue and nest everything that is currently in airbyte-integrations/connectors/source-jdbc-models in airbyte-integrations/connectors/source-jdbc/models before merging.

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

cgardens · 2020-12-05T04:24:32Z

...andard-source-test/src/main/java/io/airbyte/integrations/standardtest/source/TestSource.java

+        configuredStream.setSyncMode(io.airbyte.protocol.models.SyncMode.INCREMENTAL);
+        if (Optional.ofNullable(configuredStream.getStream().getSourceDefinedCursor()).orElse(false)) {
+          configuredStream.setCursorField(configuredStream.getStream().getDefaultCursorField());
+        } else {


as the todo describes, this is really bad. i don't think we need to block postgres on fixing this, but as part of our goal to "launch" incremental, we need to figure out how to include it into our standard tests more ergonomically and transparently. will add an issue for next week. #1217

cgardens · 2020-12-05T04:25:15Z

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractJooqSource implements Source {


see main description. intentionally alot of copy and paste going on right now. goal next week is to reconsolidate.

reconsolidate with JDBC source correct? sounds good if so.

cgardens · 2020-12-05T04:26:11Z

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

+    } else if (jooqDataType.isString() || jooqDataType.isTemporal()) {
+      // todo (cgardens) - this is bad because only works for mysql, psql, and sqlite. these are also the
+      // only dialects that jooq supports in the free version, so this should not actually be an issue. it
+      // is still sketchy.


again something i want to try to resolve next week.

cgardens · 2020-12-05T04:32:34Z

...s/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/StateDecoratingIterator.java

+    if (messageStream.hasNext()) {
+      final AirbyteMessage message = messageStream.next();
+      final String cursorCandidate = message.getRecord().getData().get(cursorField).asText();
+      if (IncrementalUtils.compareCursors(maxCursor, cursorCandidate, cursorType) < 0) {


There's no guarantee here that our comparators will do the exact same thing as the databases. This is pretty scary, but haven't come up with a bullet proof way of fixing it yet.

I was thinking about what query I could submit to the database to do this for me. Doing it as individual queries would be too expensive, but if I can batch them somehow it might be okay. Haven't decided if this is a right now problem or not. Open to input.

sherifnada

Main comments are around:

might be missing something, but I'm unsure if the current dynamic of current vs. initial cursor is correct
why do we have two methods of discovery? can we have just one to avoid risk of them going out of sync?
is it possible/needed to paginate full refresh queries to the db? ideally incremental too.
is it possible to do cursor comparison DB side?

Rest of the comments are style or nits

...andard-source-test/src/main/java/io/airbyte/integrations/standardtest/source/TestSource.java

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

...nnectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/JdbcStateManager.java

sherifnada · 2020-12-07T19:21:56Z

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

+  }
+
+  private static Stream<Record> executeFullRefreshQuery(Database database, List<org.jooq.Field<?>> jooqFields, Table<?> table) throws SQLException {
+    return database.query(ctx -> ctx.select(jooqFields)


what do you think about doing paging in this query to avoid huge queries on the db? ideally we'd do the same with incremental

i don't think that this should be an issue. postgres should do all of this in a streaming fashion so it shouldn't technically cause problems. like i think the preferred way to handle this is have the db do this for you. i think you only mess with pagination if you really have to for doing stuff in web apps and what not.

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

sherifnada · 2020-12-07T22:55:48Z

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

+  }
+
+  @Override
+  public AirbyteCatalog discover(JsonNode config) throws Exception {


why doesn't this exclude schemas the same way internal does? feels like right now there are two distinct discoverable schemas, one external and one internal. Ideally we have the same I think, but I may be missing a detail here

i think it does. if you follow discover => getTables => discoverInternal. so they use the same code path. discover just has to wrap it in our data model, while internally we don't wrap it so we can use the more granularly defined types.

ok so I'm missing a detail then ;)

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

sherifnada · 2020-12-07T23:03:40Z

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJooqSource.java

+        final Optional<String> initialCursorOptional = stateManager.getOriginalCursor(streamName);
+
+        final Stream<AirbyteMessage> internalMessageStream;
+        if (initialCursorOptional.isPresent()) {


the logic right now seems to be: if there was ever a cursor, do incremental. else do fullrefresh.

Shouldn't it be the other way around i.e: if there is a current cursor AND that cursor is the same as the initial cursor do incremental otherwise FR? might be missing something.

i think you're right. i'll double check this.

sherifnada · 2020-12-07T23:06:12Z

...nnectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/IncrementalUtils.java

+  // x < 0 mean replace original
+  // x == 0 means keep original
+  // x > 0 means keep original
+  public static int compareCursors(String original, String candidate, JsonSchemaPrimitive type) {


can this comparison be done DB side?

if we can we should. see: https://github.com/airbytehq/airbyte/pull/1172/files#r536510322. i have some performance concerns around it.

😏 select MAX(col1) from ( values ('a'), ('b'), ('c')) as x (col1);

😏😏😏😏😏🤡 🥨 😏 ctx.select(DSL.max(DSL.field("col1"))).from(DSL.values(DSL.row("1"), DSL.row("3"), DSL.row("2")).as("t", "col1"))

will decide tomorrow if i'm going to do this in this iteration. but it's nifty.

going to hold off on this. i think it is low risk for right now and i don't want to put the work in here if it looks like i'm going to replace it with a jdbcsource anyway.

michel-tricot

It is not ready to merge right?

michel-tricot · 2020-12-08T03:56:18Z

...nnectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/JdbcStateManager.java

+  }
+
+  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+  private static CursorInfo createCursorInfoForStream(String streamName,


this function is swallowing errors with the logging. I don't think we should log if we are unable to find the cursor fields and just do a full refresh. I think we should fail instead.

That's not what is happening here. None of these are failure states. Given that it is a valid configuration, these are all possible states that need to be handled. Let me know if you you see a case in here that should be a failure case. I've reviewed each of them and they all seem reasonable to me.

michel-tricot · 2020-12-08T04:01:38Z

...s/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/StateDecoratingIterator.java

+      return new AirbyteMessage().withType(Type.STATE).withState(stateMessage);
+    } else {
+      return endOfData();
+


NIT: remove new line

cgardens · 2020-12-08T04:38:03Z

correct still iterating on this based on sherif's feedback.

cgardens · 2020-12-08T18:45:20Z

This is ready for another pass. I am working on the last todos mentioned above, but I don't think any of them will make or break your review.

…irbyte/integrations/source/jdbc/AbstractJooqSource.java Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

…ng of timezones across platforms

cgardens force-pushed the cgardens/add_state_to_jdbc branch from cb0a1f0 to a57c953 Compare December 2, 2020 22:31

michel-tricot reviewed Dec 3, 2020

View reviewed changes

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcState1.java Outdated Show resolved Hide resolved

cgardens commented Dec 3, 2020

View reviewed changes

...nnectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/IncrementalUtils.java Outdated Show resolved Hide resolved

cgardens commented Dec 3, 2020

View reviewed changes

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java Outdated Show resolved Hide resolved

cgardens force-pushed the cgardens/add_state_to_jdbc branch from 775e2ce to f70c5af Compare December 5, 2020 04:12

cgardens commented Dec 5, 2020

View reviewed changes

cgardens requested a review from michel-tricot December 5, 2020 04:34

cgardens changed the title ~~add state to jdbc source~~ add incremental to ~~jdbc~~ jooq source (and postgres) Dec 5, 2020

cgardens marked this pull request as ready for review December 5, 2020 04:35

cgardens requested review from sherifnada, ChristopheDuong and jrhizor December 5, 2020 04:35

sherifnada reviewed Dec 7, 2020

View reviewed changes

michel-tricot reviewed Dec 8, 2020

View reviewed changes

cgardens requested review from michel-tricot and sherifnada December 8, 2020 18:45

cgardens and others added 4 commits December 8, 2020 13:16

initial impl

40faa3f

Update airbyte-integrations/connectors/source-jdbc/src/main/java/io/a…

a8a6812

…irbyte/integrations/source/jdbc/AbstractJooqSource.java Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

Update airbyte-integrations/connectors/source-jdbc/src/main/java/io/a…

47a88bf

…irbyte/integrations/source/jdbc/AbstractJooqSource.java Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

code review feedback

f6a49d7

cgardens force-pushed the cgardens/add_state_to_jdbc branch from 68e5a8f to f6a49d7 Compare December 8, 2020 21:16

format

8d94461

sherifnada approved these changes Dec 8, 2020

View reviewed changes

cgardens added 8 commits December 8, 2020 16:16

bug avoidance

d255418

fix user defined test

ac4307d

feed the tests

1321a7b

format

22f125f

use dates instead of timestamps to get around inconsitent jooq handli…

eda376c

…ng of timezones across platforms

update checklist with mention to configure catalog as incremental

f7adcea

bump version

c6c9b33

add config

a201ce6

cgardens merged commit 25689ee into master Dec 9, 2020

cgardens deleted the cgardens/add_state_to_jdbc branch December 9, 2020 05:14

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add incremental to jdbc jooq source (and postgres) #1172

add incremental to jdbc jooq source (and postgres) #1172

cgardens commented Dec 2, 2020 •

edited

Loading

cgardens Dec 3, 2020

cgardens Dec 3, 2020

cgardens Dec 3, 2020 •

edited

Loading

cgardens Dec 5, 2020

cgardens Dec 5, 2020

sherifnada Dec 7, 2020

cgardens Dec 7, 2020

cgardens Dec 5, 2020

cgardens Dec 5, 2020

sherifnada left a comment

sherifnada Dec 7, 2020

cgardens Dec 7, 2020 •

edited

Loading

sherifnada Dec 7, 2020

cgardens Dec 7, 2020

sherifnada Dec 7, 2020

sherifnada Dec 7, 2020

cgardens Dec 7, 2020

sherifnada Dec 7, 2020

cgardens Dec 7, 2020

cgardens Dec 8, 2020

cgardens Dec 8, 2020

cgardens Dec 8, 2020

cgardens Dec 8, 2020

michel-tricot left a comment

michel-tricot Dec 8, 2020

cgardens Dec 8, 2020

michel-tricot Dec 8, 2020

cgardens commented Dec 8, 2020

cgardens commented Dec 8, 2020


		public class MoreStreams {

		public static <T> Stream<T> toStream(Iterator<T> iterator) {

add incremental to ~~jdbc~~ jooq source (and postgres) #1172

add incremental to ~~jdbc~~ jooq source (and postgres) #1172

Conversation

cgardens commented Dec 2, 2020 • edited Loading

What

How

Decisions to make

Checklist

Reading Order

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgardens Dec 3, 2020 • edited Loading

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

sherifnada left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgardens Dec 7, 2020 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

michel-tricot 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

cgardens commented Dec 8, 2020

cgardens commented Dec 8, 2020

add incremental to jdbc jooq source (and postgres) #1172

add incremental to jdbc jooq source (and postgres) #1172

cgardens commented Dec 2, 2020 •

edited

Loading

cgardens Dec 3, 2020 •

edited

Loading

cgardens Dec 7, 2020 •

edited

Loading