jprdonnelly/airbyte
1
0
Fork 0
mirror of synced 2025-12-19 18:14:56 -05:00
Code Issues Packages Projects Releases Wiki Activity

trigger-cdc: fix empty changes case

Browse Source
This commit is contained in:
matt.bayley
2025-12-18 22:09:23 -08:00
parent 14d4ce4f4f
commit f3b8079ae2
2 changed files with 9 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
airbyte-cdk/bulk/toolkits/extract-trigger/src/main/kotlin/io/airbyte/cdk/TriggerPartition.kt
View File

@@ -100,7 +100,7 @@ class TriggerUnsplittableSnapshotWithCursorPartition(
get() = get() =
TriggerStreamStateValue.cursorIncrementalCheckpoint( TriggerStreamStateValue.cursorIncrementalCheckpoint(
cursor, cursor,
cursorCheckpoint = streamState.cursorUpperBound!!, cursorCheckpoint = streamState.cursorUpperBound ?: Jsons.nullNode(),
) )
override val cursorUpperBoundQuery: SelectQuery override val cursorUpperBoundQuery: SelectQuery
@@ -276,7 +276,7 @@ sealed class TriggerCursorPartition(
JdbcCursorPartition<TriggerStreamState> { JdbcCursorPartition<TriggerStreamState> {
val cursorUpperBound: JsonNode val cursorUpperBound: JsonNode
get() = explicitCursorUpperBound ?: streamState.cursorUpperBound!! get() = explicitCursorUpperBound ?: streamState.cursorUpperBound ?: Jsons.nullNode()
val cursorUpperBoundFrom: From = val cursorUpperBoundFrom: From =
if (triggerCdcPartitionState == null) from if (triggerCdcPartitionState == null) from
else else

7
airbyte-cdk/bulk/toolkits/extract-trigger/src/main/kotlin/io/airbyte/cdk/TriggerPartitionFactory.kt
View File

@@ -63,6 +63,13 @@ class TriggerPartitionFactory(
val stream: Stream = streamFeedBootstrap.feed val stream: Stream = streamFeedBootstrap.feed
val streamState: TriggerStreamState = streamState(streamFeedBootstrap) val streamState: TriggerStreamState = streamState(streamFeedBootstrap)
val opaqueStateValue: OpaqueStateValue? = streamFeedBootstrap.currentState val opaqueStateValue: OpaqueStateValue? = streamFeedBootstrap.currentState
// An empty table stream state will be marked as a nullNode. This prevents repeated attempt
// to read it
if (opaqueStateValue?.isNull == true) {
return null
}
if (opaqueStateValue == null) { if (opaqueStateValue == null) {
return coldStart(streamState) return coldStart(streamState)
} }