Fix the mismatch between processed record count and actual data count…

… for inventory dumper (apache#33996)
jiangML · Dec 11, 2024 · 7562fcf · 7562fcf
1 parent 8cb4939
commit 7562fcf
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 4 deletions.
diff --git a/...org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java b/...org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java
@@ -144,7 +144,8 @@ private void dumpByPage(final Connection connection, final PipelineTableMetaData
         AtomicLong rowCount = new AtomicLong();
         IngestPosition position = dumperContext.getCommonContext().getPosition();
         do {
-            QueryRange queryRange = new QueryRange(((PrimaryKeyIngestPosition<?>) position).getBeginValue(), firstQuery, ((PrimaryKeyIngestPosition<?>) position).getEndValue());
+            QueryRange queryRange = new QueryRange(((PrimaryKeyIngestPosition<?>) position).getBeginValue(), firstQuery && dumperContext.isFirstDump(),
+                    ((PrimaryKeyIngestPosition<?>) position).getEndValue());
             InventoryQueryParameter<?> queryParam = new InventoryRangeQueryParameter(queryRange);
             List<Record> dataRecords = dumpByPage(connection, queryParam, rowCount, tableMetaData);
             if (dataRecords.size() > 1 && Objects.deepEquals(getFirstUniqueKeyValue(dataRecords, 0), getFirstUniqueKeyValue(dataRecords, dataRecords.size() - 1))) {
@@ -169,7 +170,8 @@ private List<Record> dumpByPage(final Connection connection,
                                     final InventoryQueryParameter<?> queryParam, final AtomicLong rowCount, final PipelineTableMetaData tableMetaData) throws SQLException {
         DatabaseType databaseType = dumperContext.getCommonContext().getDataSourceConfig().getDatabaseType();
         int batchSize = dumperContext.getBatchSize();
-        try (PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(databaseType, connection, buildDumpByPageSQL(queryParam), batchSize)) {
+        String sql = buildDumpByPageSQL(queryParam);
+        try (PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(databaseType, connection, sql, batchSize)) {
             runningStatement.set(preparedStatement);
             setParameters(preparedStatement, queryParam);
             try (ResultSet resultSet = preparedStatement.executeQuery()) {

diff --git a/...che/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumperContext.java b/...che/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumperContext.java
@@ -58,6 +58,8 @@ public final class InventoryDumperContext {
 
     private JobRateLimitAlgorithm rateLimitAlgorithm;
 
+    private boolean firstDump = true;
+
     public InventoryDumperContext(final DumperCommonContext commonContext) {
         this.commonContext = new DumperCommonContext(
                 commonContext.getDataSourceName(), commonContext.getDataSourceConfig(), commonContext.getTableNameMapper(), commonContext.getTableAndSchemaNameMapper());

diff --git a/...sphere/data/pipeline/core/preparer/inventory/splitter/InventoryDumperContextSplitter.java b/...sphere/data/pipeline/core/preparer/inventory/splitter/InventoryDumperContextSplitter.java
@@ -93,7 +93,7 @@ private Collection<InventoryDumperContext> splitByPrimaryKey(final InventoryDump
         JobRateLimitAlgorithm rateLimitAlgorithm = jobProcessContext.getReadRateLimitAlgorithm();
         int i = 0;
         for (IngestPosition each : getInventoryPositions(dumperContext, jobItemContext)) {
-            result.add(createPrimaryKeySplitDumperContext(dumperContext, each, i++, batchSize, rateLimitAlgorithm));
+            result.add(createPrimaryKeySplitDumperContext(dumperContext, each, i++, batchSize, rateLimitAlgorithm, jobItemContext));
         }
         return result;
     }
@@ -149,7 +149,8 @@ private Range<Long> getUniqueKeyValuesRange(final TransmissionJobItemContext job
     }
 
     private InventoryDumperContext createPrimaryKeySplitDumperContext(final InventoryDumperContext dumperContext, final IngestPosition position,
-                                                                      final int shardingItem, final int batchSize, final JobRateLimitAlgorithm rateLimitAlgorithm) {
+                                                                      final int shardingItem, final int batchSize, final JobRateLimitAlgorithm rateLimitAlgorithm,
+                                                                      final TransmissionJobItemContext jobItemContext) {
         InventoryDumperContext result = new InventoryDumperContext(dumperContext.getCommonContext());
         result.getCommonContext().setPosition(position);
         result.setShardingItem(shardingItem);
@@ -159,6 +160,11 @@ private InventoryDumperContext createPrimaryKeySplitDumperContext(final Inventor
         result.setInsertColumnNames(dumperContext.getInsertColumnNames());
         result.setBatchSize(batchSize);
         result.setRateLimitAlgorithm(rateLimitAlgorithm);
+        result.setFirstDump(isFirstDump(jobItemContext));
         return result;
     }
+
+    private boolean isFirstDump(final TransmissionJobItemContext jobItemContext) {
+        return null == jobItemContext.getInitProgress() && jobItemContext.getProcessedRecordsCount() == 0;
+    }
 }