Skip to content

db-sources: disable counts for state messages for FULL_REFRESH streams #38208

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 10 commits into from
May 15, 2024
Merged

db-sources: disable counts for state messages for FULL_REFRESH streams #38208

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
b07dc5d
db-sources: disable counts for state messages for FULL_REFRESH streams
subodh1810 May 15, 2024
a797f47
format
subodh1810 May 15, 2024
ea5bcbe
fix test
subodh1810 May 15, 2024
863571b
bump up version
xiaohansong May 15, 2024
bc74d36
bump connector
xiaohansong May 15, 2024
4db8359
cdk test change
xiaohansong May 15, 2024
81c4d96
run connector test
xiaohansong May 15, 2024
cfcd9ad
mysql count validation
xiaohansong May 15, 2024
74ed7de
format
xiaohansong May 15, 2024
027e9a4
use published cdk
xiaohansong May 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 19 additions & 6 deletions .../main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/SourceStateIterator.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import com.google.common.collect.AbstractIterator
import io.airbyte.protocol.models.v0.AirbyteMessage
import io.airbyte.protocol.models.v0.AirbyteStateStats
import io.airbyte.protocol.models.v0.ConfiguredAirbyteStream
import io.airbyte.protocol.models.v0.SyncMode
import java.time.Duration
import java.time.Instant
import java.time.OffsetDateTime
Expand Down Expand Up @@ -41,9 +42,11 @@ open class SourceStateIterator<T>(
) {
val stateMessage =
sourceStateMessageProducer.generateStateMessageAtCheckpoint(stream)
stateMessage!!.withSourceStats(
AirbyteStateStats().withRecordCount(recordCount.toDouble())
)
if (shouldAttachCountWithState()) {
stateMessage!!.withSourceStats(
AirbyteStateStats().withRecordCount(recordCount.toDouble())
)
}

recordCount = 0L
lastCheckpoint = Instant.now()
Expand All @@ -64,9 +67,11 @@ open class SourceStateIterator<T>(
hasEmittedFinalState = true
val finalStateMessageForStream =
sourceStateMessageProducer.createFinalStateMessage(stream)
finalStateMessageForStream!!.withSourceStats(
AirbyteStateStats().withRecordCount(recordCount.toDouble())
)
if (shouldAttachCountWithState()) {
finalStateMessageForStream!!.withSourceStats(
AirbyteStateStats().withRecordCount(recordCount.toDouble())
)
}
recordCount = 0L
return AirbyteMessage()
.withType(AirbyteMessage.Type.STATE)
Expand All @@ -76,6 +81,14 @@ open class SourceStateIterator<T>(
}
}

/**
* We are disabling counts for FULL_REFRESH streams cause there is are issues with it. We should
* re-enable it once we do the work for project Counts: Emit Counts in Full Refresh
*/
private fun shouldAttachCountWithState(): Boolean {
return stream?.syncMode != SyncMode.FULL_REFRESH
}

// This method is used to check if we should emit a state message. If the record count is set to
// 0,
// we should not emit a state message.
Expand Down
1 change: 1 addition & 0 deletions ...n/io/airbyte/cdk/integrations/source/relationaldb/state/CursorStateMessageProducerTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -446,6 +446,7 @@ internal class CursorStateMessageProducerTest {
NAMESPACE,
Field.of(UUID_FIELD_NAME, JsonSchemaType.STRING)
)
.withSyncMode(SyncMode.INCREMENTAL)
.withCursorField(List.of(UUID_FIELD_NAME))

private val EMPTY_STATE_MESSAGE = createEmptyStateMessage(0.0)
Expand Down
Loading