nit fixes

Author: gisripa

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcDestinationHandler.java

@@ -161,12 +161,16 @@ private CompletionStage<DestinationInitialState> retrieveState(final StreamConfi
     return CompletableFuture.supplyAsync(() -> {
       try {
         final Optional<TableDefinition> finalTableDefinition = findExistingTable(streamConfig.id());
-        // Only evaluate schema mismatch & final table emptiness if the final table exists.
-        boolean isSchemaMismatch = false;
-        boolean isFinalTableEmpty = true;
+        final boolean isSchemaMismatch;
+        final boolean isFinalTableEmpty;
         if (finalTableDefinition.isPresent()) {
           isSchemaMismatch = !existingSchemaMatchesStreamConfig(streamConfig, finalTableDefinition.get());
           isFinalTableEmpty = isFinalTableEmpty(streamConfig.id());
+        } else {
+          // If the final table doesn't exist, then by definition it doesn't have a schema mismatch and has no
+          // records.
+          isSchemaMismatch = false;
+          isFinalTableEmpty = true;
         }
         final InitialRawTableState initialRawTableState = getInitialRawTableState(streamConfig.id());
         return new DestinationInitialStateImpl(streamConfig, finalTableDefinition.isPresent(), initialRawTableState,

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/test/java/io/airbyte/integrations/base/destination/typing_deduping/MockSqlGenerator.java

@@ -32,12 +32,6 @@ public Sql createTable(final StreamConfig stream, final String suffix, final boo
     return Sql.of("CREATE TABLE " + stream.id().finalTableId("", suffix));
   }
 
-  // @Override
-  // public boolean existingSchemaMatchesStreamConfig(final StreamConfig stream, final String
-  // existingTable) throws TableNotMigratedException {
-  // return false;
-  // }
-
   @Override
   public Sql updateTable(final StreamConfig stream,
                          final String finalSuffix,

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/testFixtures/java/io/airbyte/integrations/base/destination/typing_deduping/BaseSqlGeneratorIntegrationTest.java

@@ -373,6 +373,11 @@ public void incrementalDedupSameNameNamespace() throws Exception {
     verifyRecordCounts(1, rawRecords, 1, finalRecords);
   }
 
+  private DestinationInitialState getOnly(final List<DestinationInitialState> initialStates) {
+    assertEquals(1, initialStates.size());
+    return initialStates.getFirst();
+  }
+
   /**
    * Run a full T+D update for an incremental-dedup stream, writing to a final table with "_foo"
    * suffix, with values for all data types. Verifies all behaviors for all types:
@@ -398,10 +403,7 @@ public void allTypes() throws Exception {
         streamId,
         BaseTypingDedupingTest.readRecords("sqlgenerator/alltypes_inputrecords.jsonl"));
 
-    List<DestinationInitialState> initialStates =
-        destinationHandler.gatherInitialState(List.of(incrementalDedupStream));
-    assertEquals(1, initialStates.size());
-    DestinationInitialState initialState = initialStates.getFirst();
+    DestinationInitialState initialState = getOnly(destinationHandler.gatherInitialState(List.of(incrementalDedupStream)));
     assertTrue(initialState.isFinalTableEmpty(), "Final table should be empty before T+D");
 
     TypeAndDedupeTransaction.executeTypeAndDedupe(generator, destinationHandler, incrementalDedupStream, Optional.empty(), "");
@@ -411,9 +413,7 @@ public void allTypes() throws Exception {
         dumpRawTableRecords(streamId),
         "sqlgenerator/alltypes_expectedrecords_final.jsonl",
         dumpFinalTableRecords(streamId, ""));
-    initialStates = destinationHandler.gatherInitialState(List.of(incrementalDedupStream));
-    assertEquals(1, initialStates.size());
-    initialState = initialStates.getFirst();
+    initialState = getOnly(destinationHandler.gatherInitialState(List.of(incrementalDedupStream)));
     assertFalse(initialState.isFinalTableEmpty(), "Final table should not be empty after T+D");
   }
 
@@ -428,19 +428,14 @@ public void allTypesUnsafe() throws Exception {
         streamId,
         BaseTypingDedupingTest.readRecords("sqlgenerator/alltypes_unsafe_inputrecords.jsonl"));
 
-    List<DestinationInitialState> initialStates =
-        destinationHandler.gatherInitialState(List.of(incrementalDedupStream));
-    assertEquals(1, initialStates.size());
-    DestinationInitialState initialState = initialStates.getFirst();
+    DestinationInitialState initialState = getOnly(destinationHandler.gatherInitialState(List.of(incrementalDedupStream)));
     assertTrue(initialState.isFinalTableEmpty(), "Final table should be empty before T+D");
 
     // Instead of using the full T+D transaction, explicitly run with useSafeCasting=false.
     final Sql unsafeSql = generator.updateTable(incrementalDedupStream, "", Optional.empty(), false);
     destinationHandler.execute(unsafeSql);
 
-    initialStates = destinationHandler.gatherInitialState(List.of(incrementalDedupStream));
-    assertEquals(1, initialStates.size());
-    initialState = initialStates.getFirst();
+    initialState = getOnly(destinationHandler.gatherInitialState(List.of(incrementalDedupStream)));
     assertFalse(initialState.isFinalTableEmpty(), "Final table should not be empty after T+D");
   }