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/25 06:46:16 UTC
(shardingsphere) branch master updated: Add new streaming query impl in DATA_MATCH consistency check for sharding table on MySQL (#35786)
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 79c4207ae7c Add new streaming query impl in DATA_MATCH consistency check for sharding table on MySQL (#35786)
79c4207ae7c is described below
commit 79c4207ae7c8c15a2db285cd71f14a31dd1c35ce
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Wed Jun 25 14:46:10 2025 +0800
Add new streaming query impl in DATA_MATCH consistency check for sharding table on MySQL (#35786)
* Add streaming-range-type prop for DATA_MATCH consistency check
* Add more test
* Update migration E2E for DATA_MATCH 'streaming-range-type' prop
* Update code style
---
.../DataMatchTableDataConsistencyChecker.java | 31 +++++++-
.../RecordSingleTableInventoryCalculator.java | 41 ++++++++--
.../table/calculator/StreamingRangeType.java | 26 ++++++
...PipelineDataConsistencyCalculateSQLBuilder.java | 6 +-
.../DataMatchTableDataConsistencyCheckerTest.java | 34 ++++++--
.../fixture/FixturePipelineSQLBuilder.java | 2 +-
...lineDataConsistencyCalculateSQLBuilderTest.java | 31 ++++++--
.../sql/PipelineInventoryDumpSQLBuilderTest.java | 8 +-
.../cases/migration/AbstractMigrationE2EIT.java | 11 +++
.../general/MySQLMigrationGeneralE2EIT.java | 7 +-
.../RecordSingleTableInventoryCalculatorTest.java | 92 ++++++++++++----------
11 files changed, 217 insertions(+), 72 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyChecker.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyChecker.java
index 59243852019..04d707102db 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyChecker.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyChecker.java
@@ -22,11 +22,13 @@ import org.apache.shardingsphere.data.pipeline.core.consistencycheck.result.Tabl
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.result.TableDataConsistencyCheckResult;
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator.RecordSingleTableInventoryCalculator;
import org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator.SingleTableInventoryCalculator;
+import org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator.StreamingRangeType;
import org.apache.shardingsphere.data.pipeline.core.exception.param.PipelineInvalidParameterException;
import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import org.apache.shardingsphere.infra.spi.ShardingSphereServiceLoader;
import org.apache.shardingsphere.infra.spi.annotation.SPIDescription;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Optional;
import java.util.Properties;
@@ -39,13 +41,20 @@ public final class DataMatchTableDataConsistencyChecker implements TableDataCons
private static final String CHUNK_SIZE_KEY = "chunk-size";
+ private static final String STREAMING_RANGE_TYPE_KEY = "streaming-range-type";
+
private static final int DEFAULT_CHUNK_SIZE = 1000;
+ private static final StreamingRangeType DEFAULT_STREAMING_RANGE_TYPE = StreamingRangeType.SMALL;
+
private int chunkSize;
+ private StreamingRangeType streamingRangeType;
+
@Override
public void init(final Properties props) {
chunkSize = getChunkSize(props);
+ streamingRangeType = getStreamingRangeType(props);
}
private int getChunkSize(final Properties props) {
@@ -65,9 +74,22 @@ public final class DataMatchTableDataConsistencyChecker implements TableDataCons
return result;
}
+ private StreamingRangeType getStreamingRangeType(final Properties props) {
+ String streamingRangeTypeText = props.getProperty(STREAMING_RANGE_TYPE_KEY);
+ if (Strings.isNullOrEmpty(streamingRangeTypeText)) {
+ return DEFAULT_STREAMING_RANGE_TYPE;
+ }
+ try {
+ return StreamingRangeType.valueOf(streamingRangeTypeText.toUpperCase());
+ } catch (final IllegalArgumentException ex) {
+ throw new PipelineInvalidParameterException("'streaming-range-type' is not a valid value: `" + streamingRangeTypeText
+ + "`, expected values are " + Arrays.toString(StreamingRangeType.values()));
+ }
+ }
+
@Override
public TableInventoryChecker buildTableInventoryChecker(final TableInventoryCheckParameter param) {
- return new DataMatchTableInventoryChecker(param, chunkSize);
+ return new DataMatchTableInventoryChecker(param, chunkSize, streamingRangeType);
}
@Override
@@ -88,9 +110,12 @@ public final class DataMatchTableDataConsistencyChecker implements TableDataCons
private final int chunkSize;
- DataMatchTableInventoryChecker(final TableInventoryCheckParameter param, final int chunkSize) {
+ private final StreamingRangeType streamingRangeType;
+
+ DataMatchTableInventoryChecker(final TableInventoryCheckParameter param, final int chunkSize, final StreamingRangeType streamingRangeType) {
super(param);
this.chunkSize = chunkSize;
+ this.streamingRangeType = streamingRangeType;
}
@Override
@@ -103,7 +128,7 @@ public final class DataMatchTableDataConsistencyChecker implements TableDataCons
@Override
protected SingleTableInventoryCalculator buildSingleTableInventoryCalculator() {
- return new RecordSingleTableInventoryCalculator(chunkSize);
+ return new RecordSingleTableInventoryCalculator(chunkSize, streamingRangeType);
}
}
}
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 338fd6abfbd..f1317092003 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
@@ -55,15 +55,18 @@ import java.util.Optional;
@RequiredArgsConstructor
public final class RecordSingleTableInventoryCalculator extends AbstractStreamingSingleTableInventoryCalculator {
+ private static final int DEFAULT_STREAMING_CHUNK_COUNT = 100;
+
private final int chunkSize;
private final int streamingChunkCount;
+ private final StreamingRangeType streamingRangeType;
+
private final EqualsBuilder equalsBuilder = new EqualsBuilder();
- public RecordSingleTableInventoryCalculator(final int chunkSize) {
- this.chunkSize = chunkSize;
- streamingChunkCount = 100;
+ public RecordSingleTableInventoryCalculator(final int chunkSize, final StreamingRangeType streamingRangeType) {
+ this(chunkSize, DEFAULT_STREAMING_CHUNK_COUNT, streamingRangeType);
}
@Override
@@ -85,6 +88,9 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
if (QueryType.POINT_QUERY == param.getQueryType()) {
return pointQuery(param, columnValueReaderEngine);
}
+ if (StreamingRangeType.LARGE == streamingRangeType) {
+ return allQuery(param, columnValueReaderEngine);
+ }
if (param.getUniqueKeys().size() <= 1) {
return rangeQueryWithSingleColumUniqueKey(param, columnValueReaderEngine, 1);
}
@@ -112,6 +118,26 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
return result;
}
+ private List<Map<String, Object>> allQuery(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()));
+ result.add(readRecord(columnValueReaderEngine, resultSet, resultSetMetaData));
+ if (result.size() == chunkSize) {
+ break;
+ }
+ }
+ if (result.isEmpty()) {
+ QuietlyCloser.close(calculationContext);
+ }
+ 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<>();
@@ -231,13 +257,14 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
}
private String getQuerySQL(final SingleTableInventoryCalculateParameter param) {
- ShardingSpherePreconditions.checkState(param.getUniqueKeys() != null && !param.getUniqueKeys().isEmpty() && null != param.getFirstUniqueKey(),
+ ShardingSpherePreconditions.checkState(null != param.getUniqueKeys() && !param.getUniqueKeys().isEmpty() && null != param.getFirstUniqueKey(),
() -> new UnsupportedOperationException("Record inventory calculator does not support table without unique key and primary key now."));
PipelineDataConsistencyCalculateSQLBuilder pipelineSQLBuilder = new PipelineDataConsistencyCalculateSQLBuilder(param.getDatabaseType());
Collection<String> columnNames = param.getColumnNames().isEmpty() ? Collections.singleton("*") : param.getColumnNames();
switch (param.getQueryType()) {
case RANGE_QUERY:
- return pipelineSQLBuilder.buildQueryRangeOrderingSQL(param.getTable(), columnNames, param.getUniqueKeysNames(), param.getQueryRange(), param.getShardingColumnsNames());
+ return pipelineSQLBuilder.buildQueryRangeOrderingSQL(param.getTable(), columnNames, param.getUniqueKeysNames(), param.getQueryRange(),
+ StreamingRangeType.SMALL == streamingRangeType, param.getShardingColumnsNames());
case POINT_QUERY:
return pipelineSQLBuilder.buildPointQuerySQL(param.getTable(), columnNames, param.getUniqueKeysNames(), param.getShardingColumnsNames());
default:
@@ -258,7 +285,9 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
if (null != queryRange.getUpper()) {
preparedStatement.setObject(parameterIndex++, queryRange.getUpper());
}
- preparedStatement.setObject(parameterIndex, chunkSize * streamingChunkCount);
+ if (StreamingRangeType.SMALL == streamingRangeType) {
+ preparedStatement.setObject(parameterIndex, chunkSize * streamingChunkCount);
+ }
} else if (queryType == QueryType.POINT_QUERY) {
Collection<Object> uniqueKeysValues = param.getUniqueKeysValues();
ShardingSpherePreconditions.checkNotNull(uniqueKeysValues,
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/StreamingRangeType.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/StreamingRangeType.java
new file mode 100644
index 00000000000..ad0503daa61
--- /dev/null
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/StreamingRangeType.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.core.consistencycheck.table.calculator;
+
+/**
+ * Streaming range type.
+ */
+public enum StreamingRangeType {
+
+ SMALL, LARGE
+}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java
index 36ef5cb96f7..bb1a90bc9e4 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilder.java
@@ -51,12 +51,14 @@ public final class PipelineDataConsistencyCalculateSQLBuilder {
* @param columnNames column names
* @param uniqueKeys unique keys, it may be primary key, not null
* @param queryRange query range
+ * @param pageQuery whether it is page query
* @param shardingColumnsNames sharding columns names
* @return built SQL
*/
public String buildQueryRangeOrderingSQL(final QualifiedTable qualifiedTable, final Collection<String> columnNames, final List<String> uniqueKeys, final QueryRange queryRange,
- final List<String> shardingColumnsNames) {
- return dialectSQLBuilder.wrapWithPageQuery(buildQueryRangeOrderingSQL0(qualifiedTable, columnNames, uniqueKeys, queryRange, shardingColumnsNames));
+ final boolean pageQuery, final List<String> shardingColumnsNames) {
+ String result = buildQueryRangeOrderingSQL0(qualifiedTable, columnNames, uniqueKeys, queryRange, shardingColumnsNames);
+ return pageQuery ? dialectSQLBuilder.wrapWithPageQuery(result) : result;
}
private String buildQueryRangeOrderingSQL0(final QualifiedTable qualifiedTable, final Collection<String> columnNames, final List<String> uniqueKeys, final QueryRange queryRange,
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyCheckerTest.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyCheckerTest.java
index a895895a4b4..7857bfdcd17 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyCheckerTest.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/DataMatchTableDataConsistencyCheckerTest.java
@@ -33,26 +33,48 @@ class DataMatchTableDataConsistencyCheckerTest {
@SneakyThrows(ReflectiveOperationException.class)
@Test
- void assertInitSuccess() {
+ void assertChunkSizeInitSuccess() {
for (String each : Arrays.asList("1", "1000")) {
DataMatchTableDataConsistencyChecker checker = new DataMatchTableDataConsistencyChecker();
- checker.init(buildAlgorithmProperties(each));
+ checker.init(buildChunkSizeProperties(each));
String actual = Plugins.getMemberAccessor().get(DataMatchTableDataConsistencyChecker.class.getDeclaredField("chunkSize"), checker).toString();
assertThat(actual, is(each));
}
}
@Test
- void assertInitFailure() {
- assertThrows(PipelineInvalidParameterException.class, () -> new DataMatchTableDataConsistencyChecker().init(buildAlgorithmProperties("xyz")));
+ void assertChunkSizeInitFailure() {
+ assertThrows(PipelineInvalidParameterException.class, () -> new DataMatchTableDataConsistencyChecker().init(buildChunkSizeProperties("xyz")));
for (String each : Arrays.asList("0", "-1")) {
- assertThrows(PipelineInvalidParameterException.class, () -> new DataMatchTableDataConsistencyChecker().init(buildAlgorithmProperties(each)));
+ assertThrows(PipelineInvalidParameterException.class, () -> new DataMatchTableDataConsistencyChecker().init(buildChunkSizeProperties(each)));
}
}
- private Properties buildAlgorithmProperties(final String chunkSize) {
+ @SneakyThrows(ReflectiveOperationException.class)
+ @Test
+ void assertStreamingRangeTypeInitSuccess() {
+ for (String each : Arrays.asList("small", "large", "SMALL", "LARGE")) {
+ DataMatchTableDataConsistencyChecker checker = new DataMatchTableDataConsistencyChecker();
+ checker.init(buildStreamingRangeTypeProperties(each));
+ String actual = Plugins.getMemberAccessor().get(DataMatchTableDataConsistencyChecker.class.getDeclaredField("streamingRangeType"), checker).toString();
+ assertThat(actual, is(each.toUpperCase()));
+ }
+ }
+
+ @Test
+ void assertStreamingRangeTypeInitFailure() {
+ assertThrows(PipelineInvalidParameterException.class, () -> new DataMatchTableDataConsistencyChecker().init(buildStreamingRangeTypeProperties("xyz")));
+ }
+
+ private Properties buildChunkSizeProperties(final String chunkSize) {
Properties result = new Properties();
result.put("chunk-size", chunkSize);
return result;
}
+
+ private Properties buildStreamingRangeTypeProperties(final String streamingRangeType) {
+ Properties result = new Properties();
+ result.put("streaming-range-type", streamingRangeType);
+ return result;
+ }
}
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/fixture/FixturePipelineSQLBuilder.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/fixture/FixturePipelineSQLBuilder.java
index e831ef9f4ae..5dcc1929350 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/fixture/FixturePipelineSQLBuilder.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/fixture/FixturePipelineSQLBuilder.java
@@ -43,7 +43,7 @@ public final class FixturePipelineSQLBuilder implements DialectPipelineSQLBuilde
@Override
public String wrapWithPageQuery(final String sql) {
- return sql;
+ return sql + " LIMIT ?";
}
@Override
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java
index 0644ad2c4ef..ba07df46bdd 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineDataConsistencyCalculateSQLBuilderTest.java
@@ -43,21 +43,40 @@ class PipelineDataConsistencyCalculateSQLBuilderTest {
private final PipelineDataConsistencyCalculateSQLBuilder sqlBuilder = new PipelineDataConsistencyCalculateSQLBuilder(TypedSPILoader.getService(DatabaseType.class, "FIXTURE"));
@Test
- void assertBuildQueryRangeOrderingSQLWithoutQueryCondition() {
+ void assertBuildQueryRangeOrderingSQLPageQuery() {
String actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
- new QueryRange(1, true, 5), SHARDING_COLUMNS_NAMES);
+ new QueryRange(1, true, 5), true, SHARDING_COLUMNS_NAMES);
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>=? AND order_id<=? ORDER BY order_id ASC, status ASC, user_id ASC LIMIT ?"));
+ actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
+ new QueryRange(1, false, 5), true, SHARDING_COLUMNS_NAMES);
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? AND order_id<=? ORDER BY order_id ASC, status ASC, user_id ASC LIMIT ?"));
+ actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
+ new QueryRange(1, false, null), true, SHARDING_COLUMNS_NAMES);
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? ORDER BY order_id ASC, status ASC, user_id ASC LIMIT ?"));
+ actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
+ new QueryRange(null, false, 5), true, SHARDING_COLUMNS_NAMES);
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id<=? ORDER BY order_id ASC, status ASC, user_id ASC LIMIT ?"));
+ actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
+ new QueryRange(null, false, null), true, SHARDING_COLUMNS_NAMES);
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order ORDER BY order_id ASC, status ASC, user_id ASC LIMIT ?"));
+ }
+
+ @Test
+ void assertBuildQueryRangeOrderingSQLNotPageQuery() {
+ String actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
+ new QueryRange(1, true, 5), false, SHARDING_COLUMNS_NAMES);
assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>=? AND order_id<=? ORDER BY order_id ASC, status ASC, user_id ASC"));
actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
- new QueryRange(1, false, 5), SHARDING_COLUMNS_NAMES);
+ new QueryRange(1, false, 5), false, SHARDING_COLUMNS_NAMES);
assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? AND order_id<=? ORDER BY order_id ASC, status ASC, user_id ASC"));
actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
- new QueryRange(1, false, null), SHARDING_COLUMNS_NAMES);
+ new QueryRange(1, false, null), false, SHARDING_COLUMNS_NAMES);
assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? ORDER BY order_id ASC, status ASC, user_id ASC"));
actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
- new QueryRange(null, false, 5), SHARDING_COLUMNS_NAMES);
+ new QueryRange(null, false, 5), false, SHARDING_COLUMNS_NAMES);
assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id<=? ORDER BY order_id ASC, status ASC, user_id ASC"));
actual = sqlBuilder.buildQueryRangeOrderingSQL(new QualifiedTable(null, "t_order"), COLUMN_NAMES, UNIQUE_KEYS,
- new QueryRange(null, false, null), SHARDING_COLUMNS_NAMES);
+ new QueryRange(null, false, null), false, SHARDING_COLUMNS_NAMES);
assertThat(actual, is("SELECT order_id,user_id,status FROM t_order ORDER BY order_id ASC, status ASC, user_id ASC"));
}
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineInventoryDumpSQLBuilderTest.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineInventoryDumpSQLBuilderTest.java
index 839a51664e6..8921dda36fd 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineInventoryDumpSQLBuilderTest.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/sql/PipelineInventoryDumpSQLBuilderTest.java
@@ -34,17 +34,17 @@ class PipelineInventoryDumpSQLBuilderTest {
@Test
void assertBuildDivisibleSQL() {
String actual = sqlBuilder.buildDivisibleSQL(new BuildDivisibleSQLParameter(null, "t_order", Arrays.asList("order_id", "user_id", "status"), "order_id", true, true));
- assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>=? AND order_id<=? ORDER BY order_id ASC"));
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>=? AND order_id<=? ORDER BY order_id ASC LIMIT ?"));
actual = sqlBuilder.buildDivisibleSQL(new BuildDivisibleSQLParameter(null, "t_order", Arrays.asList("order_id", "user_id", "status"), "order_id", false, true));
- assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? AND order_id<=? ORDER BY order_id ASC"));
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? AND order_id<=? ORDER BY order_id ASC LIMIT ?"));
}
@Test
void assertBuildUnlimitedDivisibleSQL() {
String actual = sqlBuilder.buildDivisibleSQL(new BuildDivisibleSQLParameter(null, "t_order", Arrays.asList("order_id", "user_id", "status"), "order_id", true, false));
- assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>=? ORDER BY order_id ASC"));
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>=? ORDER BY order_id ASC LIMIT ?"));
actual = sqlBuilder.buildDivisibleSQL(new BuildDivisibleSQLParameter(null, "t_order", Arrays.asList("order_id", "user_id", "status"), "order_id", false, false));
- assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? ORDER BY order_id ASC"));
+ assertThat(actual, is("SELECT order_id,user_id,status FROM t_order WHERE order_id>? ORDER BY order_id ASC LIMIT ?"));
}
@Test
diff --git a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/AbstractMigrationE2EIT.java b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/AbstractMigrationE2EIT.java
index b4a30eb6081..1904fb5f4b3 100644
--- a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/AbstractMigrationE2EIT.java
+++ b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/AbstractMigrationE2EIT.java
@@ -189,4 +189,15 @@ public abstract class AbstractMigrationE2EIT {
+ "))";
return String.format(sql, jobId, algorithmType);
}
+
+ protected Properties convertToProperties(final Map<String, String> map) {
+ Properties result = new Properties();
+ if (null == map || map.isEmpty()) {
+ return result;
+ }
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ result.setProperty(entry.getKey(), entry.getValue());
+ }
+ return result;
+ }
}
diff --git a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java
index 99c2acc6ede..0f4dcdecbb6 100644
--- a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java
+++ b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java
@@ -37,6 +37,7 @@ import org.awaitility.Awaitility;
import org.junit.jupiter.api.condition.EnabledIf;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
import javax.sql.DataSource;
import java.sql.SQLException;
@@ -90,11 +91,9 @@ class MySQLMigrationGeneralE2EIT extends AbstractMigrationE2EIT {
DataSource jdbcDataSource = containerComposer.generateShardingSphereDataSourceFromProxy();
containerComposer.assertOrderRecordExist(jdbcDataSource, "t_order", 10000);
containerComposer.assertOrderRecordExist(jdbcDataSource, "t_order_item", 10000);
- Properties algorithmProps = new Properties();
- algorithmProps.setProperty("chunk-size", "300");
- assertMigrationSuccessById(containerComposer, orderJobId, "DATA_MATCH", algorithmProps);
+ assertMigrationSuccessById(containerComposer, orderJobId, "DATA_MATCH", convertToProperties(ImmutableMap.of("chunk-size", "300", "streaming-range-type", "SMALL")));
String orderItemJobId = getJobIdByTableName(containerComposer, "ds_0.t_order_item");
- assertMigrationSuccessById(containerComposer, orderItemJobId, "DATA_MATCH", algorithmProps);
+ assertMigrationSuccessById(containerComposer, orderItemJobId, "DATA_MATCH", convertToProperties(ImmutableMap.of("chunk-size", "300", "streaming-range-type", "LARGE")));
Awaitility.await().pollDelay(2L, TimeUnit.SECONDS).until(() -> true);
assertMigrationSuccessById(containerComposer, orderItemJobId, "CRC32_MATCH", new Properties());
for (String each : listJobId(containerComposer)) {
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 5943500ca3a..6ed7368a137 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
@@ -31,7 +31,6 @@ 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;
@@ -103,25 +102,28 @@ class RecordSingleTableInventoryCalculatorTest {
preparedStatement.executeUpdate();
}
- @Test
- void assertCalculateOfRangeQueryFromBeginWithOrderIdUniqueKey() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(4);
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfRangeQueryFromBeginWithOrderIdUniqueKey(final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(4, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildOrderIdUniqueKey(), QueryType.RANGE_QUERY);
assertQueryRangeCalculatedResult(calculator, param, new QueryRange(0, false, null), 4, 4);
}
- @Test
- void assertCalculateOfRangeQueryFromMiddleWithOrderIdUniqueKey() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(4);
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfRangeQueryFromMiddleWithOrderIdUniqueKey(final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(4, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildOrderIdUniqueKey(), QueryType.RANGE_QUERY);
assertQueryRangeCalculatedResult(calculator, param, new QueryRange(4, false, null), 4, 8);
}
- @Test
- void assertCalculateOfRangeQueryWithMultiColumnUniqueKeys() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000);
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfRangeQueryWithMultiColumnUniqueKeys(final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000, StreamingRangeType.valueOf(streamingRangeType));
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);
@@ -140,14 +142,16 @@ class RecordSingleTableInventoryCalculatorTest {
assertThat(actual.getMaxUniqueKeyValue().get(), is(expectedMaxUniqueKeyValue));
}
- @Test
- void assertCalculateOfRangeQueryWithMultiColumnUniqueKeys2() throws SQLException {
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfRangeQueryWithMultiColumnUniqueKeys2(final String streamingRangeType) throws SQLException {
try (Connection connection = dataSource.getConnection()) {
+ connection.createStatement().execute("DROP TABLE IF EXISTS test3");
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);
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000, StreamingRangeType.valueOf(streamingRangeType));
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);
@@ -155,9 +159,10 @@ class RecordSingleTableInventoryCalculatorTest {
assertQueryRangeCalculatedResult(calculator, param, new QueryRange(5, true, 7), 5, 7);
}
- @Test
- void assertCalculateOfReservedRangeQueryWithMultiColumnUniqueKeys() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000);
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfReservedRangeQueryWithMultiColumnUniqueKeys(final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(1000, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
param.setQueryRange(new QueryRange(3, true, 2));
@@ -166,38 +171,42 @@ class RecordSingleTableInventoryCalculatorTest {
assertFalse(calculatedResult.isPresent());
}
- @Test
- void assertCalculateOfRangeQueryOnEmptyTableWithSingleColumnUniqueKey() throws SQLException {
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfRangeQueryOnEmptyTableWithSingleColumnUniqueKey(final String streamingRangeType) throws SQLException {
try (Connection connection = dataSource.getConnection()) {
+ connection.createStatement().execute("DROP TABLE IF EXISTS test1");
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);
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(5, StreamingRangeType.valueOf(streamingRangeType));
Optional<SingleTableInventoryCalculatedResult> calculateResult = calculator.calculateChunk(param);
QuietlyCloser.close(param.getCalculationContext());
assertFalse(calculateResult.isPresent());
}
- @Test
- void assertCalculateOfRangeQueryOnEmptyTableWithMultiColumnUniqueKeys() throws SQLException {
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfRangeQueryOnEmptyTableWithMultiColumnUniqueKeys(final String streamingRangeType) throws SQLException {
try (Connection connection = dataSource.getConnection()) {
+ connection.createStatement().execute("DROP TABLE IF EXISTS test2");
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);
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(5, StreamingRangeType.valueOf(streamingRangeType));
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);
+ @CsvSource({"100,SMALL", "2,SMALL", "3,SMALL", "100,LARGE", "2,LARGE", "3,LARGE"})
+ void assertCalculateOfRangeQueryAllWithOrderIdUniqueKeyWith3x(final int streamingChunkCount, final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3, streamingChunkCount, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildOrderIdUniqueKey(), QueryType.RANGE_QUERY);
param.setQueryRange(new QueryRange(null, false, null));
@@ -232,9 +241,10 @@ class RecordSingleTableInventoryCalculatorTest {
QuietlyCloser.close(param.getCalculationContext());
}
- @Test
- void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith50x100() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(50, 100);
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith50x100(final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(50, 100, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
param.setQueryRange(new QueryRange(null, false, null));
@@ -258,9 +268,9 @@ class RecordSingleTableInventoryCalculatorTest {
}
@ParameterizedTest
- @CsvSource({"100", "2", "3"})
- void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith3x(final int streamingChunkCount) {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3, streamingChunkCount);
+ @CsvSource({"100,SMALL", "2,SMALL", "3,SMALL", "100,LARGE", "2,LARGE", "3,LARGE"})
+ void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith3x(final int streamingChunkCount, final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3, streamingChunkCount, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
param.setQueryRange(new QueryRange(null, false, null));
@@ -296,9 +306,9 @@ class RecordSingleTableInventoryCalculatorTest {
}
@ParameterizedTest
- @CsvSource({"100", "3", "4"})
- void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith2x(final int streamingChunkCount) {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(2, streamingChunkCount);
+ @CsvSource({"100,SMALL", "3,SMALL", "4,SMALL", "100,LARGE", "3,LARGE", "4,LARGE"})
+ void assertCalculateOfRangeQueryAllWithMultiColumnUniqueKeysWith2x(final int streamingChunkCount, final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(2, streamingChunkCount, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.RANGE_QUERY);
param.setQueryRange(new QueryRange(null, false, null));
@@ -342,9 +352,10 @@ class RecordSingleTableInventoryCalculatorTest {
assertThat(recordMap.get("order_id"), is(orderId));
}
- @Test
- void assertCalculateOfPointQuery() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3);
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfPointQuery(final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildMultiColumnUniqueKeys(), QueryType.POINT_QUERY);
param.setUniqueKeysValues(Arrays.asList(3, 3));
@@ -357,9 +368,10 @@ class RecordSingleTableInventoryCalculatorTest {
assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
}
- @Test
- void assertCalculateOfPointRangeQuery() {
- RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3);
+ @ParameterizedTest
+ @CsvSource({"SMALL", "LARGE"})
+ void assertCalculateOfPointRangeQuery(final String streamingRangeType) {
+ RecordSingleTableInventoryCalculator calculator = new RecordSingleTableInventoryCalculator(3, StreamingRangeType.valueOf(streamingRangeType));
SingleTableInventoryCalculateParameter param = new SingleTableInventoryCalculateParameter(dataSource, new QualifiedTable(null, "t_order"),
Collections.emptyList(), buildUserIdUniqueKey(), QueryType.POINT_QUERY);
param.setUniqueKeysValues(Collections.singleton(3));