SQL Sources should include previous sync's cursor value in SELECT statement as not to skip rows added mid-sync

[evantahler]

Description

Consider the following users table.

id	email	name	created_at	updated_at
1	evan@example.com	Evan	2022-01-01 00:00:00	2022-01-01 00:00:00
2	brian@example.com	Brian	2022-01-02 00:00:00	2022-01-02 00:00:00
3	andy@example.com	Andy	2022-01-03 00:00:00	2022-01-03 00:00:00

A common sync would be to create in incremental sync, using the updated_at timestamp column + de-duped history. As users are added or updated the updated_at column is changed and the next sync will pick them up as the cursor can only move forward in time.

There's a problem however at the boundary of the sync. If the sync itself was running at 2022-01-03 00:00:00, right when a new user was added, they would be not be included in the sync:

id	email	name	created_at	updated_at
1	evan@example.com	Evan	2022-01-01 00:00:00	2022-01-01 00:00:00
2	brian@example.com	Brian	2022-01-02 00:00:00	2022-01-02 00:00:00
3	andy@example.com	Andy	2022-01-03 00:00:00	2022-01-03 00:00:00
4	greg@example.com	Greg	2022-01-03 00:00:00	2022-01-03 00:00:00

We won't pick up user 4 (at least until something about them changes and their updated_at is bumped) because in subsequent syncs, we use the greater-than comparison to choose which records to import. This can lead to missing records if they were inserted as the sync was running.

Local confirmation

To confirm this behavior, create a source table with the 3 rows listed in the first example, and run the sync. You will have 3 records in your destination. Then add a 4th row with the an updated_at matching the current cursor ("2022-01-03 00:00:00" in this example). Sync again, and see that user 4 did not make it to the destination.

Possible Solutions

1. Rather than selecting records from SQL sources matching "SELECT %s FROM %s WHERE %s > ?", we could use >=, e.g. "SELECT %s FROM %s WHERE %s >= ?". This solution is the most robust because it also works for numeric (vs time) cursors as well. This will have the negative side effect of almost always creating a duplicate entry (the previous sync's most recent record is very likely to be returned again at the start of a subsequent sync), and normalization will need to deal with it. This is not a problem with de-duped history, but is weird in all other sync modes.

2. Building from solution 1 above (using >=), we could store both the cursor value and the offset in the STATE. This solves the duplicate row problem introduced above. The state stored could remember the previous sync's cursor and OFFSET. In this way, the next sync could continue from the same cursor position + offset to be sure that now new records appeared within the same cursor that we haven't seen yet:

SELECT * from users LIMIT 100 # got some records
SELECT * from users LIMIT 100 OFFSET 3 # got 0 records after using offset
# first sync ends, cursor is {value: "2022-01-03 00:00:00" + offset: 0}

# Next sync starts
SELECT * from users WHERE updated_at >= "2022-01-03 00:00:00" LIMIT 100 OFFSET 1 # use previous offset + 1 # 1 new row found!

So, the state for an incremental SQL cursor source should look like:

{ users: { offset: 0, cursor: "2022-01-03 00:00:00"}}

Probably bad solutions

1. If we could detect that the cursor was a time field, we could append AND ${cursor} >= NOW() - INTERVAL '2 seconds'. In this way, we would ignore "too recent" values that might be inserted at the time of the sync. This doesn't solve the problem with numeric cursors, only "time"-like cursors.
2. Introduce a sleep and one more read before the sync is determined to be complete, e.g.:

SELECT * from users LIMIT 100 # got some records
SELECT * from users LIMIT 100 OFFSET 3 # got 0 records after using offset
# SLEEP 2. Sleep enough time to ensure that the cursor's minimum resolution (often 1 second) has passed
SELECT * from users LIMIT 100  OFFSET 3 # got 1 record that was inserted during the sleep
# LOOP or END if still 0 rows

3. Don't ever send any RECORDS from the latest cursor value. You'll never see any duplicates, but the destination will always be some level of "behind" the source

Misc

This approach assumes that our source database have an implicit, static sort which will return rows in the same order without needing an ORDER BY. This is true for postgres and mysql... is it true for all DB sources? We may also need to add an ORDER BY clause.

[grishick]

Hey team! Please add your planning poker estimate with ZenHub @edgao @ryankfu @subodh1810 @tuliren

[tuliren]

The OFFSET approach, the second possible solution, probably does not work. The return order of the records is unknown other than the updated_at. So it is possible that the newly added record shows up as the first row:

id	email	name	created_at	updated_at
1	greg@example.com	Greg	2022-01-03 00:00:00	2022-01-03 00:00:00
2	andy@example.com	Andy	2022-01-03 00:00:00	2022-01-03 00:00:00

which will be skipped with a query with OFFSET 1.

[evantahler]

The return order of the records is unknown other than the updated_at

We will probably need to also sort by the cursor column for this solution. There's added db load.

[tuliren]

We will probably need to also sort by the cursor column for this solution. There's added db load.

Oh, right. I missed the sorting part in the MISC section.

---

The chosen solution, solution 1, can lead to lots of duplications. Imagine a connection that runs every 5 minutes, if the source database has no updates, the destination will get 12x duplicated records for the cursor value per hour.

There are a few ways to reduce the amount of duplications.

• Track the number of rows synced together with the cursor value. So the state becomes { users: { count: 3, cursor: "2022-01-03 00:00:00"}}. Each time before the incremental sync, count the number of rows with cursorColumn = cursorValue. If the result is different from the one in the state, we know there are changes, and use >= for the incremental query. Otherwise, use > to avoid the duplication.
• The above approach has one flaw. If someone removed one row and added another, the row count will remain the same, but the data is different. A more precise approach is to store not just the count, but a hash of all rows with cursorColumn = cursorValue. The state becomes { users: { hash: "c2RmO2xqYXM7bG", cursor: "2022-01-03 00:00:00"}} Only query with >= if the hash is different. The downside of this approach is that we need to so lots of hashing on the fly.

(Another approach is to exclude the max cursor value in the current sync, i.e. query with cursorColumn >= cursorValue and cursorColumn < currentMaxCursorValue. The data rows with the currentMaxCursorValue will be synced in full next time. However, after a second thought, this solution does not work because the destination can never get the data with currentMaxCursorValue, which will be pretty frustrating for the users.)

So maybe we should just implement solution 2.

[tuliren]

Actually solution 2 is very hard to implement with our current code structure. Right now, we chain together a series of iterators ahead of time. Each iterator is created based on one concrete SQL query. To implement solution 2, we need to keep firing queries inside the iterator until it reaches an offset with no data. If we go with this solution, it will be significantly more complicated than a few hours of work.

---

Update. Actually we can get the total number of records first, and pre-define the offsets. So this is not a problem.