You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by me...@apache.org on 2025/06/18 03:52:26 UTC
(shardingsphere) branch master updated: Improve data consistency check DATA_MATCH impl (#35740)
This is an automated email from the ASF dual-hosted git repository.
menghaoran pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 9d4373e652f Improve data consistency check DATA_MATCH impl (#35740)
9d4373e652f is described below
commit 9d4373e652fc10a380bb8e475a9e4be6f41d8791
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Wed Jun 18 11:52:20 2025 +0800
Improve data consistency check DATA_MATCH impl (#35740)
* Improve data consistency check DATA_MATCH impl performance
* Clean SingleTableInventoryCalculateParameter old constructor
* Refactor readRecord
* Update code style
---
.../DataConsistencyCheckUtils.java | 18 +-
.../table/MatchingTableInventoryChecker.java | 8 +-
.../table/calculator/CalculationContext.java | 52 ++--
.../RecordSingleTableInventoryCalculator.java | 240 ++++++++++++-------
.../SingleTableInventoryCalculateParameter.java | 10 -
.../DataConsistencyCheckUtilsTest.java | 15 ++
.../CRC32SingleTableInventoryCalculatorTest.java | 3 +-
.../RecordSingleTableInventoryCalculatorTest.java | 262 ++++++++++++++++++---
8 files changed, 466 insertions(+), 142 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtils.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtils.java
index 89154fcd90f..afa37863d18 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtils.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtils.java
@@ -22,8 +22,8 @@ import lombok.NoArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.builder.EqualsBuilder;
-
import org.jspecify.annotations.Nullable;
+
import java.math.BigDecimal;
import java.math.BigInteger;
import java.math.RoundingMode;
@@ -69,6 +69,22 @@ public final class DataConsistencyCheckUtils {
return true;
}
+ /**
+ * Is first unique key value matched.
+ *
+ * @param thisRecord this record
+ * @param thatRecord that record
+ * @param uniqueKey unique key
+ * @param equalsBuilder equals builder
+ * @return true if matched, otherwise false
+ */
+ public static boolean isFirstUniqueKeyValueMatched(final Map<String, Object> thisRecord, final Map<String, Object> thatRecord, final String uniqueKey, final EqualsBuilder equalsBuilder) {
+ if (thisRecord.isEmpty() || thatRecord.isEmpty()) {
+ return false;
+ }
+ return isMatched(equalsBuilder, getFirstUniqueKeyValue(thisRecord, uniqueKey), getFirstUniqueKeyValue(thatRecord, uniqueKey));
+ }
+
/**
* Whether column values are matched or not.
*
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java
index f17f66b6914..4d56aafcc0a 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableInventoryChecker.java
@@ -28,6 +28,8 @@ import org.apache.shardingsphere.data.pipeline.core.consistencycheck.result.yaml
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator.SingleTableInventoryCalculateParameter;
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator.SingleTableInventoryCalculator;
import org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperationType;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.query.QueryType;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.query.range.QueryRange;
import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobUpdateProgress;
import org.apache.shardingsphere.infra.exception.core.external.sql.type.kernel.category.PipelineSQLException;
import org.apache.shardingsphere.infra.exception.core.external.sql.type.wrapper.SQLWrapperException;
@@ -76,9 +78,11 @@ public abstract class MatchingTableInventoryChecker implements TableInventoryChe
private TableDataConsistencyCheckResult checkSingleTableInventoryData(final TableInventoryCheckParameter param, final ThreadPoolExecutor executor) {
SingleTableInventoryCalculateParameter sourceParam = new SingleTableInventoryCalculateParameter(param.getSourceDataSource(), param.getSourceTable(),
- param.getColumnNames(), param.getUniqueKeys(), param.getProgressContext().getSourceTableCheckPositions().get(param.getSourceTable().getTableName()));
+ param.getColumnNames(), param.getUniqueKeys(), QueryType.RANGE_QUERY);
+ sourceParam.setQueryRange(new QueryRange(param.getProgressContext().getSourceTableCheckPositions().get(param.getSourceTable().getTableName()), false, null));
SingleTableInventoryCalculateParameter targetParam = new SingleTableInventoryCalculateParameter(param.getTargetDataSource(), param.getTargetTable(),
- param.getColumnNames(), param.getUniqueKeys(), param.getProgressContext().getTargetTableCheckPositions().get(param.getTargetTable().getTableName()));
+ param.getColumnNames(), param.getUniqueKeys(), QueryType.RANGE_QUERY);
+ targetParam.setQueryRange(new QueryRange(param.getProgressContext().getTargetTableCheckPositions().get(param.getTargetTable().getTableName()), false, null));
SingleTableInventoryCalculator sourceCalculator = buildSingleTableInventoryCalculator();
this.sourceCalculator = sourceCalculator;
SingleTableInventoryCalculator targetCalculator = buildSingleTableInventoryCalculator();
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CalculationContext.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CalculationContext.java
index f0297044bf8..927d3ba490f 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CalculationContext.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CalculationContext.java
@@ -17,11 +17,15 @@
package org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator;
+import lombok.Getter;
import org.apache.shardingsphere.infra.util.close.QuietlyCloser;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -36,7 +40,10 @@ public final class CalculationContext implements AutoCloseable {
private final AtomicReference<ResultSet> resultSet = new AtomicReference<>();
- private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final AtomicBoolean databaseResourcesReady = new AtomicBoolean(false);
+
+ @Getter
+ private final Deque<Map<String, Object>> recordDeque = new LinkedList<>();
/**
* Get connection.
@@ -57,21 +64,21 @@ public final class CalculationContext implements AutoCloseable {
}
/**
- * Get result set.
+ * Set prepared statement.
*
- * @return result set
+ * @param preparedStatement prepared statement
*/
- public ResultSet getResultSet() {
- return resultSet.get();
+ public void setPreparedStatement(final PreparedStatement preparedStatement) {
+ this.preparedStatement.set(preparedStatement);
}
/**
- * Set prepared statement.
+ * Get result set.
*
- * @param preparedStatement prepared statement
+ * @return result set
*/
- public void setPreparedStatement(final PreparedStatement preparedStatement) {
- this.preparedStatement.set(preparedStatement);
+ public ResultSet getResultSet() {
+ return resultSet.get();
}
/**
@@ -84,17 +91,34 @@ public final class CalculationContext implements AutoCloseable {
}
/**
- * Is closed.
+ * Check if database resources are ready.
*
- * @return closed or not
+ * @return true if database resources are ready, false otherwise
*/
- public boolean isClosed() {
- return closed.get();
+ public boolean isDatabaseResourcesReady() {
+ return databaseResourcesReady.get();
+ }
+
+ /**
+ * Set database resources ready.
+ *
+ * @param databaseResourcesReady true if database resources are ready, false otherwise
+ */
+ public void setDatabaseResourcesReady(final boolean databaseResourcesReady) {
+ this.databaseResourcesReady.set(databaseResourcesReady);
}
@Override
public void close() {
- closed.set(true);
+ resetDatabaseResources();
+ recordDeque.clear();
+ }
+
+ /**
+ * Reset database resources.
+ */
+ public void resetDatabaseResources() {
+ setDatabaseResourcesReady(false);
QuietlyCloser.close(resultSet.get());
QuietlyCloser.close(preparedStatement.get());
QuietlyCloser.close(connection.get());
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java
index 5862e9b5c8f..338fd6abfbd 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java
@@ -18,14 +18,15 @@
package org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator;
import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.DataConsistencyCheckUtils;
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.result.RecordSingleTableInventoryCalculatedResult;
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.result.SingleTableInventoryCalculatedResult;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCancelingException;
import org.apache.shardingsphere.data.pipeline.core.exception.data.PipelineTableDataConsistencyCheckLoadingFailedException;
-import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.query.range.QueryRange;
-import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.query.QueryType;
import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.column.InventoryColumnValueReaderEngine;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.query.QueryType;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.query.range.QueryRange;
import org.apache.shardingsphere.data.pipeline.core.query.JDBCStreamQueryBuilder;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql.PipelineDataConsistencyCalculateSQLBuilder;
import org.apache.shardingsphere.infra.annotation.HighFrequencyInvocation;
@@ -40,11 +41,11 @@ import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Collections;
+import java.util.Deque;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
/**
@@ -56,53 +57,38 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
private final int chunkSize;
+ private final int streamingChunkCount;
+
+ private final EqualsBuilder equalsBuilder = new EqualsBuilder();
+
+ public RecordSingleTableInventoryCalculator(final int chunkSize) {
+ this.chunkSize = chunkSize;
+ streamingChunkCount = 100;
+ }
+
@Override
public Optional<SingleTableInventoryCalculatedResult> calculateChunk(final SingleTableInventoryCalculateParameter param) {
- List<Map<String, Object>> records = calculateChunk(param, QueryType.RANGE_QUERY == param.getQueryType());
+ List<Map<String, Object>> records = calculateChunk0(param);
if (records.isEmpty()) {
return Optional.empty();
}
String firstUniqueKey = param.getFirstUniqueKey().getName();
- if (QueryType.POINT_QUERY == param.getQueryType()) {
- return convertRecordsToResult(records, firstUniqueKey);
- }
- if (records.size() == chunkSize) {
- Object minUniqueKeyValue = DataConsistencyCheckUtils.getFirstUniqueKeyValue(records.get(0), firstUniqueKey);
- removeLastRecords(records, param);
- if (!records.isEmpty()) {
- updateQueryRangeLower(param, records, firstUniqueKey);
- return convertRecordsToResult(records, firstUniqueKey);
- }
- SingleTableInventoryCalculateParameter newParam = buildNewCalculateParameter(param, minUniqueKeyValue);
- records = calculateChunk(newParam, false);
- if (!records.isEmpty()) {
- updateQueryRangeLower(param, records, firstUniqueKey);
- return convertRecordsToResult(records, firstUniqueKey);
- }
- return Optional.empty();
+ if (QueryType.RANGE_QUERY == param.getQueryType()) {
+ updateQueryRangeLower(param, records, firstUniqueKey);
}
- updateQueryRangeLower(param, records, firstUniqueKey);
return convertRecordsToResult(records, firstUniqueKey);
}
- private List<Map<String, Object>> calculateChunk(final SingleTableInventoryCalculateParameter param, final boolean isRangeQuery) {
- try (CalculationContext calculationContext = getOrCreateCalculationContext(param)) {
- List<Map<String, Object>> result = new LinkedList<>();
- InventoryColumnValueReaderEngine columnValueReaderEngine = new InventoryColumnValueReaderEngine(param.getDatabaseType());
- ResultSet resultSet = calculationContext.getResultSet();
- ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
- while (resultSet.next()) {
- ShardingSpherePreconditions.checkState(!isCanceling(), () -> new PipelineJobCancelingException("Calculate chunk canceled, qualified table: %s", param.getTable()));
- Map<String, Object> columnRecord = new LinkedHashMap<>();
- for (int columnIndex = 1, columnCount = resultSetMetaData.getColumnCount(); columnIndex <= columnCount; columnIndex++) {
- columnRecord.put(resultSetMetaData.getColumnLabel(columnIndex), columnValueReaderEngine.read(resultSet, resultSetMetaData, columnIndex));
- }
- result.add(columnRecord);
- if (isRangeQuery && result.size() == chunkSize) {
- break;
- }
+ private List<Map<String, Object>> calculateChunk0(final SingleTableInventoryCalculateParameter param) {
+ InventoryColumnValueReaderEngine columnValueReaderEngine = new InventoryColumnValueReaderEngine(param.getDatabaseType());
+ try {
+ if (QueryType.POINT_QUERY == param.getQueryType()) {
+ return pointQuery(param, columnValueReaderEngine);
}
- return result;
+ if (param.getUniqueKeys().size() <= 1) {
+ return rangeQueryWithSingleColumUniqueKey(param, columnValueReaderEngine, 1);
+ }
+ return rangeQueryWithMultiColumUniqueKeys(param, columnValueReaderEngine);
} catch (final PipelineSQLException | PipelineJobCancelingException ex) {
throw ex;
// CHECKSTYLE:OFF
@@ -112,39 +98,136 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
}
}
- private CalculationContext getOrCreateCalculationContext(final SingleTableInventoryCalculateParameter param) {
- CalculationContext result = (CalculationContext) param.getCalculationContext();
- if (null != result && !result.isClosed()) {
- return result;
+ private List<Map<String, Object>> pointQuery(final SingleTableInventoryCalculateParameter param, final InventoryColumnValueReaderEngine columnValueReaderEngine) throws SQLException {
+ List<Map<String, Object>> result = new LinkedList<>();
+ CalculationContext calculationContext = prepareCalculationContext(param);
+ prepareDatabaseResources(calculationContext, param);
+ ResultSet resultSet = calculationContext.getResultSet();
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ while (resultSet.next()) {
+ ShardingSpherePreconditions.checkState(!isCanceling(), () -> new PipelineJobCancelingException("Calculate chunk canceled, qualified table: %s", param.getTable()));
+ Map<String, Object> record = readRecord(columnValueReaderEngine, resultSet, resultSetMetaData);
+ result.add(record);
}
+ return result;
+ }
+
+ private List<Map<String, Object>> rangeQueryWithSingleColumUniqueKey(final SingleTableInventoryCalculateParameter param,
+ final InventoryColumnValueReaderEngine columnValueReaderEngine, final int round) throws SQLException {
+ List<Map<String, Object>> result = new LinkedList<>();
+ CalculationContext calculationContext = prepareCalculationContext(param);
+ prepareDatabaseResources(calculationContext, param);
+ ResultSet resultSet = calculationContext.getResultSet();
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ while (resultSet.next()) {
+ ShardingSpherePreconditions.checkState(!isCanceling(), () -> new PipelineJobCancelingException("Calculate chunk canceled, qualified table: %s", param.getTable()));
+ result.add(readRecord(columnValueReaderEngine, resultSet, resultSetMetaData));
+ if (result.size() == chunkSize) {
+ return result;
+ }
+ }
+ calculationContext.resetDatabaseResources();
+ if (result.isEmpty() && 1 == round) {
+ return rangeQueryWithSingleColumUniqueKey(param, columnValueReaderEngine, round + 1);
+ }
+ return result;
+ }
+
+ private List<Map<String, Object>> rangeQueryWithMultiColumUniqueKeys(final SingleTableInventoryCalculateParameter param,
+ final InventoryColumnValueReaderEngine columnValueReaderEngine) throws SQLException {
+ CalculationContext calculationContext = prepareCalculationContext(param);
+ if (calculationContext.getRecordDeque().size() > chunkSize) {
+ return queryFromBuffer(calculationContext.getRecordDeque());
+ }
+ doRangeQueryWithMultiColumUniqueKeys(param, calculationContext, columnValueReaderEngine);
+ return queryFromBuffer(calculationContext.getRecordDeque());
+ }
+
+ private void doRangeQueryWithMultiColumUniqueKeys(final SingleTableInventoryCalculateParameter param, final CalculationContext calculationContext,
+ final InventoryColumnValueReaderEngine columnValueReaderEngine) throws SQLException {
+ prepareDatabaseResources(calculationContext, param);
+ ResultSet resultSet = calculationContext.getResultSet();
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<String, Object> previousRecord = calculationContext.getRecordDeque().pollLast();
+ List<Map<String, Object>> duplicateRecords = new LinkedList<>();
+ if (null != previousRecord) {
+ duplicateRecords.add(previousRecord);
+ }
+ while (resultSet.next()) {
+ ShardingSpherePreconditions.checkState(!isCanceling(), () -> new PipelineJobCancelingException("Calculate chunk canceled, qualified table: %s", param.getTable()));
+ Map<String, Object> record = readRecord(columnValueReaderEngine, resultSet, resultSetMetaData);
+ if (null == previousRecord || DataConsistencyCheckUtils.isFirstUniqueKeyValueMatched(previousRecord, record, param.getFirstUniqueKey().getName(), equalsBuilder)) {
+ duplicateRecords.add(record);
+ previousRecord = record;
+ continue;
+ }
+ previousRecord = record;
+ if (!duplicateRecords.isEmpty()) {
+ calculationContext.getRecordDeque().addAll(duplicateRecords);
+ duplicateRecords.clear();
+ }
+ if (calculationContext.getRecordDeque().size() >= chunkSize) {
+ calculationContext.getRecordDeque().add(record);
+ return;
+ }
+ duplicateRecords.add(record);
+ }
+ calculationContext.resetDatabaseResources();
+ if (!duplicateRecords.isEmpty()) {
+ calculationContext.getRecordDeque().addAll(pointRangeQuery(param, duplicateRecords.get(0), columnValueReaderEngine));
+ }
+ }
+
+ private List<Map<String, Object>> pointRangeQuery(final SingleTableInventoryCalculateParameter param, final Map<String, Object> duplicateRecord,
+ final InventoryColumnValueReaderEngine columnValueReaderEngine) throws SQLException {
+ Object duplicateUniqueKeyValue = DataConsistencyCheckUtils.getFirstUniqueKeyValue(duplicateRecord, param.getFirstUniqueKey().getName());
+ SingleTableInventoryCalculateParameter newParam = buildPointRangeQueryCalculateParameter(param, duplicateUniqueKeyValue);
try {
- result = createCalculationContext(param);
- fulfillCalculationContext(result, param);
- // CHECKSTYLE:OFF
- } catch (final SQLException | RuntimeException ex) {
- // CHECKSTYLE:ON
- QuietlyCloser.close(result);
- throw new PipelineTableDataConsistencyCheckLoadingFailedException(param.getTable(), ex);
+ return pointQuery(newParam, columnValueReaderEngine);
+ } finally {
+ QuietlyCloser.close(newParam.getCalculationContext());
+ }
+ }
+
+ private List<Map<String, Object>> queryFromBuffer(final Deque<Map<String, Object>> recordDeque) {
+ List<Map<String, Object>> result = new LinkedList<>();
+ while (true) {
+ Map<String, Object> record = recordDeque.pollFirst();
+ if (null == record) {
+ break;
+ }
+ result.add(record);
+ if (result.size() == chunkSize) {
+ break;
+ }
}
return result;
}
- private CalculationContext createCalculationContext(final SingleTableInventoryCalculateParameter param) throws SQLException {
- Connection connection = param.getDataSource().getConnection();
- CalculationContext result = new CalculationContext();
- result.setConnection(connection);
+ private CalculationContext prepareCalculationContext(final SingleTableInventoryCalculateParameter param) {
+ CalculationContext result = (CalculationContext) param.getCalculationContext();
+ if (null != result) {
+ return result;
+ }
+ result = new CalculationContext();
param.setCalculationContext(result);
return result;
}
- private void fulfillCalculationContext(final CalculationContext calculationContext, final SingleTableInventoryCalculateParameter param) throws SQLException {
+ private void prepareDatabaseResources(final CalculationContext calculationContext, final SingleTableInventoryCalculateParameter param) throws SQLException {
+ if (calculationContext.isDatabaseResourcesReady()) {
+ return;
+ }
+ Connection connection = param.getDataSource().getConnection();
+ calculationContext.setConnection(connection);
String sql = getQuerySQL(param);
- PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(param.getDatabaseType(), calculationContext.getConnection(), sql, chunkSize);
+ PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(param.getDatabaseType(), connection, sql, chunkSize);
setCurrentStatement(preparedStatement);
calculationContext.setPreparedStatement(preparedStatement);
setParameters(preparedStatement, param);
ResultSet resultSet = preparedStatement.executeQuery();
calculationContext.setResultSet(resultSet);
+ calculationContext.setDatabaseResourcesReady(true);
}
private String getQuerySQL(final SingleTableInventoryCalculateParameter param) {
@@ -175,7 +258,7 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
if (null != queryRange.getUpper()) {
preparedStatement.setObject(parameterIndex++, queryRange.getUpper());
}
- preparedStatement.setObject(parameterIndex, chunkSize);
+ preparedStatement.setObject(parameterIndex, chunkSize * streamingChunkCount);
} else if (queryType == QueryType.POINT_QUERY) {
Collection<Object> uniqueKeysValues = param.getUniqueKeysValues();
ShardingSpherePreconditions.checkNotNull(uniqueKeysValues,
@@ -184,8 +267,10 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
for (Object each : uniqueKeysValues) {
preparedStatement.setObject(parameterIndex++, each);
}
- if (!param.getShardingColumnsNames().isEmpty()) {
+ if (null != param.getShardingColumnsNames() && !param.getShardingColumnsNames().isEmpty()) {
List<Object> shardingColumnsValues = param.getShardingColumnsValues();
+ ShardingSpherePreconditions.checkNotNull(shardingColumnsValues,
+ () -> new PipelineTableDataConsistencyCheckLoadingFailedException(param.getTable(), new RuntimeException("Sharding columns values is null when names not empty.")));
for (Object each : shardingColumnsValues) {
preparedStatement.setObject(parameterIndex++, each);
}
@@ -195,39 +280,30 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
}
}
- private void removeLastRecords(final List<Map<String, Object>> records, final SingleTableInventoryCalculateParameter param) {
- Object minUniqueKeyValue = DataConsistencyCheckUtils.getFirstUniqueKeyValue(records.get(0), param.getFirstUniqueKey().getName());
- Object maxUniqueKeyValue = DataConsistencyCheckUtils.getFirstUniqueKeyValue(records.get(records.size() - 1), param.getFirstUniqueKey().getName());
- if (Objects.equals(minUniqueKeyValue, maxUniqueKeyValue)) {
- records.clear();
- return;
- }
- records.remove(records.size() - 1);
- for (int i = records.size() - 1; i >= 0; i--) {
- if (Objects.deepEquals(maxUniqueKeyValue, DataConsistencyCheckUtils.getFirstUniqueKeyValue(records.get(i), param.getFirstUniqueKey().getName()))) {
- records.remove(i);
- } else {
- break;
- }
- }
- }
-
- private Optional<SingleTableInventoryCalculatedResult> convertRecordsToResult(final List<Map<String, Object>> records, final String firstUniqueKey) {
- Object maxUniqueKeyValue = DataConsistencyCheckUtils.getFirstUniqueKeyValue(records.get(records.size() - 1), firstUniqueKey);
- return Optional.of(new RecordSingleTableInventoryCalculatedResult(maxUniqueKeyValue, records));
- }
-
- private SingleTableInventoryCalculateParameter buildNewCalculateParameter(final SingleTableInventoryCalculateParameter param, final Object minUniqueKeyValue) {
+ private SingleTableInventoryCalculateParameter buildPointRangeQueryCalculateParameter(final SingleTableInventoryCalculateParameter param, final Object uniqueKeyValue) {
SingleTableInventoryCalculateParameter result = new SingleTableInventoryCalculateParameter(param.getDataSource(), param.getTable(), param.getColumnNames(),
Collections.singletonList(param.getFirstUniqueKey()), QueryType.POINT_QUERY);
- result.setUniqueKeysValues(Collections.singletonList(minUniqueKeyValue));
+ result.setUniqueKeysValues(Collections.singletonList(uniqueKeyValue));
result.setShardingColumnsNames(param.getShardingColumnsNames());
result.setShardingColumnsValues(param.getShardingColumnsValues());
return result;
}
+ private Map<String, Object> readRecord(final InventoryColumnValueReaderEngine columnValueReaderEngine, final ResultSet resultSet, final ResultSetMetaData resultSetMetaData) throws SQLException {
+ Map<String, Object> result = new LinkedHashMap<>();
+ for (int columnIndex = 1, columnCount = resultSetMetaData.getColumnCount(); columnIndex <= columnCount; columnIndex++) {
+ result.put(resultSetMetaData.getColumnLabel(columnIndex), columnValueReaderEngine.read(resultSet, resultSetMetaData, columnIndex));
+ }
+ return result;
+ }
+
private void updateQueryRangeLower(final SingleTableInventoryCalculateParameter param, final List<Map<String, Object>> records, final String firstUniqueKey) {
Object maxUniqueKeyValue = DataConsistencyCheckUtils.getFirstUniqueKeyValue(records.get(records.size() - 1), firstUniqueKey);
param.setQueryRange(new QueryRange(maxUniqueKeyValue, false, param.getQueryRange().getUpper()));
}
+
+ private Optional<SingleTableInventoryCalculatedResult> convertRecordsToResult(final List<Map<String, Object>> records, final String firstUniqueKey) {
+ Object maxUniqueKeyValue = DataConsistencyCheckUtils.getFirstUniqueKeyValue(records.get(records.size() - 1), firstUniqueKey);
+ return Optional.of(new RecordSingleTableInventoryCalculatedResult(maxUniqueKeyValue, records));
+ }
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/SingleTableInventoryCalculateParameter.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/SingleTableInventoryCalculateParameter.java
index 00623221a3d..6b417d65a07 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/SingleTableInventoryCalculateParameter.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/SingleTableInventoryCalculateParameter.java
@@ -68,16 +68,6 @@ public final class SingleTableInventoryCalculateParameter {
private final QueryType queryType;
- public SingleTableInventoryCalculateParameter(final PipelineDataSource dataSource,
- final QualifiedTable table, final List<String> columnNames, final List<PipelineColumnMetaData> uniqueKeys, final Object tableCheckPosition) {
- this.dataSource = dataSource;
- this.table = table;
- this.columnNames = columnNames;
- this.uniqueKeys = uniqueKeys;
- queryType = QueryType.RANGE_QUERY;
- setQueryRange(new QueryRange(tableCheckPosition, false, null));
- }
-
/**
* Get database type.
*
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtilsTest.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtilsTest.java
index a494d544d84..37706c513cb 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtilsTest.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/DataConsistencyCheckUtilsTest.java
@@ -17,17 +17,23 @@
package org.apache.shardingsphere.data.pipeline.core.consistencycheck;
+import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang3.builder.EqualsBuilder;
import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class DataConsistencyCheckUtilsTest {
+ private static final List<String> UNIQUE_KEYS_NAMES = Arrays.asList("order_id", "user_id");
+
@Test
void assertIsIntegerEquals() {
EqualsBuilder equalsBuilder = new EqualsBuilder();
@@ -52,4 +58,13 @@ class DataConsistencyCheckUtilsTest {
assertTrue(DataConsistencyCheckUtils.isMatched(equalsBuilder, new Timestamp(time), new Timestamp(time / 10L * 10L + 1L)));
assertFalse(DataConsistencyCheckUtils.isMatched(equalsBuilder, new Timestamp(time), new Timestamp(time + 1000L)));
}
+
+ @Test
+ void assertIsFirstUniqueKeysValueMatched() {
+ Map<String, Object> record1 = ImmutableMap.of("order_id", 101, "user_id", 1, "status", "ok");
+ Map<String, Object> record2 = ImmutableMap.of("order_id", 102, "user_id", 2, "status", "ok");
+ EqualsBuilder equalsBuilder = new EqualsBuilder();
+ assertTrue(DataConsistencyCheckUtils.isFirstUniqueKeyValueMatched(record1, record1, UNIQUE_KEYS_NAMES.get(0), equalsBuilder));
+ assertFalse(DataConsistencyCheckUtils.isFirstUniqueKeyValueMatched(record1, record2, UNIQUE_KEYS_NAMES.get(0), equalsBuilder));
+ }
}
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32SingleTableInventoryCalculatorTest.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32SingleTableInventoryCalculatorTest.java
index c23537e161d..f67acacc9fe 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32SingleTableInventoryCalculatorTest.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/CRC32SingleTableInventoryCalculatorTest.java
@@ -20,6 +20,7 @@ package org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calc
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.result.SingleTableInventoryCalculatedResult;
import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSource;
import org.apache.shardingsphere.data.pipeline.core.exception.data.PipelineTableDataConsistencyCheckLoadingFailedException;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.query.QueryType;
import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColumnMetaData;
import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
@@ -65,7 +66,7 @@ class CRC32SingleTableInventoryCalculatorTest {
void setUp() throws SQLException {
DatabaseType databaseType = TypedSPILoader.getService(DatabaseType.class, "FIXTURE");
List<PipelineColumnMetaData> uniqueKeys = Collections.singletonList(new PipelineColumnMetaData(1, "id", Types.INTEGER, "integer", false, true, true));
- parameter = new SingleTableInventoryCalculateParameter(pipelineDataSource, new QualifiedTable(null, "foo_tbl"), Arrays.asList("foo_col", "bar_col"), uniqueKeys, Collections.emptyMap());
+ parameter = new SingleTableInventoryCalculateParameter(pipelineDataSource, new QualifiedTable(null, "foo_tbl"), Arrays.asList("foo_col", "bar_col"), uniqueKeys, QueryType.RANGE_QUERY);
when(pipelineDataSource.getDatabaseType()).thenReturn(databaseType);
when(pipelineDataSource.getConnection()).thenReturn(connection);
}
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculatorTest.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculatorTest.java
index f0956e547a2..9ca0a5896b9 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculatorTest.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculatorTest.java
@@ -28,9 +28,12 @@ import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColum
import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.infra.util.close.QuietlyCloser;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -45,6 +48,7 @@ import java.util.Optional;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class RecordSingleTableInventoryCalculatorTest {
@@ -100,51 +104,215 @@ class RecordSingleTableInventoryCalculatorTest {
}
@Test
- void assertCalculateOfAllQueryFromBegin() {
+ void assertCalculateOfRangeQueryFromBeginWithOrderIdUniqueKey() {
RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(5);
- SingleTableInventoryCalculateParameter param = generateParameter(dataSource, 0);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildOrderIdUniqueKey(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(0, false, null));
Optional<SingleTableInventoryCalculatedResult> calculateResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
assertTrue(calculateResult.isPresent());
SingleTableInventoryCalculatedResult actual = calculateResult.get();
assertTrue(actual.getMaxUniqueKeyValue().isPresent());
- assertThat(actual.getMaxUniqueKeyValue().get(), is(4));
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(5));
}
@Test
- void assertCalculateOfAllQueryFromMiddle() {
+ void assertCalculateOfRangeQueryFromMiddleWithOrderIdUniqueKey() {
RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(5);
- SingleTableInventoryCalculateParameter param = generateParameter(dataSource, 5);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildOrderIdUniqueKey(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(5, false, null));
Optional<SingleTableInventoryCalculatedResult> calculateResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
assertTrue(calculateResult.isPresent());
SingleTableInventoryCalculatedResult actual = calculateResult.get();
assertTrue(actual.getMaxUniqueKeyValue().isPresent());
- assertThat(actual.getMaxUniqueKeyValue().get(), is(9));
- }
-
- private SingleTableInventoryCalculateParameter generateParameter(final PipelineDataSource dataSource, final Object dataCheckPosition) {
- List<PipelineColumnMetaData> uniqueKeys = Collections.singletonList(new PipelineColumnMetaData(1, "order_id", Types.INTEGER, "integer", false, true, true));
- return new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"), Collections.emptyList(), uniqueKeys, dataCheckPosition);
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(10));
}
@Test
- void assertCalculateOfRangeQuery() {
+ void assertCalculateOfRangeQueryWithMultiColumnUniqueKeys() {
RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000);
- SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(
- dataSource, new QualifiedTable(null, "t_order"), Collections.emptyList(), buildUniqueKeys(), QueryType.RANGE_QUERY);
- param.setQueryRange(new QueryRange(3, true, 7));
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
+ assertQueryRangeCalculatedResult(calculator, param, new QueryRange(3, true, 6), 8, 6);
+ assertQueryRangeCalculatedResult(calculator, param, new QueryRange(3, false, 6), 3, 6);
+ }
+
+ private void assertQueryRangeCalculatedResult(final RecordSingleTableInventoryCalculator calculator, final SingleTableInventoryCalculateParameter param, final QueryRange queryRange,
+ final int expectedRecordsCount, final int expectedMaxUniqueKeyValue) {
+ param.setQueryRange(queryRange);
Optional<SingleTableInventoryCalculatedResult> calculatedResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
assertTrue(calculatedResult.isPresent());
SingleTableInventoryCalculatedResult actual = calculatedResult.get();
- assertThat(actual.getRecordsCount(), is(8));
+ assertThat(actual.getRecordsCount(), is(expectedRecordsCount));
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(expectedMaxUniqueKeyValue));
+ }
+
+ @Test
+ void assertCalculateOfRangeQueryWithMultiColumnUniqueKeys2() throws SQLException {
+ try (Connection connection = dataSource.getConnection()) {
+ connection.createStatement().execute("CREATE TABLE test3 (user_id INT NOT NULL, order_id INT, status VARCHAR(12))");
+ connection.createStatement().execute(
+ "INSERT INTO test3 (user_id,order_id,status) VALUES (3,1,'ok'),(3,2,'ok'),(4,3,'ok'),(4,4,'ok'),(5,5,'ok'),(5,6,'ok'),(6,7,'ok'),(6,8,'ok'),(7,9,'ok')");
+ }
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "test3"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
+ assertQueryRangeCalculatedResult(calculator, param, new QueryRange(3, true, 4), 4, 4);
+ assertQueryRangeCalculatedResult(calculator, param, new QueryRange(5, true, 6), 4, 6);
+ assertQueryRangeCalculatedResult(calculator, param, new QueryRange(5, true, 7), 5, 7);
+ }
+
+ @Test
+ void assertCalculateOfReservedRangeQueryWithMultiColumnUniqueKeys() {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(3, true, 2));
+ Optional<SingleTableInventoryCalculatedResult> calculatedResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
+ assertFalse(calculatedResult.isPresent());
+ }
+
+ @Test
+ void assertCalculateOfRangeQueryOnEmptyTableWithSingleColumnUniqueKey() throws SQLException {
+ try (Connection connection = dataSource.getConnection()) {
+ connection.createStatement().execute("CREATE TABLE test1 (user_id INT NOT NULL, order_id INT, status VARCHAR(12), PRIMARY KEY (user_id))");
+ }
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "test1"),
+ Collections.emptyList(), buildUserIdUniqueKey(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(0, false, null));
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(5);
+ Optional<SingleTableInventoryCalculatedResult> calculateResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
+ assertFalse(calculateResult.isPresent());
+ }
+
+ @Test
+ void assertCalculateOfRangeQueryOnEmptyTableWithMultiColumnUniqueKeys() throws SQLException {
+ try (Connection connection = dataSource.getConnection()) {
+ connection.createStatement().execute("CREATE TABLE test2 (user_id INT NOT NULL, order_id INT, status VARCHAR(12), PRIMARY KEY (user_id, order_id))");
+ }
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "test2"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(null, false, null));
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(5);
+ Optional<SingleTableInventoryCalculatedResult> calculateResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
+ assertFalse(calculateResult.isPresent());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"100", "2", "3"})
+ void assertCalculateOfRangeQueryAllWithOrderIdUniqueKeyWith3x(final int streamingChunkCount) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3, streamingChunkCount);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildOrderIdUniqueKey(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(null, false, null));
+ Iterator<SingleTableInventoryCalculatedResult> resultIterator = calculator.calculate(param).iterator();
+ RecordSingleTableInventoryCalculatedResult actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(3));
+ assertRecord(actual.getRecords().get(0), 1, 1);
+ assertRecord(actual.getRecords().get(1), 2, 2);
+ assertRecord(actual.getRecords().get(2), 3, 3);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
+ actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(3));
+ assertRecord(actual.getRecords().get(0), 3, 4);
+ assertRecord(actual.getRecords().get(1), 3, 5);
+ assertRecord(actual.getRecords().get(2), 3, 6);
assertTrue(actual.getMaxUniqueKeyValue().isPresent());
assertThat(actual.getMaxUniqueKeyValue().get(), is(6));
+ actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(3));
+ assertRecord(actual.getRecords().get(0), 3, 7);
+ assertRecord(actual.getRecords().get(1), 4, 8);
+ assertRecord(actual.getRecords().get(2), 5, 9);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(9));
+ actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(1));
+ assertRecord(actual.getRecords().get(0), 6, 10);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(10));
+ assertFalse(resultIterator.hasNext());
+ QuietlyCloser.close(param.getCalculationContext());
}
@Test
- void assertCalculateOfRangeQueryAll() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3);
- SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource,
- new QualifiedTable(null, "t_order"), Collections.emptyList(), buildUniqueKeys(), QueryType.RANGE_QUERY);
+ void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith50x100() {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(50, 100);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(null, false, null));
+ Iterator<SingleTableInventoryCalculatedResult> resultIterator = calculator.calculate(param).iterator();
+ RecordSingleTableInventoryCalculatedResult actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(10));
+ assertRecord(actual.getRecords().get(0), 1, 1);
+ assertRecord(actual.getRecords().get(1), 2, 2);
+ assertRecord(actual.getRecords().get(2), 3, 3);
+ assertRecord(actual.getRecords().get(3), 3, 4);
+ assertRecord(actual.getRecords().get(4), 3, 5);
+ assertRecord(actual.getRecords().get(5), 3, 6);
+ assertRecord(actual.getRecords().get(6), 3, 7);
+ assertRecord(actual.getRecords().get(7), 4, 8);
+ assertRecord(actual.getRecords().get(8), 5, 9);
+ assertRecord(actual.getRecords().get(9), 6, 10);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(6));
+ assertFalse(resultIterator.hasNext());
+ QuietlyCloser.close(param.getCalculationContext());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"100", "2", "3"})
+ void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith3x(final int streamingChunkCount) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3, streamingChunkCount);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
+ param.setQueryRange(new QueryRange(null, false, null));
+ Iterator<SingleTableInventoryCalculatedResult> resultIterator = calculator.calculate(param).iterator();
+ RecordSingleTableInventoryCalculatedResult actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(3));
+ assertRecord(actual.getRecords().get(0), 1, 1);
+ assertRecord(actual.getRecords().get(1), 2, 2);
+ assertRecord(actual.getRecords().get(2), 3, 3);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
+ actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(3));
+ assertRecord(actual.getRecords().get(0), 3, 4);
+ assertRecord(actual.getRecords().get(1), 3, 5);
+ assertRecord(actual.getRecords().get(2), 3, 6);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
+ actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(3));
+ assertRecord(actual.getRecords().get(0), 3, 7);
+ assertRecord(actual.getRecords().get(1), 4, 8);
+ assertRecord(actual.getRecords().get(2), 5, 9);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(5));
+ actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(1));
+ assertRecord(actual.getRecords().get(0), 6, 10);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(6));
+ assertFalse(resultIterator.hasNext());
+ QuietlyCloser.close(param.getCalculationContext());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"100", "3", "4"})
+ void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith2x(final int streamingChunkCount) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(2, streamingChunkCount);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
param.setQueryRange(new QueryRange(null, false, null));
Iterator<SingleTableInventoryCalculatedResult> resultIterator = calculator.calculate(param).iterator();
RecordSingleTableInventoryCalculatedResult actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
@@ -154,25 +322,31 @@ class RecordSingleTableInventoryCalculatorTest {
assertTrue(actual.getMaxUniqueKeyValue().isPresent());
assertThat(actual.getMaxUniqueKeyValue().get(), is(2));
actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
- assertThat(actual.getRecordsCount(), is(5));
+ assertThat(actual.getRecordsCount(), is(2));
assertRecord(actual.getRecords().get(0), 3, 3);
assertRecord(actual.getRecords().get(1), 3, 4);
- assertRecord(actual.getRecords().get(2), 3, 5);
- assertRecord(actual.getRecords().get(3), 3, 6);
- assertRecord(actual.getRecords().get(4), 3, 7);
assertTrue(actual.getMaxUniqueKeyValue().isPresent());
assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
assertThat(actual.getRecordsCount(), is(2));
- assertRecord(actual.getRecords().get(0), 4, 8);
- assertRecord(actual.getRecords().get(1), 5, 9);
+ assertRecord(actual.getRecords().get(0), 3, 5);
+ assertRecord(actual.getRecords().get(1), 3, 6);
assertTrue(actual.getMaxUniqueKeyValue().isPresent());
- assertThat(actual.getMaxUniqueKeyValue().get(), is(5));
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
- assertThat(actual.getRecordsCount(), is(1));
- assertRecord(actual.getRecords().get(0), 6, 10);
+ assertThat(actual.getRecordsCount(), is(2));
+ assertRecord(actual.getRecords().get(0), 3, 7);
+ assertRecord(actual.getRecords().get(1), 4, 8);
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(4));
+ actual = (RecordSingleTableInventoryCalculatedResult) resultIterator.next();
+ assertThat(actual.getRecordsCount(), is(2));
+ assertRecord(actual.getRecords().get(0), 5, 9);
+ assertRecord(actual.getRecords().get(1), 6, 10);
assertTrue(actual.getMaxUniqueKeyValue().isPresent());
assertThat(actual.getMaxUniqueKeyValue().get(), is(6));
+ assertFalse(resultIterator.hasNext());
+ QuietlyCloser.close(param.getCalculationContext());
}
private void assertRecord(final Map<String, Object> recordMap, final int userId, final int orderId) {
@@ -183,10 +357,11 @@ class RecordSingleTableInventoryCalculatorTest {
@Test
void assertCalculateOfPointQuery() {
RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3);
- SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource,
- new QualifiedTable(null, "t_order"), Collections.emptyList(), buildUniqueKeys(), QueryType.POINT_QUERY);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.POINT_QUERY);
param.setUniqueKeysValues(Arrays.asList(3, 3));
Optional<SingleTableInventoryCalculatedResult> calculatedResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
assertTrue(calculatedResult.isPresent());
SingleTableInventoryCalculatedResult actual = calculatedResult.get();
assertThat(actual.getRecordsCount(), is(1));
@@ -194,7 +369,30 @@ class RecordSingleTableInventoryCalculatorTest {
assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
}
- private List<PipelineColumnMetaData> buildUniqueKeys() {
+ @Test
+ void assertCalculateOfPointRangeQuery() {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3);
+ SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
+ Collections.emptyList(), buildUserIdUniqueKey(), QueryType.POINT_QUERY);
+ param.setUniqueKeysValues(Collections.singleton(3));
+ Optional<SingleTableInventoryCalculatedResult> calculatedResult = calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
+ assertTrue(calculatedResult.isPresent());
+ SingleTableInventoryCalculatedResult actual = calculatedResult.get();
+ assertThat(actual.getRecordsCount(), is(5));
+ assertTrue(actual.getMaxUniqueKeyValue().isPresent());
+ assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
+ }
+
+ private List<PipelineColumnMetaData> buildUserIdUniqueKey() {
+ return Collections.singletonList(new PipelineColumnMetaData(1, "user_id", Types.INTEGER, "integer", false, true, true));
+ }
+
+ private List<PipelineColumnMetaData> buildOrderIdUniqueKey() {
+ return Collections.singletonList(new PipelineColumnMetaData(1, "order_id", Types.INTEGER, "integer", false, true, true));
+ }
+
+ private List<PipelineColumnMetaData> buildMultiColumnUniqueKeys() {
return Arrays.asList(
new PipelineColumnMetaData(1, "user_id", Types.INTEGER, "integer", false, true, true),
new PipelineColumnMetaData(2, "order_id", Types.INTEGER, "integer", false, true, true));