Add order by clause in incremental query

Author: tuliren

airbyte-integrations/connectors/source-bigquery/src/main/java/io/airbyte/integrations/source/bigquery/BigQuerySource.java

@@ -138,12 +138,12 @@ public AutoCloseableIterator<JsonNode> queryTableIncremental(final BigQueryDatab
                                                                final String tableName,
                                                                final String cursorField,
                                                                final StandardSQLTypeName cursorFieldType,
-                                                               final String cursor) {
+                                                               final String cursorValue) {
     return queryTableWithParams(database, String.format("SELECT %s FROM %s WHERE %s > ?",
         enquoteIdentifierList(columnNames),
         getFullTableName(schemaName, tableName),
         cursorField),
-        sourceOperations.getQueryParameter(cursorFieldType, cursor));
+        sourceOperations.getQueryParameter(cursorFieldType, cursorValue));
   }
 
   private AutoCloseableIterator<JsonNode> queryTableWithParams(final BigQueryDatabase database,

airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

@@ -266,21 +266,24 @@ public AutoCloseableIterator<JsonNode> queryTableIncremental(final JdbcDatabase
                                                                final String tableName,
                                                                final String cursorField,
                                                                final Datatype cursorFieldType,
-                                                               final String cursor) {
+                                                               final String cursorValue) {
     LOGGER.info("Queueing query for table: {}", tableName);
     return AutoCloseableIterators.lazyIterator(() -> {
       try {
         final Stream<JsonNode> stream = database.unsafeQuery(
             connection -> {
               LOGGER.info("Preparing query for table: {}", tableName);
-              final String sql = String.format("SELECT %s FROM %s WHERE %s > ?",
+              final String quotedCursorField = sourceOperations.enquoteIdentifier(connection, cursorField);
+              final String sql = String.format("SELECT %s FROM %s WHERE %s > ? ORDER BY %s ASC",
                   sourceOperations.enquoteIdentifierList(connection, columnNames),
                   sourceOperations
                       .getFullyQualifiedTableNameWithQuoting(connection, schemaName, tableName),
-                  sourceOperations.enquoteIdentifier(connection, cursorField));
+                  quotedCursorField,
+                  quotedCursorField
+              );
 
               final PreparedStatement preparedStatement = connection.prepareStatement(sql);
-              sourceOperations.setStatementField(preparedStatement, 1, cursorFieldType, cursor);
+              sourceOperations.setStatementField(preparedStatement, 1, cursorFieldType, cursorValue);
               LOGGER.info("Executing query for table: {}", tableName);
               return preparedStatement;
             },

airbyte-integrations/connectors/source-mongodb-v2/src/main/java/io.airbyte.integrations.source.mongodb/MongoDbSource.java

@@ -186,8 +186,8 @@ public AutoCloseableIterator<JsonNode> queryTableIncremental(final MongoDatabase
                                                                final String tableName,
                                                                final String cursorField,
                                                                final BsonType cursorFieldType,
-                                                               final String cursor) {
-    final Bson greaterComparison = gt(cursorField, MongoUtils.getBsonValue(cursorFieldType, cursor));
+                                                               final String cursorValue) {
+    final Bson greaterComparison = gt(cursorField, MongoUtils.getBsonValue(cursorFieldType, cursorValue));
     return queryTable(database, columnNames, tableName, greaterComparison);
   }
 

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java

@@ -97,7 +97,7 @@ public AutoCloseableIterator<JsonNode> queryTableIncremental(final JdbcDatabase
                                                                final String tableName,
                                                                final String cursorField,
                                                                final JDBCType cursorFieldType,
-                                                               final String cursor) {
+                                                               final String cursorValue) {
     LOGGER.info("Queueing query for table: {}", tableName);
     return AutoCloseableIterators.lazyIterator(() -> {
       try {
@@ -115,7 +115,7 @@ public AutoCloseableIterator<JsonNode> queryTableIncremental(final JdbcDatabase
               LOGGER.info("Prepared SQL query for queryTableIncremental is: " + sql);
 
               final PreparedStatement preparedStatement = connection.prepareStatement(sql);
-              sourceOperations.setStatementField(preparedStatement, 1, cursorFieldType, cursor);
+              sourceOperations.setStatementField(preparedStatement, 1, cursorFieldType, cursorValue);
               LOGGER.info("Executing query for table: {}", tableName);
               return preparedStatement;
             },
@@ -279,8 +279,8 @@ protected void assertCdcSchemaQueryable(final JsonNode config, final JdbcDatabas
     final List<JsonNode> queryResponse = database.queryJsons(connection -> {
       boolean isAzureSQL = false;
 
-      try (Statement stmt = connection.createStatement();
-          ResultSet editionRS = stmt.executeQuery("SELECT ServerProperty('Edition')")) {
+      try (final Statement stmt = connection.createStatement();
+          final ResultSet editionRS = stmt.executeQuery("SELECT ServerProperty('Edition')")) {
         isAzureSQL = editionRS.next() && "SQL Azure".equals(editionRS.getString(1));
       }
 

airbyte-integrations/connectors/source-relational-db/src/main/java/io/airbyte/integrations/source/relationaldb/AbstractDbSource.java

@@ -239,8 +239,7 @@ protected AutoCloseableIterator<AirbyteMessage> createReadIterator(final Databas
         airbyteMessageIterator = getFullRefreshStream(database, streamName, namespace, selectedDatabaseFields, table, emittedAt);
       }
 
-      final JsonSchemaPrimitive cursorType = IncrementalUtils
-          .getCursorType(airbyteStream, cursorField);
+      final JsonSchemaPrimitive cursorType = IncrementalUtils.getCursorType(airbyteStream, cursorField);
 
       iterator = AutoCloseableIterators.transform(autoCloseableIterator -> new StateDecoratingIterator(
           autoCloseableIterator,
@@ -494,13 +493,6 @@ public abstract AutoCloseableIterator<JsonNode> queryTableFullRefresh(final Data
    * Read incremental data from a table. Incremental read should returns only records where cursor
    * column value is bigger than cursor.
    *
-   * @param database source database
-   * @param columnNames interested column names
-   * @param schemaName table namespace
-   * @param tableName target table
-   * @param cursorField cursor field name
-   * @param cursorFieldType cursor field type
-   * @param cursor cursor value
    * @return iterator with read data
    */
   public abstract AutoCloseableIterator<JsonNode> queryTableIncremental(Database database,
@@ -509,7 +501,7 @@ public abstract AutoCloseableIterator<JsonNode> queryTableIncremental(Database d
                                                                         String tableName,
                                                                         String cursorField,
                                                                         DataType cursorFieldType,
-                                                                        String cursor);
+                                                                        String cursorValue);
 
   private Database createDatabaseInternal(final JsonNode sourceConfig) throws Exception {
     final Database database = createDatabase(sourceConfig);