Updated query to use TOP instead of LIMIT

airbytehq · Jan 12, 2024 · dbe4d60 · dbe4d60
1 parent 7b4dfea
commit dbe4d60
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 4 deletions.
diff --git a/...ctors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcHelper.java b/...ctors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcHelper.java
@@ -57,7 +57,7 @@ static boolean isCdc(final JsonNode config) {
     return false;
   }
 
-  static Properties getDebeziumProperties(final JdbcDatabase database, final ConfiguredAirbyteCatalog catalog, final boolean isSnapshot) {
+  public static Properties getDebeziumProperties(final JdbcDatabase database, final ConfiguredAirbyteCatalog catalog, final boolean isSnapshot) {
     final JsonNode config = database.getSourceConfig();
     final JsonNode dbConfig = database.getDatabaseConfig();
 

diff --git a/...urce-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlDebeziumConverter.java b/...urce-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlDebeziumConverter.java
@@ -38,7 +38,7 @@ public class MssqlDebeziumConverter implements CustomConverter<SchemaBuilder, Re
   private static final String SMALLMONEY_TYPE = "SMALLMONEY";
   private static final String GEOMETRY = "GEOMETRY";
   private static final String GEOGRAPHY = "GEOGRAPHY";
-  private static final String DEBEZIUM_DATETIMEOFFSET_FORMAT = "yyyy-MM-dd HH:mm:ss XXX";
+  private static final String DEBEZIUM_DATETIMEOFFSET_FORMAT = "yyyy-MM-dd HH:mm:ss[.][SSSSSSS] XXX";
 
   private static final String DATETIME_FORMAT_MICROSECONDS = "yyyy-MM-dd'T'HH:mm:ss[.][SSSSSS]";
 
@@ -144,6 +144,7 @@ private void registerDateTimeOffSet(final RelationalColumn field,
       }
 
       if (input instanceof DateTimeOffset) {
+        LOGGER.info("*** input:{}", input);
         return DataTypeUtils.toISO8601String(
             OffsetDateTime.parse(input.toString(),
                 DateTimeFormatter.ofPattern(DEBEZIUM_DATETIMEOFFSET_FORMAT)));

diff --git a/...nnectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java b/...nnectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java
@@ -470,8 +470,6 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(final
                                                                              final Map<String, TableInfo<CommonField<JDBCType>>> tableNameToTable,
                                                                              final StateManager stateManager,
                                                                              final Instant emittedAt) {
-    // rewire snapshot here
-    // Duy add user cursor here
     final JsonNode sourceConfig = database.getSourceConfig();
     if (MssqlCdcHelper.isCdc(sourceConfig) && isAnyStreamIncrementalSyncMode(catalog)) {
       LOGGER.info("using OC + CDC");