Fix InventoryDumper first time dump SQL without ORDER BY on multiple columns unique key table (#34736)

Author: sandynz

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java

@@ -332,6 +332,9 @@ private void consumeResultSetToChannel(final PipelineTableMetaData tableMetaData
     private String buildFetchAllSQLWithStreamingQuery() {
         String schemaName = dumperContext.getCommonContext().getTableAndSchemaNameMapper().getSchemaName(dumperContext.getLogicTableName());
         List<String> columnNames = dumperContext.getQueryColumnNames();
+        if (dumperContext.hasUniqueKey()) {
+            return sqlBuilder.buildFetchAllSQL(schemaName, dumperContext.getActualTableName(), columnNames, dumperContext.getUniqueKeyColumns().get(0).getName());
+        }
         return sqlBuilder.buildFetchAllSQL(schemaName, dumperContext.getActualTableName(), columnNames);
     }
 

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineInventoryDumpSQLBuilder.java

@@ -104,4 +104,20 @@ public String buildFetchAllSQL(final String schemaName, final String tableName,
         String queryColumns = columnNames.stream().map(sqlSegmentBuilder::getEscapedIdentifier).collect(Collectors.joining(","));
         return String.format("SELECT %s FROM %s", queryColumns, qualifiedTableName);
     }
+    
+    /**
+     * Build fetch all inventory dump SQL.
+     *
+     * @param schemaName schema name
+     * @param tableName tableName
+     * @param columnNames column names
+     * @param uniqueKey unique key
+     * @return built SQL
+     */
+    public String buildFetchAllSQL(final String schemaName, final String tableName, final Collection<String> columnNames, final String uniqueKey) {
+        String qualifiedTableName = sqlSegmentBuilder.getQualifiedTableName(schemaName, tableName);
+        String queryColumns = columnNames.stream().map(sqlSegmentBuilder::getEscapedIdentifier).collect(Collectors.joining(","));
+        String quotedUniqueKey = sqlSegmentBuilder.getEscapedIdentifier(uniqueKey);
+        return String.format("SELECT %s FROM %s ORDER BY %s ASC", queryColumns, qualifiedTableName, quotedUniqueKey);
+    }
 }

kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineInventoryDumpSQLBuilderTest.java

@@ -65,5 +65,9 @@ void assertBuildFetchAllSQL() {
         assertThat(actual, is("SELECT order_id,user_id,status FROM t_order"));
         actual = sqlBuilder.buildFetchAllSQL(null, "t_order", Collections.singletonList("*"));
         assertThat(actual, is("SELECT * FROM t_order"));
+        actual = sqlBuilder.buildFetchAllSQL(null, "t_order", Arrays.asList("order_id", "user_id", "status"), "order_id");
+        assertThat(actual, is("SELECT order_id,user_id,status FROM t_order ORDER BY order_id ASC"));
+        actual = sqlBuilder.buildFetchAllSQL(null, "t_order", Collections.singletonList("*"), "order_id");
+        assertThat(actual, is("SELECT * FROM t_order ORDER BY order_id ASC"));
     }
 }