Unexpected and inconsistent Incremental | Append behavior for MSSQL Source -> MSSQL Destination

[cohenj20]

TL:DR

I'm using airbyte to replicate data between two sql systems. I'm pulling four tables and using a cursor column in each table to achieve Incremental - Append syncs. When I run the first sync all records are extracted and loaded as expected. Without modifying any data in the source tables, I run another sync. All tables but one extract and load 0 records as expected. One table however extract and loads one row from the source, even though the value in the cursor column was not changed.

---

Overview

I'm building out an Airbyte demo for my company. I'm using an instance of the Microsoft AdventureWorks2019 SQL Server database as the source system and a SQL Server Database called Raw on the same server as the destination. I had no trouble configuring the source and destination connectors and creating the connection. However, I'm observing some inconsistently unexpected behavior with the Incremental | Append + Deduped and Incremental | Append replication strategies. Below I review the connector information and the details regarding the behavior I'm seeing.

Connector Information

Source: AdventureWorks2019 MS SQL Server (Certified Connector)
Destination: Raw MS SQL Server (Community Connector)

Source Table Schemas

Customer Table

Person Table

SalesOrderDetail Table

SalesOrderHeader Table

As seen in the screenshots, each table has a replication method of Incremental | Append + Deduped. The cursor columns are all the field ModifiedDate, which is present in every table in the AdventureWorks2019 DB.

First Sync

Expectations: No state exists for these streams, so a full-refresh is expected where all records are extracted and loaded.
Observations: Sync completes without error and all records are extracted and loaded to the destination.

Important Log snippets:

# No state
2024-04-19 18:49:04 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):241 No cursor field set in catalog but not present in state. Stream: Sales_Customer, New Cursor Field: ModifiedDate. Resetting cursor value
2024-04-19 18:49:04 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):241 No cursor field set in catalog but not present in state. Stream: Person_Person, New Cursor Field: ModifiedDate. Resetting cursor value
2024-04-19 18:49:04 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):241 No cursor field set in catalog but not present in state. Stream: Sales_SalesOrderHeader, New Cursor Field: ModifiedDate. Resetting cursor value
2024-04-19 18:49:04 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):241 No cursor field set in catalog but not present in state. Stream: Sales_SalesOrderDetail, New Cursor Field: ModifiedDate. Resetting cursor value

# Replication Summary
2024-04-19 18:50:01 replication-orchestrator > sync summary: {
  "status" : "completed",
  "recordsSynced" : 192574,
  "bytesSynced" : 72715801,
  "startTime" : 1713552536353,
  "endTime" : 1713552601845,
  "totalStats" : {
    "bytesCommitted" : 72715801,
    "bytesEmitted" : 72715801,
    "destinationStateMessagesEmitted" : 13,
    "destinationWriteEndTime" : 1713552601508,
    "destinationWriteStartTime" : 1713552536376,
    "meanSecondsBeforeSourceStateMessageEmitted" : 3,
    "maxSecondsBeforeSourceStateMessageEmitted" : 6,
    "maxSecondsBetweenStateMessageEmittedandCommitted" : 11,
    "meanSecondsBetweenStateMessageEmittedandCommitted" : 5,
    "recordsEmitted" : 192574,
    "recordsCommitted" : 192574,
    "replicationEndTime" : 1713552601842,
    "replicationStartTime" : 1713552536353,
    "sourceReadEndTime" : 1713552589761,
    "sourceReadStartTime" : 1713552536377,
    "sourceStateMessagesEmitted" : 21
  },
  "streamStats" : [ {
    "streamName" : "Customer",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 3756244,
      "bytesEmitted" : 3756244,
      "recordsEmitted" : 19820,
      "recordsCommitted" : 19820
    }
  }, {
    "streamName" : "Person",
    "streamNamespace" : "Person",
    "stats" : {
      "bytesCommitted" : 16049012,
      "bytesEmitted" : 16049012,
      "recordsEmitted" : 19972,
      "recordsCommitted" : 19972
    }
  }, {
    "streamName" : "SalesOrderHeader",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 18426358,
      "bytesEmitted" : 18426358,
      "recordsEmitted" : 31465,
      "recordsCommitted" : 31465
    }
  }, {
    "streamName" : "SalesOrderDetail",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 34484187,
      "bytesEmitted" : 34484187,
      "recordsEmitted" : 121317,
      "recordsCommitted" : 121317
    }
  } ],
  "performanceMetrics" : {
    "processFromSource" : {
      "elapsedTimeInNanos" : 3087790081,
      "executionCount" : 192607,
      "avgExecTimeInNanos" : 16031.556906031452
    },
    "readFromSource" : {
      "elapsedTimeInNanos" : 14984986547,
      "executionCount" : 228113,
      "avgExecTimeInNanos" : 65691.06779096325
    },
    "processFromDest" : {
      "elapsedTimeInNanos" : 152447965,
      "executionCount" : 13,
      "avgExecTimeInNanos" : 1.1726766538461538E7
    },
    "writeToDest" : {
      "elapsedTimeInNanos" : 36425292430,
      "executionCount" : 192595,
      "avgExecTimeInNanos" : 189128.96196682157
    },
    "readFromDest" : {
      "elapsedTimeInNanos" : 60409413427,
      "executionCount" : 284579,
      "avgExecTimeInNanos" : 212276.42737868923
    }
  }
}

Second Sync (No change to source data)

Expectations: Now state exists for these streams. Because source tables did not change, it is expected that no rows would be extracted and loaded.
Observations: Sync completes without error, however, one stream exhibits unexpected behavior: it extracts and loads the one record in the table whose cursor column value = saved state for the cursor column.

Important Log snippets:

# the other streams have records that match the saved state cursor value too
2024-04-19 18:53:28 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Sales_Customer. Cursor Field: ModifiedDate Value: 2024-09-12T11:15:07.263 Count: 1
2024-04-19 18:53:28 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Person_Person. Cursor Field: ModifiedDate Value: 2024-01-07T00:00:00.000000 Count: 1
2024-04-19 18:53:28 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Sales_SalesOrderHeader. Cursor Field: ModifiedDate Value: 2014-07-07T00:00:00.000000 Count: 40
2024-04-19 18:53:28 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Sales_SalesOrderDetail. Cursor Field: ModifiedDate Value: 2014-06-30T00:00:00.000000 Count: 96

# but only Customer emits the row
2024-04-19 18:53:29 source > state message creation: Sales_SalesOrderDetail 2014-06-30T00:00:00.000000 0
...
2024-04-19 18:53:29 source > state message creation: Sales_Customer 2024-09-12T11:15:07.263000 1
...
2024-04-19 18:53:30 source > state message creation: Sales_SalesOrderHeader 2014-07-07T00:00:00.000000 0
...
2024-04-19 18:53:28 source > state message creation: Person_Person 2024-01-07T00:00:00.000000 0

# Replication Summary
2024-04-19 18:53:42 replication-orchestrator > sync summary: {
  "status" : "completed",
  "recordsSynced" : 1,
  "bytesSynced" : 184,
  "startTime" : 1713552798181,
  "endTime" : 1713552822171,
  "totalStats" : {
    "bytesCommitted" : 184,
    "bytesEmitted" : 184,
    "destinationStateMessagesEmitted" : 4,
    "destinationWriteEndTime" : 1713552822083,
    "destinationWriteStartTime" : 1713552798197,
    "meanSecondsBeforeSourceStateMessageEmitted" : 0,
    "maxSecondsBeforeSourceStateMessageEmitted" : 1,
    "maxSecondsBetweenStateMessageEmittedandCommitted" : 12,
    "meanSecondsBetweenStateMessageEmittedandCommitted" : 11,
    "recordsEmitted" : 1,
    "recordsCommitted" : 1,
    "replicationEndTime" : 1713552822168,
    "replicationStartTime" : 1713552798181,
    "sourceReadEndTime" : 1713552810727,
    "sourceReadStartTime" : 1713552798198,
    "sourceStateMessagesEmitted" : 4
  },
  "streamStats" : [ {
    "streamName" : "Customer",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 184,
      "bytesEmitted" : 184,
      "recordsEmitted" : 1,
      "recordsCommitted" : 1
    }
  }, {
    "streamName" : "Person",
    "streamNamespace" : "Person",
    "stats" : {
      "bytesCommitted" : 0,
      "bytesEmitted" : 0,
      "recordsEmitted" : 0,
      "recordsCommitted" : 0
    }
  }, {
    "streamName" : "SalesOrderHeader",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 0,
      "bytesEmitted" : 0,
      "recordsEmitted" : 0,
      "recordsCommitted" : 0
    }
  }, {
    "streamName" : "SalesOrderDetail",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 0,
      "bytesEmitted" : 0,
      "recordsEmitted" : 0,
      "recordsCommitted" : 0
    }
  } ],
  "performanceMetrics" : {
    "processFromSource" : {
      "elapsedTimeInNanos" : 466483451,
      "executionCount" : 17,
      "avgExecTimeInNanos" : 2.7440203E7
    },
    "readFromSource" : {
      "elapsedTimeInNanos" : 7013593583,
      "executionCount" : 28659,
      "avgExecTimeInNanos" : 244725.69116158973
    },
    "processFromDest" : {
      "elapsedTimeInNanos" : 88075883,
      "executionCount" : 4,
      "avgExecTimeInNanos" : 2.201897075E7
    },
    "writeToDest" : {
      "elapsedTimeInNanos" : 10448449,
      "executionCount" : 5,
      "avgExecTimeInNanos" : 2089689.8
    },
    "readFromDest" : {
      "elapsedTimeInNanos" : 18323135068,
      "executionCount" : 226221,
      "avgExecTimeInNanos" : 80996.61423121637
    }
  }
}

Third Sync (One row changed in source data)

I decided to run a test where I modified the cursor value in one of the rows in one of the source tables.

Expectations: Because one row in the source tables did change, it is expected that that one rows would be extracted and loaded.
Observations: Sync completes without error, and the one modified row is successfully extracted and loaded into the destination. The problematic stream, however, still extracts and loads the one record in the table whose cursor column value = saved state for the cursor column.

Important Log snippets:

# existing state
2024-04-19 19:00:08 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Sales_Customer. Cursor Field: ModifiedDate Value: 2024-09-12T11:15:07.263000 Count: 1
2024-04-19 19:00:08 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Person_Person. Cursor Field: ModifiedDate Value: 2024-01-07T00:00:00.000000 Count: 1
2024-04-19 19:00:08 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Sales_SalesOrderHeader. Cursor Field: ModifiedDate Value: 2014-07-07T00:00:00.000000 Count: 40
2024-04-19 19:00:08 source > INFO main i.a.c.i.s.r.s.CursorManager(createCursorInfoForStream$io_airbyte_airbyte_cdk_java_airbyte_cdk_airbyte_cdk_db_sources):219 Found matching cursor in state. Stream: Sales_SalesOrderDetail. Cursor Field: ModifiedDate Value: 2014-06-30T00:00:00.000000 Count: 96

# Customer still emits the row even though cursor value did not change in source table row
2024-04-19 19:00:08 source > state message creation: Person_Person 2024-01-07T00:00:00.000000 0
...
2024-04-19 19:00:09 source > state message creation: Sales_SalesOrderDetail 2014-06-30T00:00:00.000000 0
...
2024-04-19 19:00:09 source > state message creation: Sales_Customer 2024-09-12T11:15:07.263000 1 
...
2024-04-19 19:00:10 source > state message creation: Sales_SalesOrderHeader 2021-06-08T00:00:00.000000 1 # table containing modified row

# Replication Summary
2024-04-19 19:00:21 replication-orchestrator > sync summary: {
  "status" : "completed",
  "recordsSynced" : 2,
  "bytesSynced" : 776,
  "startTime" : 1713553198911,
  "endTime" : 1713553221698,
  "totalStats" : {
    "bytesCommitted" : 776,
    "bytesEmitted" : 776,
    "destinationStateMessagesEmitted" : 4,
    "destinationWriteEndTime" : 1713553221591,
    "destinationWriteStartTime" : 1713553198933,
    "meanSecondsBeforeSourceStateMessageEmitted" : 0,
    "maxSecondsBeforeSourceStateMessageEmitted" : 1,
    "maxSecondsBetweenStateMessageEmittedandCommitted" : 12,
    "meanSecondsBetweenStateMessageEmittedandCommitted" : 11,
    "recordsEmitted" : 2,
    "recordsCommitted" : 2,
    "replicationEndTime" : 1713553221695,
    "replicationStartTime" : 1713553198911,
    "sourceReadEndTime" : 1713553211347,
    "sourceReadStartTime" : 1713553198934,
    "sourceStateMessagesEmitted" : 4
  },
  "streamStats" : [ {
    "streamName" : "Customer",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 184,
      "bytesEmitted" : 184,
      "recordsEmitted" : 1,
      "recordsCommitted" : 1
    }
  }, {
    "streamName" : "Person",
    "streamNamespace" : "Person",
    "stats" : {
      "bytesCommitted" : 0,
      "bytesEmitted" : 0,
      "recordsEmitted" : 0,
      "recordsCommitted" : 0
    }
  }, {
    "streamName" : "SalesOrderHeader",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 592,
      "bytesEmitted" : 592,
      "recordsEmitted" : 1,
      "recordsCommitted" : 1
    }
  }, {
    "streamName" : "SalesOrderDetail",
    "streamNamespace" : "Sales",
    "stats" : {
      "bytesCommitted" : 0,
      "bytesEmitted" : 0,
      "recordsEmitted" : 0,
      "recordsCommitted" : 0
    }
  } ],
  "performanceMetrics" : {
    "processFromSource" : {
      "elapsedTimeInNanos" : 492459512,
      "executionCount" : 18,
      "avgExecTimeInNanos" : 2.7358861777777776E7
    },
    "readFromSource" : {
      "elapsedTimeInNanos" : 6984113777,
      "executionCount" : 149380,
      "avgExecTimeInNanos" : 46754.00841478109
    },
    "processFromDest" : {
      "elapsedTimeInNanos" : 107814402,
      "executionCount" : 4,
      "avgExecTimeInNanos" : 2.69536005E7
    },
    "writeToDest" : {
      "elapsedTimeInNanos" : 8435155,
      "executionCount" : 6,
      "avgExecTimeInNanos" : 1405859.1666666667
    },
    "readFromDest" : {
      "elapsedTimeInNanos" : 17200002718,
      "executionCount" : 94671,
      "avgExecTimeInNanos" : 181681.85313348332
    }
  }
}

Please let me know if anyone has questions or requires further clarification. Hope you guys can help!

[gera0220]

Maybe is because of this:

https://docs.airbyte.com/using-airbyte/core-concepts/sync-modes/incremental-append-deduped#inclusive-cursors