Set lower bound included flag in snapshot partitions (#69177)

Author: rodireich

airbyte-integrations/connectors/source-mysql/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 3.51.2
+  dockerImageTag: 3.51.3
   dockerRepository: airbyte/source-mysql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
   githubIssueLabel: source-mysql

airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceJdbcPartition.kt

@@ -255,6 +255,7 @@ class MySqlSourceJdbcSplittableCdcRfrSnapshotPartition(
     primaryKey: List<Field>,
     override val lowerBound: List<JsonNode>?,
     override val upperBound: List<JsonNode>?,
+    override val isLowerBoundIncluded: Boolean,
 ) : MySqlSourceJdbcResumablePartition(selectQueryGenerator, streamState, primaryKey) {
     override val completeState: OpaqueStateValue
         get() =
@@ -364,6 +365,7 @@ class MySqlSourceJdbcSplittableSnapshotWithCursorPartition(
     override val upperBound: List<JsonNode>?,
     cursor: Field,
     cursorUpperBound: JsonNode?,
+    override val isLowerBoundIncluded: Boolean
 ) :
     MySqlSourceJdbcCursorPartition(
         selectQueryGenerator,

airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceJdbcPartitionFactory.kt

@@ -466,18 +466,21 @@ class MySqlSourceJdbcPartitionFactory(
                 true -> effectiveLowerBound
                 false -> type.jsonDecoder.decode(lowerBound[0])
             }
-        return calculateBoundaries(opaqueStateValues, lowerBound, upperBound)?.map { (l, u) ->
-            MySqlSourceJdbcSplittableSnapshotWithCursorPartition(
-                selectQueryGenerator,
-                streamState,
-                checkpointColumns,
-                listOf(stateValueToJsonNode(checkpointColumns[0], l.toString())),
-                u?.let { listOf(stateValueToJsonNode(checkpointColumns[0], u.toString())) },
-                //                listOf(stateValueToJsonNode(checkpointColumns[0], u.toString())),
-                cursor,
-                cursorUpperBound
-            )
-        }
+        return calculateBoundaries(opaqueStateValues, lowerBound, upperBound)
+            ?.entries
+            ?.mapIndexed { index, (l, u) ->
+                MySqlSourceJdbcSplittableSnapshotWithCursorPartition(
+                    selectQueryGenerator,
+                    streamState,
+                    checkpointColumns,
+                    listOf(stateValueToJsonNode(checkpointColumns[0], l.toString())),
+                    u?.let { listOf(stateValueToJsonNode(checkpointColumns[0], u.toString())) },
+                    cursor,
+                    cursorUpperBound,
+                    // The first partition includes the lower bound
+                    index == 0
+                )
+            }
     }
 
     private fun MySqlSourceJdbcRfrSnapshotPartition.split(
@@ -538,15 +541,19 @@ class MySqlSourceJdbcPartitionFactory(
                 false -> type.jsonDecoder.decode(lowerBound[0])
             }
 
-        return calculateBoundaries(opaqueStateValues, lowerBound, upperBound)?.map { (l, u) ->
-            MySqlSourceJdbcSplittableCdcRfrSnapshotPartition(
-                selectQueryGenerator,
-                streamState,
-                checkpointColumns,
-                listOf(stateValueToJsonNode(checkpointColumns[0], l.toString())),
-                u?.let { listOf(stateValueToJsonNode(checkpointColumns[0], u.toString())) },
-            )
-        }
+        return calculateBoundaries(opaqueStateValues, lowerBound, upperBound)
+            ?.entries
+            ?.mapIndexed { index, (l, u) ->
+                MySqlSourceJdbcSplittableCdcRfrSnapshotPartition(
+                    selectQueryGenerator,
+                    streamState,
+                    checkpointColumns,
+                    listOf(stateValueToJsonNode(checkpointColumns[0], l.toString())),
+                    u?.let { listOf(stateValueToJsonNode(checkpointColumns[0], u.toString())) },
+                    // The first partition includes the lower bound
+                    index == 0
+                )
+            }
     }
 
     private fun <T> calculateBoundaries(

docs/integrations/sources/mysql.md

@@ -230,7 +230,8 @@ Any database or table encoding combination of charset and collation is supported
 
 | Version     | Date       | Pull Request                                               | Subject                                                                                                                                         |
 |:------------|:-----------|:-----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------|
-| 3.51.2      | 2025-11-2  | [69104](https://github.com/airbytehq/airbyte/pull/69104)   | Better partitioning for tables with GUID string primary key.                                                                                    |
+| 3.51.3      | 2025-11-05 | [69177](https://github.com/airbytehq/airbyte/pull/69177)   | Fix a bug in CDC snapshot queries leading to omission of the first record in some cases.                                                                                |
+| 3.51.2      | 2025-11-02 | [69104](https://github.com/airbytehq/airbyte/pull/69104)   | Better partitioning for tables with GUID string primary key.                                                                                    |
 | 3.51.1      | 2025-10-24 | [68652](https://github.com/airbytehq/airbyte/pull/68652)   | Bump CDK version to the latest to resolve issue with database Views in CDC mode.                                                                |
 | 3.51.0      | 2025-10-15 | [68094](https://github.com/airbytehq/airbyte/pull/66515)   | Bump to the latest CDK for improved Protobuf encoding in socket mode.                                                                           |
 | 3.50.9      | 2025-10-06 | [67151](https://github.com/airbytehq/airbyte/pull/66515)   | Fix CDC decorating fields encoding to Protobuf                                                                                                  |