You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by pa...@apache.org on 2022/05/11 02:44:46 UTC
[shardingsphere] branch master updated: Support string type primary key table scaling (#17511)
This is an automated email from the ASF dual-hosted git repository.
panjuan 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 f9b04bad274 Support string type primary key table scaling (#17511)
f9b04bad274 is described below
commit f9b04bad274c24f90010b26ab45f791748ec9050
Author: Hongsheng Zhong <sa...@126.com>
AuthorDate: Wed May 11 10:44:39 2022 +0800
Support string type primary key table scaling (#17511)
* InventoryTaskSplitter support string column type
* Rename PrimaryKeyPosition to IntegerPrimaryKeyPosition
* Support string type primary key
* Refactor PrimaryKeyPosition
* Inventory dumper support string primary key
* DATA_MATCH consistency calculate alg support string primary key
* Inventory dumper support string primary key
* Workaround for MogDB inventory dump
* Workaround for MogDB data consistency check
* FinishedCheckJob ignore ongoing check
* Unit test
---
.../ingest/InventoryDumperConfiguration.java | 2 +
...osition.java => IntegerPrimaryKeyPosition.java} | 50 ++++++++--------
.../api/ingest/position/PrimaryKeyPosition.java | 44 ++++++--------
.../ingest/position/PrimaryKeyPositionFactory.java | 67 ++++++++++++++++++++++
...Position.java => StringPrimaryKeyPosition.java} | 36 +++++-------
.../spi/sqlbuilder/PipelineSQLBuilder.java | 10 ++--
...DataMatchDataConsistencyCalculateAlgorithm.java | 61 ++++++++++++--------
.../ingest/dumper/AbstractInventoryDumper.java | 62 ++++++++++++--------
.../data/pipeline/core/job/FinishedCheckJob.java | 11 ++++
.../job/progress/yaml/JobProgressYamlSwapper.java | 4 +-
.../sqlbuilder/AbstractPipelineSQLBuilder.java | 26 ++++++---
.../data/pipeline/core/util/PipelineJdbcUtils.java | 60 +++++++++++++++++++
.../rulealtered/prepare/InventoryTaskSplitter.java | 60 ++++++++++---------
.../fixture/FixturePipelineSQLBuilder.java | 4 +-
.../api/impl/GovernanceRepositoryAPIImplTest.java | 2 +
...est.java => IntegerPrimaryKeyPositionTest.java} | 13 +++--
...Test.java => StringPrimaryKeyPositionTest.java} | 17 +++---
.../core/job/progress/JobProgressTest.java | 4 +-
.../progress/yaml/JobProgressYamlSwapperTest.java | 2 +-
.../data/pipeline/core/task/InventoryTaskTest.java | 10 +++-
.../prepare/InventoryTaskSplitterTest.java | 19 ++++--
.../config_sharding_sphere_jdbc_source.yaml | 2 +-
.../test/resources/job-progress-no-finished.yaml | 4 +-
.../test/resources/job-progress-no-inventory.yaml | 6 --
.../src/test/resources/job-progress.yaml | 2 +-
25 files changed, 380 insertions(+), 198 deletions(-)
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java
index 9b4ba152b57..1d5635aec6c 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/config/ingest/InventoryDumperConfiguration.java
@@ -38,6 +38,8 @@ public final class InventoryDumperConfiguration extends DumperConfiguration {
// TODO rename to uniqueKey
private String primaryKey;
+ private Integer uniqueKeyDataType;
+
private Integer shardingItem;
private int batchSize = 1000;
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/IntegerPrimaryKeyPosition.java
similarity index 58%
copy from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java
copy to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/IntegerPrimaryKeyPosition.java
index 11203fcb9ed..cb38fa617be 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/IntegerPrimaryKeyPosition.java
@@ -17,43 +17,45 @@
package org.apache.shardingsphere.data.pipeline.api.ingest.position;
-import com.google.common.base.Preconditions;
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-
/**
- * Use primary key as position.
+ * Integer primary key position.
*/
-@RequiredArgsConstructor
-@Getter
-public final class PrimaryKeyPosition implements IngestPosition<PrimaryKeyPosition> {
+public final class IntegerPrimaryKeyPosition extends PrimaryKeyPosition<Long> implements IngestPosition<IntegerPrimaryKeyPosition> {
private final long beginValue;
private final long endValue;
- /**
- * Init by string data.
- *
- * @param data string data
- * @return primary key position
- */
- public static PrimaryKeyPosition init(final String data) {
- String[] array = data.split(",");
- Preconditions.checkArgument(array.length == 2, "Unknown primary key position: " + data);
- return new PrimaryKeyPosition(Long.parseLong(array[0]), Long.parseLong(array[1]));
+ public IntegerPrimaryKeyPosition(final long beginValue, final long endValue) {
+ this.beginValue = beginValue;
+ this.endValue = endValue;
+ }
+
+ @Override
+ public Long getBeginValue() {
+ return beginValue;
+ }
+
+ @Override
+ public Long getEndValue() {
+ return endValue;
+ }
+
+ @Override
+ protected Long convert(final String value) {
+ return Long.parseLong(value);
+ }
+
+ @Override
+ protected char getType() {
+ return 'i';
}
@Override
- public int compareTo(final PrimaryKeyPosition position) {
+ public int compareTo(final IntegerPrimaryKeyPosition position) {
if (null == position) {
return 1;
}
return Long.compare(beginValue, position.beginValue);
}
-
- @Override
- public String toString() {
- return String.format("%d,%d", beginValue, endValue);
- }
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java
index 11203fcb9ed..6f79125bc67 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java
@@ -17,43 +17,31 @@
package org.apache.shardingsphere.data.pipeline.api.ingest.position;
-import com.google.common.base.Preconditions;
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-
/**
- * Use primary key as position.
+ * Primary key position.
*/
-@RequiredArgsConstructor
-@Getter
-public final class PrimaryKeyPosition implements IngestPosition<PrimaryKeyPosition> {
-
- private final long beginValue;
+public abstract class PrimaryKeyPosition<T> {
- private final long endValue;
+ /**
+ * Get begin value.
+ *
+ * @return begin value
+ */
+ public abstract T getBeginValue();
/**
- * Init by string data.
+ * Get end value.
*
- * @param data string data
- * @return primary key position
+ * @return end value
*/
- public static PrimaryKeyPosition init(final String data) {
- String[] array = data.split(",");
- Preconditions.checkArgument(array.length == 2, "Unknown primary key position: " + data);
- return new PrimaryKeyPosition(Long.parseLong(array[0]), Long.parseLong(array[1]));
- }
+ public abstract T getEndValue();
- @Override
- public int compareTo(final PrimaryKeyPosition position) {
- if (null == position) {
- return 1;
- }
- return Long.compare(beginValue, position.beginValue);
- }
+ protected abstract T convert(String value);
+
+ protected abstract char getType();
@Override
- public String toString() {
- return String.format("%d,%d", beginValue, endValue);
+ public final String toString() {
+ return String.format("%s,%s,%s", getType(), getBeginValue(), getEndValue());
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPositionFactory.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPositionFactory.java
new file mode 100644
index 00000000000..55647d6aef1
--- /dev/null
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPositionFactory.java
@@ -0,0 +1,67 @@
+/*
+ * 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.api.ingest.position;
+
+import com.google.common.base.Preconditions;
+import lombok.NonNull;
+
+/**
+ * Primary key position factory.
+ */
+public final class PrimaryKeyPositionFactory {
+
+ /**
+ * New instance by string data.
+ *
+ * @param data string data
+ * @return primary key position
+ */
+ public static IngestPosition<?> newInstance(final String data) {
+ String[] array = data.split(",");
+ Preconditions.checkArgument(3 == array.length, "Unknown primary key position: " + data);
+ Preconditions.checkArgument(1 == array[0].length(), "Invalid primary key position type: " + array[0]);
+ char type = array[0].charAt(0);
+ String beginValue = array[1];
+ String endValue = array[2];
+ switch (type) {
+ case 'i':
+ return new IntegerPrimaryKeyPosition(Long.parseLong(beginValue), Long.parseLong(endValue));
+ case 's':
+ return new StringPrimaryKeyPosition(beginValue, endValue);
+ default:
+ throw new IllegalArgumentException("Unknown primary key position type: " + type);
+ }
+ }
+
+ /**
+ * New instance by begin value and end value.
+ *
+ * @param beginValue begin value
+ * @param endValue end value
+ * @return ingest position
+ */
+ public static IngestPosition<?> newInstance(final @NonNull Object beginValue, final @NonNull Object endValue) {
+ if (beginValue instanceof Number) {
+ return new IntegerPrimaryKeyPosition(((Number) beginValue).longValue(), ((Number) endValue).longValue());
+ } else if (beginValue instanceof CharSequence) {
+ return new StringPrimaryKeyPosition(beginValue.toString(), endValue.toString());
+ } else {
+ throw new IllegalArgumentException("Unknown begin value type: " + beginValue.getClass().getName());
+ }
+ }
+}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/StringPrimaryKeyPosition.java
similarity index 56%
copy from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java
copy to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/StringPrimaryKeyPosition.java
index 11203fcb9ed..948d4ec0b92 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/PrimaryKeyPosition.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/position/StringPrimaryKeyPosition.java
@@ -17,43 +17,35 @@
package org.apache.shardingsphere.data.pipeline.api.ingest.position;
-import com.google.common.base.Preconditions;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
/**
- * Use primary key as position.
+ * String primary key position.
*/
@RequiredArgsConstructor
@Getter
-public final class PrimaryKeyPosition implements IngestPosition<PrimaryKeyPosition> {
+public final class StringPrimaryKeyPosition extends PrimaryKeyPosition<String> implements IngestPosition<StringPrimaryKeyPosition> {
- private final long beginValue;
+ private final String beginValue;
- private final long endValue;
+ private final String endValue;
- /**
- * Init by string data.
- *
- * @param data string data
- * @return primary key position
- */
- public static PrimaryKeyPosition init(final String data) {
- String[] array = data.split(",");
- Preconditions.checkArgument(array.length == 2, "Unknown primary key position: " + data);
- return new PrimaryKeyPosition(Long.parseLong(array[0]), Long.parseLong(array[1]));
+ @Override
+ protected String convert(final String value) {
+ return value;
}
@Override
- public int compareTo(final PrimaryKeyPosition position) {
- if (null == position) {
- return 1;
- }
- return Long.compare(beginValue, position.beginValue);
+ protected char getType() {
+ return 's';
}
@Override
- public String toString() {
- return String.format("%d,%d", beginValue, endValue);
+ public int compareTo(final StringPrimaryKeyPosition position) {
+ if (null == position) {
+ return 1;
+ }
+ return beginValue.compareTo(position.getBeginValue());
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java
index 1b69b8d2d9b..c5661e93664 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/sqlbuilder/PipelineSQLBuilder.java
@@ -44,14 +44,16 @@ public interface PipelineSQLBuilder extends TypedSPI {
}
/**
- * Build inventory dump SQL.
+ * Build inventory dump first SQL.
*
* @param schemaName schema name
* @param tableName table name
* @param uniqueKey unique key
+ * @param uniqueKeyDataType unique key data type
+ * @param firstQuery whether it's the first time query
* @return inventory dump SQL
*/
- String buildInventoryDumpSQL(String schemaName, String tableName, String uniqueKey);
+ String buildInventoryDumpSQL(String schemaName, String tableName, String uniqueKey, int uniqueKeyDataType, boolean firstQuery);
/**
* Build insert SQL.
@@ -117,10 +119,10 @@ public interface PipelineSQLBuilder extends TypedSPI {
* @param schemaName schema name
* @param tableName table name
* @param uniqueKey unique key, it may be primary key, not null
- * @param startUniqueValue start unique value, not null
+ * @param firstQuery first query
* @return query SQL
*/
- String buildChunkedQuerySQL(String schemaName, String tableName, String uniqueKey, Number startUniqueValue);
+ String buildChunkedQuerySQL(String schemaName, String tableName, String uniqueKey, boolean firstQuery);
/**
* Build check empty SQL.
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
index 1f53ed80b13..8ca2576d731 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/DataMatchDataConsistencyCalculateAlgorithm.java
@@ -31,7 +31,6 @@ import org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder
import org.apache.shardingsphere.infra.database.type.DatabaseType;
import org.apache.shardingsphere.infra.database.type.DatabaseTypeFactory;
-import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
@@ -41,8 +40,10 @@ import java.sql.SQLXML;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
+import java.util.Map;
import java.util.Optional;
import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
/**
@@ -62,6 +63,10 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
private int chunkSize;
+ private final Map<String, String> firstSQLCache = new ConcurrentHashMap<>();
+
+ private final Map<String, String> laterSQLCache = new ConcurrentHashMap<>();
+
@Override
public void init(final Properties props) {
this.props = props;
@@ -79,27 +84,19 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
@Override
protected Optional<Object> calculateChunk(final DataConsistencyCalculateParameter parameter) {
- String logicTableName = parameter.getLogicTableName();
- PipelineSQLBuilder sqlBuilder = PipelineSQLBuilderFactory.getInstance(parameter.getDatabaseType());
- String uniqueKey = parameter.getUniqueKey();
CalculatedResult previousCalculatedResult = (CalculatedResult) parameter.getPreviousCalculatedResult();
- Number startUniqueKeyValue = null != previousCalculatedResult ? previousCalculatedResult.getMaxUniqueKeyValue() : -1;
- String sql = sqlBuilder.buildChunkedQuerySQL(parameter.getTableNameSchemaNameMapping().getSchemaName(logicTableName), logicTableName, uniqueKey, startUniqueKeyValue);
- try {
- return query(parameter.getDataSource(), sql, uniqueKey, startUniqueKeyValue, chunkSize);
- } catch (final SQLException ex) {
- throw new PipelineDataConsistencyCheckFailedException(String.format("table %s data check failed.", logicTableName), ex);
- }
- }
-
- private Optional<Object> query(final DataSource dataSource, final String sql, final String uniqueKey, final Number startUniqueKeyValue, final int chunkSize) throws SQLException {
+ String sql = getQuerySQL(parameter);
try (
- Connection connection = dataSource.getConnection();
+ Connection connection = parameter.getDataSource().getConnection();
PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
- preparedStatement.setObject(1, startUniqueKeyValue);
- preparedStatement.setInt(2, chunkSize);
+ if (null == previousCalculatedResult) {
+ preparedStatement.setInt(1, chunkSize);
+ } else {
+ preparedStatement.setObject(1, previousCalculatedResult.getMaxUniqueKeyValue());
+ preparedStatement.setInt(2, chunkSize);
+ }
Collection<Collection<Object>> records = new LinkedList<>();
- Number maxUniqueKeyValue = null;
+ Object maxUniqueKeyValue = null;
try (ResultSet resultSet = preparedStatement.executeQuery()) {
while (resultSet.next()) {
ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
@@ -109,10 +106,25 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
record.add(resultSet.getObject(columnIndex));
}
records.add(record);
- maxUniqueKeyValue = (Number) resultSet.getObject(uniqueKey);
+ maxUniqueKeyValue = resultSet.getObject(parameter.getUniqueKey());
}
}
return records.isEmpty() ? Optional.empty() : Optional.of(new CalculatedResult(maxUniqueKeyValue, records.size(), records));
+ } catch (final SQLException ex) {
+ throw new PipelineDataConsistencyCheckFailedException(String.format("table %s data check failed.", parameter.getLogicTableName()), ex);
+ }
+ }
+
+ private String getQuerySQL(final DataConsistencyCalculateParameter parameter) {
+ PipelineSQLBuilder sqlBuilder = PipelineSQLBuilderFactory.getInstance(parameter.getDatabaseType());
+ String logicTableName = parameter.getLogicTableName();
+ String schemaName = parameter.getTableNameSchemaNameMapping().getSchemaName(logicTableName);
+ String uniqueKey = parameter.getUniqueKey();
+ String cacheKey = schemaName.toLowerCase() + "." + logicTableName.toLowerCase();
+ if (null == parameter.getPreviousCalculatedResult()) {
+ return firstSQLCache.computeIfAbsent(cacheKey, s -> sqlBuilder.buildChunkedQuerySQL(schemaName, logicTableName, uniqueKey, true));
+ } else {
+ return laterSQLCache.computeIfAbsent(cacheKey, s -> sqlBuilder.buildChunkedQuerySQL(schemaName, logicTableName, uniqueKey, false));
}
}
@@ -136,7 +148,7 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
private static final class CalculatedResult {
@NonNull
- private final Number maxUniqueKeyValue;
+ private final Object maxUniqueKeyValue;
private final int recordCount;
@@ -144,16 +156,19 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
@SneakyThrows
@Override
- public boolean equals(final Object o) {
+ public boolean equals(final @NonNull Object o) {
if (this == o) {
return true;
}
if (!(o instanceof CalculatedResult)) {
+ log.warn("CalculatedResult type not match, o.className={}", o.getClass().getName());
return false;
}
final CalculatedResult that = (CalculatedResult) o;
boolean equalsFirst = new EqualsBuilder().append(getRecordCount(), that.getRecordCount()).append(getMaxUniqueKeyValue(), that.getMaxUniqueKeyValue()).isEquals();
if (!equalsFirst) {
+ log.warn("recordCount or maxUniqueKeyValue not match, recordCount1={}, recordCount2={}, maxUniqueKeyValue1={}, maxUniqueKeyValue2={}",
+ getRecordCount(), that.getRecordCount(), getMaxUniqueKeyValue(), that.getMaxUniqueKeyValue());
return false;
}
Iterator<Collection<Object>> thisIterator = this.records.iterator();
@@ -162,7 +177,7 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
Collection<Object> thisNext = thisIterator.next();
Collection<Object> thatNext = thatIterator.next();
if (thisNext.size() != thatNext.size()) {
- log.info("record column size not match, size1={}, size2={}, record1={}, record2={}", thisNext.size(), thatNext.size(), thisNext, thatNext);
+ log.warn("record column size not match, size1={}, size2={}, record1={}, record2={}", thisNext.size(), thatNext.size(), thisNext, thatNext);
return false;
}
Iterator<Object> thisNextIterator = thisNext.iterator();
@@ -174,7 +189,7 @@ public final class DataMatchDataConsistencyCalculateAlgorithm extends AbstractSt
return ((SQLXML) thisResult).getString().equals(((SQLXML) thatResult).getString());
}
if (!new EqualsBuilder().append(thisResult, thatResult).isEquals()) {
- log.info("record column value not match, value1={}, value2={}, record1={}, record2={}", thisResult, thatResult, thisNext, thatNext);
+ log.warn("record column value not match, value1={}, value2={}, record1={}, record2={}", thisResult, thatResult, thisNext, thatNext);
return false;
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
index f4d38a842b3..8d05af12bbb 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
@@ -31,6 +31,7 @@ import org.apache.shardingsphere.data.pipeline.api.ingest.position.FinishedPosit
import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PlaceholderPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPositionFactory;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.Column;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.DataRecord;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.FinishedRecord;
@@ -42,6 +43,7 @@ import org.apache.shardingsphere.data.pipeline.core.ingest.exception.IngestExcep
import org.apache.shardingsphere.data.pipeline.core.metadata.loader.PipelineTableMetaDataLoader;
import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineTableMetaData;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.PipelineSQLBuilderFactory;
+import org.apache.shardingsphere.data.pipeline.core.util.PipelineJdbcUtils;
import org.apache.shardingsphere.data.pipeline.spi.ingest.dumper.InventoryDumper;
import org.apache.shardingsphere.data.pipeline.spi.ratelimit.JobRateLimitAlgorithm;
import org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder;
@@ -61,7 +63,7 @@ import java.util.Optional;
public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor implements InventoryDumper {
@Getter(AccessLevel.PROTECTED)
- private final InventoryDumperConfiguration inventoryDumperConfig;
+ private final InventoryDumperConfiguration dumperConfig;
private final PipelineChannel channel;
@@ -80,7 +82,7 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
if (!StandardPipelineDataSourceConfiguration.class.equals(inventoryDumperConfig.getDataSourceConfig().getClass())) {
throw new UnsupportedOperationException("AbstractInventoryDumper only support StandardPipelineDataSourceConfiguration");
}
- this.inventoryDumperConfig = inventoryDumperConfig;
+ this.dumperConfig = inventoryDumperConfig;
this.channel = channel;
pipelineSQLBuilder = PipelineSQLBuilderFactory.getInstance(inventoryDumperConfig.getDataSourceConfig().getDatabaseType().getType());
this.dataSource = dataSource;
@@ -102,15 +104,21 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
}
private void dump() {
- String schemaName = inventoryDumperConfig.getSchemaName(new LogicTableName(inventoryDumperConfig.getLogicTableName()));
- String sql = pipelineSQLBuilder.buildInventoryDumpSQL(schemaName, inventoryDumperConfig.getActualTableName(), inventoryDumperConfig.getPrimaryKey());
- IngestPosition<?> position = inventoryDumperConfig.getPosition();
- log.info("inventory dump, sql={}, position={}", sql, position);
+ String schemaName = dumperConfig.getSchemaName(new LogicTableName(dumperConfig.getLogicTableName()));
+ int uniqueKeyDataType = dumperConfig.getUniqueKeyDataType();
+ String firstSQL = pipelineSQLBuilder.buildInventoryDumpSQL(schemaName, dumperConfig.getActualTableName(), dumperConfig.getPrimaryKey(), uniqueKeyDataType, true);
+ String laterSQL = pipelineSQLBuilder.buildInventoryDumpSQL(schemaName, dumperConfig.getActualTableName(), dumperConfig.getPrimaryKey(), uniqueKeyDataType, false);
+ IngestPosition<?> position = dumperConfig.getPosition();
+ log.info("inventory dump, uniqueKeyDataType={}, firstSQL={}, laterSQL={}, position={}", uniqueKeyDataType, firstSQL, laterSQL, position);
+ if (position instanceof FinishedPosition) {
+ log.info("It is already finished, ignore");
+ return;
+ }
+ Object startUniqueKeyValue = getPositionBeginValue(position);
try (Connection conn = dataSource.getConnection()) {
int round = 1;
- Number startUniqueKeyValue = getPositionBeginValue(position) - 1;
- Optional<Number> maxUniqueKeyValue;
- while ((maxUniqueKeyValue = dump0(conn, sql, startUniqueKeyValue, round++)).isPresent()) {
+ Optional<Object> maxUniqueKeyValue;
+ while ((maxUniqueKeyValue = dump0(conn, 1 == round ? firstSQL : laterSQL, uniqueKeyDataType, startUniqueKeyValue, round++)).isPresent()) {
startUniqueKeyValue = maxUniqueKeyValue.get();
if (!isRunning()) {
log.info("inventory dump, running is false, break");
@@ -132,20 +140,28 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
return tableMetaDataLazyInitializer.get();
}
- private Optional<Number> dump0(final Connection conn, final String sql, final Number startUniqueKeyValue, final int round) throws SQLException {
+ private Optional<Object> dump0(final Connection conn, final String sql, final int uniqueKeyDataType, final Object startUniqueKeyValue, final int round) throws SQLException {
if (null != rateLimitAlgorithm) {
rateLimitAlgorithm.intercept(JobOperationType.SELECT, 1);
}
PipelineTableMetaData tableMetaData = getTableMetaData();
try (PreparedStatement preparedStatement = createPreparedStatement(conn, sql)) {
- preparedStatement.setObject(1, startUniqueKeyValue);
- preparedStatement.setObject(2, getPositionEndValue(inventoryDumperConfig.getPosition()));
- preparedStatement.setInt(3, batchSize);
+ preparedStatement.setFetchSize(batchSize);
+ if (PipelineJdbcUtils.isIntegerColumn(uniqueKeyDataType)) {
+ preparedStatement.setObject(1, startUniqueKeyValue);
+ preparedStatement.setObject(2, getPositionEndValue(dumperConfig.getPosition()));
+ preparedStatement.setInt(3, batchSize);
+ } else if (PipelineJdbcUtils.isStringColumn(uniqueKeyDataType)) {
+ preparedStatement.setObject(1, startUniqueKeyValue);
+ preparedStatement.setInt(2, batchSize);
+ } else {
+ throw new IllegalArgumentException("Unsupported uniqueKeyDataType: " + uniqueKeyDataType);
+ }
try (ResultSet resultSet = preparedStatement.executeQuery()) {
ResultSetMetaData metaData = resultSet.getMetaData();
int rowCount = 0;
- Number maxUniqueKeyValue = null;
- String logicTableName = inventoryDumperConfig.getLogicTableName();
+ Object maxUniqueKeyValue = null;
+ String logicTableName = dumperConfig.getLogicTableName();
while (resultSet.next()) {
DataRecord record = new DataRecord(newPosition(resultSet), metaData.getColumnCount());
record.setType(IngestDataChangeType.INSERT);
@@ -154,7 +170,7 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
boolean isPrimaryKey = tableMetaData.isPrimaryKey(i - 1);
Object value = readValue(resultSet, i);
if (isPrimaryKey) {
- maxUniqueKeyValue = (Number) value;
+ maxUniqueKeyValue = value;
}
record.addColumn(new Column(metaData.getColumnName(i), value, true, isPrimaryKey));
}
@@ -165,7 +181,7 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
break;
}
}
- if (0 == round % 50) {
+ if (PipelineJdbcUtils.isStringColumn(uniqueKeyDataType) && 0 == round % 50) {
log.info("dump, round={}, rowCount={}, maxUniqueKeyValue={}", round, rowCount, maxUniqueKeyValue);
}
return Optional.ofNullable(maxUniqueKeyValue);
@@ -173,17 +189,17 @@ public abstract class AbstractInventoryDumper extends AbstractLifecycleExecutor
}
}
- private long getPositionBeginValue(final IngestPosition<?> position) {
- return position instanceof PrimaryKeyPosition ? ((PrimaryKeyPosition) position).getBeginValue() : 0;
+ private Object getPositionBeginValue(final IngestPosition<?> position) {
+ return ((PrimaryKeyPosition<?>) position).getBeginValue();
}
- private long getPositionEndValue(final IngestPosition<?> position) {
- return position instanceof PrimaryKeyPosition ? ((PrimaryKeyPosition) position).getEndValue() : Integer.MAX_VALUE;
+ private Object getPositionEndValue(final IngestPosition<?> position) {
+ return ((PrimaryKeyPosition<?>) position).getEndValue();
}
private IngestPosition<?> newPosition(final ResultSet rs) throws SQLException {
- return null == inventoryDumperConfig.getPrimaryKey() ? new PlaceholderPosition()
- : new PrimaryKeyPosition(rs.getLong(inventoryDumperConfig.getPrimaryKey()), ((PrimaryKeyPosition) inventoryDumperConfig.getPosition()).getEndValue());
+ return null == dumperConfig.getPrimaryKey() ? new PlaceholderPosition()
+ : PrimaryKeyPositionFactory.newInstance(rs.getObject(dumperConfig.getPrimaryKey()), ((PrimaryKeyPosition<?>) dumperConfig.getPosition()).getEndValue());
}
protected abstract PreparedStatement createPreparedStatement(Connection connection, String sql) throws SQLException;
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java
index 3a4b79d2214..06c03c52878 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/FinishedCheckJob.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.data.pipeline.core.job;
+import io.vertx.core.impl.ConcurrentHashSet;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.PipelineJobAPIFactory;
import org.apache.shardingsphere.data.pipeline.api.RuleAlteredJobAPI;
@@ -35,12 +36,15 @@ import org.apache.shardingsphere.infra.yaml.engine.YamlEngine;
import java.util.List;
import java.util.Map;
+import java.util.Set;
@Slf4j
public final class FinishedCheckJob implements SimpleJob {
private final RuleAlteredJobAPI ruleAlteredJobAPI = PipelineJobAPIFactory.getInstance();
+ private final Set<String> onCheckJobIds = new ConcurrentHashSet<>();
+
// TODO only one proxy node could do data consistency check in proxy cluster
@Override
public void execute(final ShardingContext shardingContext) {
@@ -50,9 +54,14 @@ public final class FinishedCheckJob implements SimpleJob {
continue;
}
String jobId = jobInfo.getJobId();
+ if (onCheckJobIds.contains(jobId)) {
+ log.info("check not completed for job {}, ignore", jobId);
+ continue;
+ }
if (isNotAllowDataCheck(jobId)) {
continue;
}
+ onCheckJobIds.add(jobId);
try {
// TODO refactor: dispatch to different job types
RuleAlteredJobConfiguration jobConfig = YamlEngine.unmarshal(jobInfo.getJobParameter(), RuleAlteredJobConfiguration.class, true);
@@ -93,6 +102,8 @@ public final class FinishedCheckJob implements SimpleJob {
} catch (final Exception ex) {
// CHECKSTYLE:ON
log.error("scaling job {} finish check failed!", jobId, ex);
+ } finally {
+ onCheckJobIds.remove(jobId);
}
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapper.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapper.java
index b8d28f8763a..f75a247383c 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapper.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapper.java
@@ -20,7 +20,7 @@ package org.apache.shardingsphere.data.pipeline.core.job.progress.yaml;
import com.google.common.base.Strings;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.FinishedPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PlaceholderPosition;
-import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPositionFactory;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress;
import org.apache.shardingsphere.data.pipeline.api.task.progress.IncrementalTaskProgress;
@@ -111,7 +111,7 @@ public final class JobProgressYamlSwapper {
}
private Function<Entry<String, String>, InventoryTaskProgress> getInventoryTaskProgressFunction() {
- return entry -> new InventoryTaskProgress(Strings.isNullOrEmpty(entry.getValue()) ? new PlaceholderPosition() : PrimaryKeyPosition.init(entry.getValue()));
+ return entry -> new InventoryTaskProgress(Strings.isNullOrEmpty(entry.getValue()) ? new PlaceholderPosition() : PrimaryKeyPositionFactory.newInstance(entry.getValue()));
}
private Map<String, IncrementalTaskProgress> getIncrementalTaskProgressMap(final String databaseType, final Map<String, YamlJobProgress.YamlIncremental> incremental) {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
index dcde377677e..369d1261b14 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
@@ -17,12 +17,13 @@
package org.apache.shardingsphere.data.pipeline.core.sqlbuilder;
-import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
+import lombok.NonNull;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.Column;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.DataRecord;
import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
import org.apache.shardingsphere.data.pipeline.core.record.RecordUtil;
+import org.apache.shardingsphere.data.pipeline.core.util.PipelineJdbcUtils;
import org.apache.shardingsphere.data.pipeline.spi.sqlbuilder.PipelineSQLBuilder;
import org.apache.shardingsphere.infra.database.type.DatabaseTypeFactory;
@@ -72,9 +73,18 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
}
@Override
- public String buildInventoryDumpSQL(final String schemaName, final String tableName, final String uniqueKey) {
+ public String buildInventoryDumpSQL(final String schemaName, final String tableName, final String uniqueKey, final int uniqueKeyDataType, final boolean firstQuery) {
+ String decoratedTableName = decorate(schemaName, tableName);
String quotedUniqueKey = quote(uniqueKey);
- return "SELECT * FROM " + decorate(schemaName, tableName) + " WHERE " + quotedUniqueKey + " > ? AND " + quotedUniqueKey + " <= ? ORDER BY " + quotedUniqueKey + " ASC LIMIT ?";
+ if (PipelineJdbcUtils.isIntegerColumn(uniqueKeyDataType)) {
+ return "SELECT * FROM " + decoratedTableName + " WHERE " + quotedUniqueKey + " " + (firstQuery ? ">=" : ">") + " ?"
+ + " AND " + quotedUniqueKey + " <= ? ORDER BY " + quotedUniqueKey + " ASC LIMIT ?";
+ } else if (PipelineJdbcUtils.isStringColumn(uniqueKeyDataType)) {
+ return "SELECT * FROM " + decoratedTableName + " WHERE " + quotedUniqueKey + " " + (firstQuery ? ">=" : ">") + " ?"
+ + " ORDER BY " + quotedUniqueKey + " ASC LIMIT ?";
+ } else {
+ throw new IllegalArgumentException("Unknown uniqueKeyDataType: " + uniqueKeyDataType);
+ }
}
protected String decorate(final String schemaName, final String tableName) {
@@ -173,10 +183,12 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
}
@Override
- public String buildChunkedQuerySQL(final String schemaName, final String tableName, final String uniqueKey, final Number startUniqueValue) {
- Preconditions.checkNotNull(uniqueKey, "uniqueKey is null");
- Preconditions.checkNotNull(startUniqueValue, "startUniqueValue is null");
- return "SELECT * FROM " + decorate(schemaName, tableName) + " WHERE " + quote(uniqueKey) + " > ? ORDER BY " + quote(uniqueKey) + " ASC LIMIT ?";
+ public String buildChunkedQuerySQL(final @NonNull String schemaName, final @NonNull String tableName, final @NonNull String uniqueKey, final boolean firstQuery) {
+ if (firstQuery) {
+ return "SELECT * FROM " + decorate(schemaName, tableName) + " ORDER BY " + quote(uniqueKey) + " ASC LIMIT ?";
+ } else {
+ return "SELECT * FROM " + decorate(schemaName, tableName) + " WHERE " + quote(uniqueKey) + " > ? ORDER BY " + quote(uniqueKey) + " ASC LIMIT ?";
+ }
}
@Override
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineJdbcUtils.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineJdbcUtils.java
new file mode 100644
index 00000000000..d1544c64d3a
--- /dev/null
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineJdbcUtils.java
@@ -0,0 +1,60 @@
+/*
+ * 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.util;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import java.sql.Types;
+
+/**
+ * Pipeline JDBC utils.
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class PipelineJdbcUtils {
+
+ /**
+ * Whether column is integer type.
+ *
+ * @param columnType column type, value of java.sql.Types
+ * @return true or false
+ */
+ public static boolean isIntegerColumn(final int columnType) {
+ return Types.INTEGER == columnType || Types.BIGINT == columnType || Types.SMALLINT == columnType || Types.TINYINT == columnType;
+ }
+
+ /**
+ * Whether column is string column.
+ *
+ * @param columnType column type, value of java.sql.Types
+ * @return true or false
+ */
+ public static boolean isStringColumn(final int columnType) {
+ switch (columnType) {
+ case Types.CHAR:
+ case Types.VARCHAR:
+ case Types.LONGVARCHAR:
+ case Types.NCHAR:
+ case Types.NVARCHAR:
+ case Types.LONGNVARCHAR:
+ return true;
+ default:
+ return false;
+ }
+ }
+}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java
index 46a8ca916ba..4b2972c4fce 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitter.java
@@ -23,8 +23,10 @@ import org.apache.shardingsphere.data.pipeline.api.config.ingest.InventoryDumper
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.RuleAlteredJobConfiguration;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.TaskConfiguration;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.IntegerPrimaryKeyPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PlaceholderPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.StringPrimaryKeyPosition;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress;
import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
@@ -36,6 +38,7 @@ import org.apache.shardingsphere.data.pipeline.core.metadata.loader.PipelineTabl
import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineTableMetaData;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.PipelineSQLBuilderFactory;
import org.apache.shardingsphere.data.pipeline.core.task.InventoryTask;
+import org.apache.shardingsphere.data.pipeline.core.util.PipelineJdbcUtils;
import org.apache.shardingsphere.data.pipeline.scenario.rulealtered.RuleAlteredContext;
import org.apache.shardingsphere.data.pipeline.scenario.rulealtered.RuleAlteredJobContext;
import org.apache.shardingsphere.data.pipeline.spi.ingest.channel.PipelineChannelCreator;
@@ -49,7 +52,6 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
-import java.sql.Types;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
@@ -122,6 +124,7 @@ public final class InventoryTaskSplitter {
splitDumperConfig.setActualTableName(dumperConfig.getActualTableName());
splitDumperConfig.setLogicTableName(dumperConfig.getLogicTableName());
splitDumperConfig.setPrimaryKey(dumperConfig.getPrimaryKey());
+ splitDumperConfig.setUniqueKeyDataType(dumperConfig.getUniqueKeyDataType());
splitDumperConfig.setBatchSize(batchSize);
splitDumperConfig.setRateLimitAlgorithm(rateLimitAlgorithm);
result.add(splitDumperConfig);
@@ -133,53 +136,49 @@ public final class InventoryTaskSplitter {
final DataSource dataSource, final PipelineTableMetaDataLoader metaDataLoader) {
JobProgress initProgress = jobContext.getInitProgress();
String schemaName = dumperConfig.getSchemaName(new LogicTableName(dumperConfig.getLogicTableName()));
+ String actualTableName = dumperConfig.getActualTableName();
+ PipelineTableMetaData tableMetaData = metaDataLoader.getTableMetaData(schemaName, actualTableName);
if (null != initProgress && initProgress.getStatus() != JobStatus.PREPARING_FAILURE) {
Collection<IngestPosition<?>> result = initProgress.getInventoryPosition(dumperConfig.getActualTableName()).values();
for (IngestPosition<?> each : result) {
if (each instanceof PrimaryKeyPosition) {
- String primaryKey = metaDataLoader.getTableMetaData(schemaName, dumperConfig.getActualTableName()).getPrimaryKeyColumns().get(0);
+ String primaryKey = tableMetaData.getPrimaryKeyColumns().get(0);
dumperConfig.setPrimaryKey(primaryKey);
+ dumperConfig.setUniqueKeyDataType(tableMetaData.getColumnMetaData(primaryKey).getDataType());
break;
}
}
// Do NOT filter FinishedPosition here, since whole inventory tasks are required in job progress when persisting to register center.
return result;
}
- PipelineTableMetaData tableMetaData = metaDataLoader.getTableMetaData(schemaName, dumperConfig.getActualTableName());
- if (isSpiltByPrimaryKeyRange(tableMetaData, dumperConfig.getActualTableName())) {
- String primaryKey = tableMetaData.getPrimaryKeyColumns().get(0);
- dumperConfig.setPrimaryKey(primaryKey);
- return getPositionByPrimaryKeyRange(jobContext, dataSource, dumperConfig);
+ checkPrimaryKey(tableMetaData, actualTableName);
+ String primaryKey = tableMetaData.getPrimaryKeyColumns().get(0);
+ dumperConfig.setPrimaryKey(primaryKey);
+ int primaryKeyDataType = tableMetaData.getColumnMetaData(primaryKey).getDataType();
+ dumperConfig.setUniqueKeyDataType(primaryKeyDataType);
+ if (PipelineJdbcUtils.isIntegerColumn(primaryKeyDataType)) {
+ return getPositionByIntegerPrimaryKeyRange(jobContext, dataSource, dumperConfig);
+ } else if (PipelineJdbcUtils.isStringColumn(primaryKeyDataType)) {
+ return getPositionByStringPrimaryKeyRange();
+ } else {
+ throw new PipelineJobCreationException(String.format("Can not split range for table %s, reason: primary key is not integer or string type", actualTableName));
}
- throw new PipelineJobCreationException("Can not split by primary key range for table " + dumperConfig.getActualTableName());
}
- private boolean isSpiltByPrimaryKeyRange(final PipelineTableMetaData tableMetaData, final String tableName) {
+ private void checkPrimaryKey(final PipelineTableMetaData tableMetaData, final String tableName) {
if (null == tableMetaData) {
- log.warn("Can't split range for table {}, reason: can not get table metadata ", tableName);
- return false;
+ throw new PipelineJobCreationException(String.format("Can not split range for table %s, reason: can not get table metadata ", tableName));
}
List<String> primaryKeys = tableMetaData.getPrimaryKeyColumns();
if (null == primaryKeys || primaryKeys.isEmpty()) {
- log.warn("Can't split range for table {}, reason: no primary key", tableName);
- return false;
+ throw new PipelineJobCreationException(String.format("Can not split range for table %s, reason: no primary key", tableName));
}
if (primaryKeys.size() > 1) {
- log.warn("Can't split range for table {}, reason: primary key is union primary", tableName);
- return false;
+ throw new PipelineJobCreationException(String.format("Can not split range for table %s, reason: primary key is union primary", tableName));
}
- if (isNotIntegerPrimary(tableMetaData.getColumnMetaData(primaryKeys.get(0)).getDataType())) {
- log.warn("Can't split range for table {}, reason: primary key is not integer number", tableName);
- return false;
- }
- return true;
- }
-
- private boolean isNotIntegerPrimary(final int columnType) {
- return Types.INTEGER != columnType && Types.BIGINT != columnType && Types.SMALLINT != columnType && Types.TINYINT != columnType;
}
- private Collection<IngestPosition<?>> getPositionByPrimaryKeyRange(final RuleAlteredJobContext jobContext, final DataSource dataSource, final InventoryDumperConfiguration dumperConfig) {
+ private Collection<IngestPosition<?>> getPositionByIntegerPrimaryKeyRange(final RuleAlteredJobContext jobContext, final DataSource dataSource, final InventoryDumperConfiguration dumperConfig) {
Collection<IngestPosition<?>> result = new LinkedList<>();
RuleAlteredJobConfiguration jobConfig = jobContext.getJobConfig();
String sql = PipelineSQLBuilderFactory.getInstance(jobConfig.getSourceDatabaseType())
@@ -187,6 +186,7 @@ public final class InventoryTaskSplitter {
try (
Connection connection = dataSource.getConnection();
PreparedStatement ps = connection.prepareStatement(sql)) {
+ // TODO query minimum value less than 0
long beginId = 0;
for (int i = 0; i < Integer.MAX_VALUE; i++) {
ps.setLong(1, beginId);
@@ -201,17 +201,23 @@ public final class InventoryTaskSplitter {
log.info("getPositionByPrimaryKeyRange, endId is 0, break, tableName={}, primaryKey={}, beginId={}", dumperConfig.getActualTableName(), dumperConfig.getPrimaryKey(), beginId);
break;
}
- result.add(new PrimaryKeyPosition(beginId, endId));
+ result.add(new IntegerPrimaryKeyPosition(beginId, endId));
beginId = endId + 1;
}
}
// fix empty table missing inventory task
if (0 == result.size()) {
- result.add(new PrimaryKeyPosition(0, 0));
+ result.add(new IntegerPrimaryKeyPosition(0, 0));
}
} catch (final SQLException ex) {
throw new PipelineJobPrepareFailedException(String.format("Split task for table %s by primary key %s error", dumperConfig.getActualTableName(), dumperConfig.getPrimaryKey()), ex);
}
return result;
}
+
+ private Collection<IngestPosition<?>> getPositionByStringPrimaryKeyRange() {
+ Collection<IngestPosition<?>> result = new LinkedList<>();
+ result.add(new StringPrimaryKeyPosition("!", "~"));
+ return result;
+ }
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java
index bd473aef981..b72cfd6d331 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/test/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/algorithm/fixture/FixturePipelineSQLBuilder.java
@@ -32,7 +32,7 @@ import java.util.Set;
public final class FixturePipelineSQLBuilder implements PipelineSQLBuilder {
@Override
- public String buildInventoryDumpSQL(final String schemaName, final String tableName, final String uniqueKey) {
+ public String buildInventoryDumpSQL(final String schemaName, final String tableName, final String uniqueKey, final int uniqueKeyDataType, final boolean firstQuery) {
return "";
}
@@ -67,7 +67,7 @@ public final class FixturePipelineSQLBuilder implements PipelineSQLBuilder {
}
@Override
- public String buildChunkedQuerySQL(final String schemaName, final String tableName, final String uniqueKey, final Number startUniqueValue) {
+ public String buildChunkedQuerySQL(final String schemaName, final String tableName, final String uniqueKey, final boolean firstQuery) {
return "";
}
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/api/impl/GovernanceRepositoryAPIImplTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/api/impl/GovernanceRepositoryAPIImplTest.java
index 830e0c2393b..9532fb43625 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/api/impl/GovernanceRepositoryAPIImplTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/api/impl/GovernanceRepositoryAPIImplTest.java
@@ -42,6 +42,7 @@ import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEve
import org.junit.BeforeClass;
import org.junit.Test;
+import java.sql.Types;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
@@ -152,6 +153,7 @@ public final class GovernanceRepositoryAPIImplTest {
dumperConfig.setActualTableName("t_order");
dumperConfig.setLogicTableName("t_order");
dumperConfig.setPrimaryKey("order_id");
+ dumperConfig.setUniqueKeyDataType(Types.INTEGER);
dumperConfig.setShardingItem(0);
PipelineDataSourceWrapper dataSource = mock(PipelineDataSourceWrapper.class);
PipelineTableMetaDataLoader metaDataLoader = new PipelineTableMetaDataLoader(dataSource);
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/PrimaryKeyPositionTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/IntegerPrimaryKeyPositionTest.java
similarity index 72%
copy from shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/PrimaryKeyPositionTest.java
copy to shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/IntegerPrimaryKeyPositionTest.java
index cd75576b571..1b5cf49a7b0 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/PrimaryKeyPositionTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/IntegerPrimaryKeyPositionTest.java
@@ -17,32 +17,33 @@
package org.apache.shardingsphere.data.pipeline.core.ingest.position;
-import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.IntegerPrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPositionFactory;
import org.junit.Test;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
-public final class PrimaryKeyPositionTest {
+public final class IntegerPrimaryKeyPositionTest {
@Test
public void assertCompareTo() {
- PrimaryKeyPosition position1 = new PrimaryKeyPosition(1, 100);
- PrimaryKeyPosition position2 = new PrimaryKeyPosition(101, 200);
+ IntegerPrimaryKeyPosition position1 = new IntegerPrimaryKeyPosition(1, 100);
+ IntegerPrimaryKeyPosition position2 = new IntegerPrimaryKeyPosition(101, 200);
assertThat(position1.compareTo(null), is(1));
assertTrue(position1.compareTo(position2) < 0);
}
@Test
public void assertInit() {
- PrimaryKeyPosition position = PrimaryKeyPosition.init("1,100");
+ IntegerPrimaryKeyPosition position = (IntegerPrimaryKeyPosition) PrimaryKeyPositionFactory.newInstance("i,1,100");
assertThat(position.getBeginValue(), is(1L));
assertThat(position.getEndValue(), is(100L));
}
@Test
public void assertToString() {
- assertThat(new PrimaryKeyPosition(1, 100).toString(), is("1,100"));
+ assertThat(new IntegerPrimaryKeyPosition(1, 100).toString(), is("i,1,100"));
}
}
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/PrimaryKeyPositionTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/StringPrimaryKeyPositionTest.java
similarity index 67%
rename from shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/PrimaryKeyPositionTest.java
rename to shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/StringPrimaryKeyPositionTest.java
index cd75576b571..8ef24db459e 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/PrimaryKeyPositionTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/position/StringPrimaryKeyPositionTest.java
@@ -17,32 +17,33 @@
package org.apache.shardingsphere.data.pipeline.core.ingest.position;
-import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPositionFactory;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.StringPrimaryKeyPosition;
import org.junit.Test;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
-public final class PrimaryKeyPositionTest {
+public final class StringPrimaryKeyPositionTest {
@Test
public void assertCompareTo() {
- PrimaryKeyPosition position1 = new PrimaryKeyPosition(1, 100);
- PrimaryKeyPosition position2 = new PrimaryKeyPosition(101, 200);
+ StringPrimaryKeyPosition position1 = new StringPrimaryKeyPosition("abc", "def");
+ StringPrimaryKeyPosition position2 = new StringPrimaryKeyPosition("hi", "jk");
assertThat(position1.compareTo(null), is(1));
assertTrue(position1.compareTo(position2) < 0);
}
@Test
public void assertInit() {
- PrimaryKeyPosition position = PrimaryKeyPosition.init("1,100");
- assertThat(position.getBeginValue(), is(1L));
- assertThat(position.getEndValue(), is(100L));
+ StringPrimaryKeyPosition position = (StringPrimaryKeyPosition) PrimaryKeyPositionFactory.newInstance("s,hi,jk");
+ assertThat(position.getBeginValue(), is("hi"));
+ assertThat(position.getEndValue(), is("jk"));
}
@Test
public void assertToString() {
- assertThat(new PrimaryKeyPosition(1, 100).toString(), is("1,100"));
+ assertThat(new StringPrimaryKeyPosition("hi", "jk").toString(), is("s,hi,jk"));
}
}
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
index 4e009c28731..2c037589b42 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
@@ -20,7 +20,7 @@ package org.apache.shardingsphere.data.pipeline.core.job.progress;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.FinishedPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PlaceholderPosition;
-import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.IntegerPrimaryKeyPosition;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress;
import org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.JobProgressYamlSwapper;
@@ -66,7 +66,7 @@ public final class JobProgressTest {
assertThat(jobProgress.getInventoryPosition("ds0").size(), is(2));
assertTrue(jobProgress.getInventoryPosition("ds0").get("ds0.t_1") instanceof FinishedPosition);
assertTrue(jobProgress.getInventoryPosition("ds1").get("ds1.t_1") instanceof PlaceholderPosition);
- assertTrue(jobProgress.getInventoryPosition("ds1").get("ds1.t_2") instanceof PrimaryKeyPosition);
+ assertTrue(jobProgress.getInventoryPosition("ds1").get("ds1.t_2") instanceof IntegerPrimaryKeyPosition);
}
@Test
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapperTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapperTest.java
index 7849735317d..3d74eb32937 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapperTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/JobProgressYamlSwapperTest.java
@@ -46,7 +46,7 @@ public final class JobProgressYamlSwapperTest {
assertThat(actual.getInventory().getFinished().length, is(2));
assertArrayEquals(actual.getInventory().getFinished(), new String[]{"ds0.t_2", "ds0.t_1"});
assertThat(actual.getInventory().getUnfinished().size(), is(2));
- assertThat(actual.getInventory().getUnfinished().get("ds1.t_2"), is("1,2"));
+ assertThat(actual.getInventory().getUnfinished().get("ds1.t_2"), is("i,1,2"));
assertThat(actual.getInventory().getUnfinished().get("ds1.t_1"), is(""));
assertThat(actual.getIncremental().size(), is(1));
assertTrue(actual.getIncremental().containsKey("ds0"));
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java
index 96a45891d01..45740aa2a16 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTaskTest.java
@@ -23,7 +23,7 @@ import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.TaskConfig
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.FinishedPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPosition;
-import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.IntegerPrimaryKeyPosition;
import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.ingest.exception.IngestException;
import org.apache.shardingsphere.data.pipeline.core.metadata.loader.PipelineTableMetaDataLoader;
@@ -38,6 +38,7 @@ import org.junit.Test;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
+import java.sql.Types;
import static org.junit.Assert.assertFalse;
@@ -69,9 +70,11 @@ public final class InventoryTaskTest {
inventoryDumperConfig.setLogicTableName("t_non_exist");
IngestPosition<?> position = taskConfig.getDumperConfig().getPosition();
if (null == position) {
- position = new PrimaryKeyPosition(0, 1000);
+ position = new IntegerPrimaryKeyPosition(0, 1000);
}
inventoryDumperConfig.setPosition(position);
+ inventoryDumperConfig.setPrimaryKey("order_id");
+ inventoryDumperConfig.setUniqueKeyDataType(Types.INTEGER);
PipelineDataSourceWrapper dataSource = DATA_SOURCE_MANAGER.getDataSource(inventoryDumperConfig.getDataSourceConfig());
PipelineTableMetaDataLoader metaDataLoader = new PipelineTableMetaDataLoader(dataSource);
try (
@@ -90,9 +93,10 @@ public final class InventoryTaskTest {
inventoryDumperConfig.setActualTableName("t_order");
inventoryDumperConfig.setLogicTableName("t_order");
inventoryDumperConfig.setPrimaryKey("order_id");
+ inventoryDumperConfig.setUniqueKeyDataType(Types.INTEGER);
IngestPosition<?> position = taskConfig.getDumperConfig().getPosition();
if (null == position) {
- position = new PrimaryKeyPosition(0, 1000);
+ position = new IntegerPrimaryKeyPosition(0, 1000);
}
inventoryDumperConfig.setPosition(position);
PipelineDataSourceWrapper dataSource = DATA_SOURCE_MANAGER.getDataSource(inventoryDumperConfig.getDataSourceConfig());
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java
index a5823bfe5fe..6bc71ec7b2b 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/prepare/InventoryTaskSplitterTest.java
@@ -19,14 +19,16 @@ package org.apache.shardingsphere.data.pipeline.scenario.rulealtered.prepare;
import org.apache.shardingsphere.data.pipeline.api.config.ingest.DumperConfiguration;
import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.TaskConfiguration;
-import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPosition;
+import org.apache.shardingsphere.data.pipeline.api.ingest.position.IntegerPrimaryKeyPosition;
import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCreationException;
import org.apache.shardingsphere.data.pipeline.core.task.InventoryTask;
import org.apache.shardingsphere.data.pipeline.core.util.JobConfigurationBuilder;
+import org.apache.shardingsphere.data.pipeline.core.util.PipelineContextUtil;
import org.apache.shardingsphere.data.pipeline.scenario.rulealtered.RuleAlteredJobContext;
import org.junit.After;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
import javax.sql.DataSource;
@@ -48,6 +50,11 @@ public final class InventoryTaskSplitterTest {
private InventoryTaskSplitter inventoryTaskSplitter;
+ @BeforeClass
+ public static void beforeClass() {
+ PipelineContextUtil.mockModeConfigAndContextManager();
+ }
+
@Before
public void setUp() {
initJobContext();
@@ -71,8 +78,8 @@ public final class InventoryTaskSplitterTest {
initEmptyTablePrimaryEnvironment(taskConfig.getDumperConfig());
List<InventoryTask> actual = inventoryTaskSplitter.splitInventoryData(jobContext);
assertThat(actual.size(), is(1));
- assertThat(((PrimaryKeyPosition) actual.get(0).getProgress().getPosition()).getBeginValue(), is(0L));
- assertThat(((PrimaryKeyPosition) actual.get(0).getProgress().getPosition()).getEndValue(), is(0L));
+ assertThat(((IntegerPrimaryKeyPosition) actual.get(0).getProgress().getPosition()).getBeginValue(), is(0L));
+ assertThat(((IntegerPrimaryKeyPosition) actual.get(0).getProgress().getPosition()).getEndValue(), is(0L));
}
@Test
@@ -81,11 +88,11 @@ public final class InventoryTaskSplitterTest {
initIntPrimaryEnvironment(taskConfig.getDumperConfig());
List<InventoryTask> actual = inventoryTaskSplitter.splitInventoryData(jobContext);
assertThat(actual.size(), is(10));
- assertThat(((PrimaryKeyPosition) actual.get(9).getProgress().getPosition()).getBeginValue(), is(91L));
- assertThat(((PrimaryKeyPosition) actual.get(9).getProgress().getPosition()).getEndValue(), is(100L));
+ assertThat(((IntegerPrimaryKeyPosition) actual.get(9).getProgress().getPosition()).getBeginValue(), is(91L));
+ assertThat(((IntegerPrimaryKeyPosition) actual.get(9).getProgress().getPosition()).getEndValue(), is(100L));
}
- @Test(expected = PipelineJobCreationException.class)
+ @Test
public void assertSplitInventoryDataWithCharPrimary() throws SQLException {
initCharPrimaryEnvironment(taskConfig.getDumperConfig());
inventoryTaskSplitter.splitInventoryData(jobContext);
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml
index ab845aa4ebf..4fd9e105258 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/config_sharding_sphere_jdbc_source.yaml
@@ -15,7 +15,7 @@
# limitations under the License.
#
-schemaName: logic_db
+databaseName: logic_db
dataSources:
ds_0:
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-finished.yaml b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-finished.yaml
index 138537e9052..4b2efffdf7c 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-finished.yaml
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-finished.yaml
@@ -23,9 +23,9 @@
# position: ''
inventory:
unfinished:
- ds0.t_1: 1,2
+ ds0.t_1: i,1,2
ds0.t_2: ''
- ds1.t_2: 1,2
+ ds1.t_2: i,1,2
ds1.t_1: ''
sourceDatabaseType: H2
status: RUNNING
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-inventory.yaml b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-inventory.yaml
index c055b8068d7..58a4d08e708 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-inventory.yaml
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress-no-inventory.yaml
@@ -21,11 +21,5 @@
lastEventTimestamps: 0
latestActiveTimeMillis: 0
position: ''
-#inventory:
-# unfinished:
-# ds0.t_1: 1,2
-# ds0.t_2: ''
-# ds1.t_2: 1,2
-# ds1.t_1: ''
sourceDatabaseType: H2
status: RUNNING
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress.yaml b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress.yaml
index 55e1f65ab3f..2333522962f 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress.yaml
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/job-progress.yaml
@@ -26,7 +26,7 @@ inventory:
- ds0.t_2
- ds0.t_1
unfinished:
- ds1.t_2: 1,2
+ ds1.t_2: i,1,2
ds1.t_1: ''
sourceDatabaseType: H2
status: RUNNING