[source-mysql] Fix NPE on cursor based full refresh

airbyte-cdk/java/airbyte-cdk/db-sources/src/testFixtures/kotlin/io/airbyte/cdk/integrations/source/jdbc/test/JdbcSourceAcceptanceTest.kt

@@ -479,6 +479,46 @@ abstract class JdbcSourceAcceptanceTest<S : Source, T : TestDatabase<*, T, *>> {
         Assertions.assertTrue(actualRecordMessages.containsAll(expectedMessages))
     }
 
+    @Test
+    @Throws(Exception::class)
+    protected fun testReadBothIncrementalAndFullRefreshStreams() {
+        val catalog = getConfiguredCatalogWithOneStream(defaultNamespace)
+        val expectedMessages: MutableList<AirbyteMessage> = ArrayList(testMessages)
+
+        val streamName2 = streamName() + 2
+        val tableName = getFullyQualifiedTableName(TABLE_NAME + 2)
+        testdb!!
+            .with(createTableQuery(tableName, "id INTEGER, name VARCHAR(200)", ""))
+            .with("INSERT INTO %s(id, name) VALUES (1,'picard')", tableName)
+            .with("INSERT INTO %s(id, name) VALUES (2, 'crusher')", tableName)
+            .with("INSERT INTO %s(id, name) VALUES (3, 'vash')", tableName)
+
+        val airbyteStream2 =
+            CatalogHelpers.createConfiguredAirbyteStream(
+                streamName2,
+                defaultNamespace,
+                Field.of(COL_ID, JsonSchemaType.NUMBER),
+                Field.of(COL_NAME, JsonSchemaType.STRING)
+            )
+        airbyteStream2.syncMode = SyncMode.INCREMENTAL
+        airbyteStream2.cursorField = java.util.List.of(COL_ID)
+        airbyteStream2.destinationSyncMode = DestinationSyncMode.APPEND
+        catalog.streams.add(airbyteStream2)
+
+        expectedMessages.addAll(getAirbyteMessagesSecondSync(streamName2))
+
+        System.out.println("catalog: " + catalog)
+
+        val actualMessages = MoreIterators.toList(source()!!.read(config(), catalog, null))
+        val actualRecordMessages = filterRecords(actualMessages)
+
+        setEmittedAtToNull(actualMessages)
+
+        Assertions.assertEquals(expectedMessages.size, actualRecordMessages.size)
+        Assertions.assertTrue(expectedMessages.containsAll(actualRecordMessages))
+        Assertions.assertTrue(actualRecordMessages.containsAll(expectedMessages))
+    }
+
     protected open fun getAirbyteMessagesSecondSync(streamName: String?): List<AirbyteMessage> {
         return testMessages
             .stream()

airbyte-integrations/connectors/source-mysql/build.gradle

@@ -8,7 +8,7 @@ plugins {
 airbyteJavaConnector {
     cdkVersionRequired = '0.31.8'
     features = ['db-sources']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 java {

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlQueryUtils.java

@@ -175,7 +175,9 @@ public static Map<io.airbyte.protocol.models.v0.AirbyteStreamNameNamespacePair,
         }
 
         LOGGER.info("Querying max cursor value for {}.{}", namespace, name);
+
         final String cursorField = cursorInfoOptional.get().getCursorField();
+        LOGGER.info("cursor field", cursorField);
         final String quotedCursorField = getIdentifierWithQuoting(cursorField, quoteString);
         final String cursorBasedSyncStatusQuery = String.format(MAX_CURSOR_VALUE_QUERY,
             quotedCursorField,

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java

@@ -14,6 +14,7 @@
 import static io.airbyte.integrations.source.mysql.MySqlQueryUtils.getTableSizeInfoForStreams;
 import static io.airbyte.integrations.source.mysql.MySqlQueryUtils.logStreamSyncStatus;
 import static io.airbyte.integrations.source.mysql.initialsync.MySqlInitialReadUtil.convertNameNamespacePairFromV0;
+import static io.airbyte.integrations.source.mysql.initialsync.MySqlInitialReadUtil.filterStreamInIncrementalMode;
 import static io.airbyte.integrations.source.mysql.initialsync.MySqlInitialReadUtil.getMySqlFullRefreshInitialLoadHandler;
 import static io.airbyte.integrations.source.mysql.initialsync.MySqlInitialReadUtil.getMySqlInitialLoadGlobalStateManager;
 import static io.airbyte.integrations.source.mysql.initialsync.MySqlInitialReadUtil.initPairToPrimaryKeyInfoMap;
@@ -467,7 +468,8 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(final
       if (isAnyStreamIncrementalSyncMode(catalog)) {
         LOGGER.info("Syncing via Primary Key");
         final MySqlCursorBasedStateManager cursorBasedStateManager = new MySqlCursorBasedStateManager(stateManager.getRawStateMessages(), catalog);
-        final InitialLoadStreams initialLoadStreams = streamsForInitialPrimaryKeyLoad(cursorBasedStateManager, catalog);
+        final InitialLoadStreams initialLoadStreams =
+            filterStreamInIncrementalMode(streamsForInitialPrimaryKeyLoad(cursorBasedStateManager, catalog));
         final Map<AirbyteStreamNameNamespacePair, CursorBasedStatus> pairToCursorBasedStatus =
             getCursorBasedSyncStatusForStreams(database, initialLoadStreams.streamsForInitialLoad(), stateManager, getQuoteString());
         final CursorBasedStreams cursorBasedStreams =

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/initialsync/MySqlInitialReadUtil.java

@@ -50,6 +50,7 @@
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteStream;
 import io.airbyte.protocol.models.v0.StreamDescriptor;
+import io.airbyte.protocol.models.v0.SyncMode;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -368,6 +369,13 @@ private static boolean streamHasPrimaryKey(final ConfiguredAirbyteStream stream)
     return stream.getStream().getSourceDefinedPrimaryKey().size() > 0;
   }
 
+  public static InitialLoadStreams filterStreamInIncrementalMode(final InitialLoadStreams stream) {
+    return new InitialLoadStreams(
+        stream.streamsForInitialLoad.stream().filter(airbyteStream -> airbyteStream.getSyncMode() == SyncMode.INCREMENTAL)
+            .collect(Collectors.toList()),
+        stream.pairToInitialLoadStatus);
+  }
+
   public static List<ConfiguredAirbyteStream> identifyStreamsToSnapshot(final ConfiguredAirbyteCatalog catalog,
                                                                         final Set<AirbyteStreamNameNamespacePair> alreadySyncedStreams) {
     final Set<AirbyteStreamNameNamespacePair> allStreams = AirbyteStreamNameNamespacePair.fromConfiguredCatalog(catalog);