You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2023/04/28 03:14:23 UTC
[shardingsphere] branch master updated: Fix table without unique key increment task write failed (#25345)
This is an automated email from the ASF dual-hosted git repository.
zhonghongsheng 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 95de254e0ca Fix table without unique key increment task write failed (#25345)
95de254e0ca is described below
commit 95de254e0cafe1c2f6b1416e1ba78ff18d35c6c9
Author: Xinze Guo <10...@users.noreply.github.com>
AuthorDate: Fri Apr 28 11:14:16 2023 +0800
Fix table without unique key increment task write failed (#25345)
* Fix table without unique key increment task write failed
* Remove delete update at no unique key e2e
* Improve show data consistency check
* Fix test
* Add more log
---
.../data/pipeline/api/ingest/record/Column.java | 2 +-
.../pipeline/core/importer/DataSourceImporter.java | 12 +++--
.../data/pipeline/core/record/RecordUtils.java | 17 ++++---
.../sqlbuilder/AbstractPipelineSQLBuilder.java | 2 +-
.../core/sqlbuilder/PipelineSQLBuilderTest.java | 26 +++++++++-
.../MigrationDataConsistencyChecker.java | 2 +
.../primarykey/IndexesMigrationE2EIT.java | 58 +++++++++++++++++++---
.../pipeline/cases/task/E2EIncrementalTask.java | 1 +
.../core/importer/DataSourceImporterTest.java | 4 +-
9 files changed, 103 insertions(+), 21 deletions(-)
diff --git a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java
index 24a25b8a765..972ac5bfc9b 100644
--- a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java
+++ b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java
@@ -46,6 +46,6 @@ public final class Column {
@Override
public String toString() {
- return String.format("%s=%s", name, value);
+ return String.format("%s: oldValue=%s, value=%s", name, oldValue, value);
}
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/DataSourceImporter.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/DataSourceImporter.java
index 7e94bfc009f..163b4fd39e8 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/DataSourceImporter.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/DataSourceImporter.java
@@ -95,7 +95,7 @@ public final class DataSourceImporter extends AbstractLifecycleExecutor implemen
@Override
protected void runBlocking() {
- int batchSize = importerConfig.getBatchSize() * 2;
+ int batchSize = importerConfig.getBatchSize();
while (isRunning()) {
List<Record> records = channel.fetchRecords(batchSize, 3, TimeUnit.SECONDS);
if (null != records && !records.isEmpty()) {
@@ -246,8 +246,14 @@ public final class DataSourceImporter extends AbstractLifecycleExecutor implemen
}
for (int i = 0; i < conditionColumns.size(); i++) {
Column keyColumn = conditionColumns.get(i);
- preparedStatement.setObject(updatedColumns.size() + i + 1, keyColumn.isUniqueKey() && keyColumn.isUpdated() ? keyColumn.getOldValue() : keyColumn.getValue());
+ // TODO There to be compatible with PostgreSQL before value is null except primary key and unsupported updating sharding value now.
+ if (shardingColumns.contains(keyColumn.getName()) && keyColumn.getOldValue() == null) {
+ preparedStatement.setObject(updatedColumns.size() + i + 1, keyColumn.getValue());
+ continue;
+ }
+ preparedStatement.setObject(updatedColumns.size() + i + 1, keyColumn.getOldValue());
}
+ // TODO if table without unique key the conditionColumns before values is null, so update will fail at PostgreSQL
int updateCount = preparedStatement.executeUpdate();
if (1 != updateCount) {
log.warn("executeUpdate failed, updateCount={}, updateSql={}, updatedColumns={}, conditionColumns={}", updateCount, updateSql, updatedColumns, conditionColumns);
@@ -276,7 +282,7 @@ public final class DataSourceImporter extends AbstractLifecycleExecutor implemen
}
int[] counts = preparedStatement.executeBatch();
if (IntStream.of(counts).anyMatch(value -> 1 != value)) {
- log.warn("batchDelete failed, counts={}, sql={}", Arrays.toString(counts), deleteSQL);
+ log.warn("batchDelete failed, counts={}, sql={}, conditionColumns={}", Arrays.toString(counts), deleteSQL, conditionColumns);
}
} finally {
batchDeleteStatement = null;
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/record/RecordUtils.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/record/RecordUtils.java
index 816747a839f..68336abd7c9 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/record/RecordUtils.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/record/RecordUtils.java
@@ -26,6 +26,7 @@ import org.apache.shardingsphere.data.pipeline.api.ingest.record.Record;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
/**
@@ -64,7 +65,11 @@ public final class RecordUtils {
result.add(each);
}
}
- return result;
+ Optional<Column> uniqueKeyColumn = dataRecord.getColumns().stream().filter(Column::isUniqueKey).findFirst();
+ if (uniqueKeyColumn.isPresent()) {
+ return result;
+ }
+ return dataRecord.getColumns();
}
/**
@@ -84,11 +89,11 @@ public final class RecordUtils {
}
/**
- * Get last normal record.
- *
- * @param records records
- * @return last normal record.
- */
+ * Get last normal record.
+ *
+ * @param records records
+ * @return last normal record.
+ */
public static Record getLastNormalRecord(final List<Record> records) {
for (int index = records.size() - 1; index >= 0; index--) {
Record record = records.get(index);
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
index a3318855aa4..56bb8588b6a 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/AbstractPipelineSQLBuilder.java
@@ -179,7 +179,7 @@ public abstract class AbstractPipelineSQLBuilder implements PipelineSQLBuilder {
private String buildWhereSQL(final Collection<Column> conditionColumns) {
StringBuilder where = new StringBuilder();
for (Column each : conditionColumns) {
- where.append(String.format("%s = ? and ", quote(each.getName())));
+ where.append(String.format("%s = ? AND ", quote(each.getName())));
}
where.setLength(where.length() - 5);
return where.toString();
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/PipelineSQLBuilderTest.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/PipelineSQLBuilderTest.java
index 823712643b1..7481a806933 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/PipelineSQLBuilderTest.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/PipelineSQLBuilderTest.java
@@ -91,7 +91,7 @@ class PipelineSQLBuilderTest {
void assertBuildUpdateSQLWithShardingColumns() {
DataRecord dataRecord = mockDataRecord("t2");
String actual = pipelineSQLBuilder.buildUpdateSQL(null, dataRecord, mockConditionColumns(dataRecord));
- assertThat(actual, is("UPDATE t2 SET c1 = ?,c2 = ?,c3 = ? WHERE id = ? and sc = ?"));
+ assertThat(actual, is("UPDATE t2 SET c1 = ?,c2 = ?,c3 = ? WHERE id = ? AND sc = ?"));
}
@Test
@@ -104,7 +104,7 @@ class PipelineSQLBuilderTest {
void assertBuildDeleteSQLWithConditionColumns() {
DataRecord dataRecord = mockDataRecord("t3");
String actual = pipelineSQLBuilder.buildDeleteSQL(null, dataRecord, mockConditionColumns(dataRecord));
- assertThat(actual, is("DELETE FROM t3 WHERE id = ? and sc = ?"));
+ assertThat(actual, is("DELETE FROM t3 WHERE id = ? AND sc = ?"));
}
private Collection<Column> mockConditionColumns(final DataRecord dataRecord) {
@@ -121,4 +121,26 @@ class PipelineSQLBuilderTest {
result.addColumn(new Column("c3", "", true, false));
return result;
}
+
+ @Test
+ void assertBuildDeleteSQLWithoutUniqueKey() {
+ String actual = pipelineSQLBuilder.buildDeleteSQL(null, mockDataRecordWithoutUniqueKey("t_order"),
+ RecordUtils.extractConditionColumns(mockDataRecordWithoutUniqueKey("t_order"), Collections.emptySet()));
+ assertThat(actual, is("DELETE FROM t_order WHERE id = ? AND name = ?"));
+ }
+
+ @Test
+ void assertBuildUpdateSQLWithoutShardingColumns() {
+ DataRecord dataRecord = mockDataRecordWithoutUniqueKey("t_order");
+ String actual = pipelineSQLBuilder.buildUpdateSQL(null, dataRecord, mockConditionColumns(dataRecord));
+ assertThat(actual, is("UPDATE t_order SET name = ? WHERE id = ? AND name = ?"));
+ }
+
+ private DataRecord mockDataRecordWithoutUniqueKey(final String tableName) {
+ DataRecord result = new DataRecord(new PlaceholderPosition(), 4);
+ result.setTableName(tableName);
+ result.addColumn(new Column("id", "", false, false));
+ result.addColumn(new Column("name", "", true, false));
+ return result;
+ }
}
diff --git a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
index b4fee8c9457..f139d9558a4 100644
--- a/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
+++ b/kernel/data-pipeline/scenario/migration/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
@@ -27,6 +27,7 @@ import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSource
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
import org.apache.shardingsphere.data.pipeline.api.job.progress.InventoryIncrementalJobItemProgress;
+import org.apache.shardingsphere.data.pipeline.api.job.progress.listener.PipelineJobProgressUpdatedParameter;
import org.apache.shardingsphere.data.pipeline.api.metadata.SchemaName;
import org.apache.shardingsphere.data.pipeline.api.metadata.SchemaTableName;
import org.apache.shardingsphere.data.pipeline.api.metadata.TableName;
@@ -83,6 +84,7 @@ public final class MigrationDataConsistencyChecker implements PipelineDataConsis
.forEach(dataNode -> sourceTableNames.add(DataNodeUtils.formatWithSchema(dataNode)))));
progressContext.setRecordsCount(getRecordsCount());
progressContext.getTableNames().addAll(sourceTableNames);
+ progressContext.onProgressUpdated(new PipelineJobProgressUpdatedParameter(0));
Map<String, DataConsistencyCheckResult> result = new LinkedHashMap<>();
PipelineDataSourceManager dataSourceManager = new DefaultPipelineDataSourceManager();
try {
diff --git a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/primarykey/IndexesMigrationE2EIT.java b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/primarykey/IndexesMigrationE2EIT.java
index daa8812b8d9..72d838da673 100644
--- a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/primarykey/IndexesMigrationE2EIT.java
+++ b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/primarykey/IndexesMigrationE2EIT.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.test.e2e.data.pipeline.cases.migration.primarykey;
+import lombok.SneakyThrows;
import org.apache.commons.codec.binary.Hex;
import org.apache.shardingsphere.data.pipeline.scenario.migration.MigrationJobType;
import org.apache.shardingsphere.infra.database.type.dialect.MySQLDatabaseType;
@@ -86,17 +87,33 @@ class IndexesMigrationE2EIT extends AbstractMigrationE2EIT {
return;
}
KeyGenerateAlgorithm keyGenerateAlgorithm = new UUIDKeyGenerateAlgorithm();
- Object uniqueKey = keyGenerateAlgorithm.generateKey();
- assertMigrationSuccess(containerComposer, sql, "user_id", keyGenerateAlgorithm, consistencyCheckAlgorithmType, () -> {
- insertOneOrder(containerComposer, uniqueKey);
- containerComposer.assertProxyOrderRecordExist("t_order", uniqueKey);
+ // TODO PostgreSQL update delete events not support if table without unique keys at increment task.
+ final Callable<Void> incrementalTaskFn = () -> {
+ Object orderId = keyGenerateAlgorithm.generateKey();
+ insertOneOrder(containerComposer, orderId);
+ if (containerComposer.getDatabaseType() instanceof MySQLDatabaseType) {
+ updateOneOrder(containerComposer, orderId, "updated");
+ deleteOneOrder(containerComposer, orderId, "updated");
+ insertOneOrder(containerComposer, keyGenerateAlgorithm.generateKey());
+ }
return null;
- });
+ };
+ assertMigrationSuccess(containerComposer, sql, "user_id", keyGenerateAlgorithm, consistencyCheckAlgorithmType, incrementalTaskFn);
}
}
+ @SneakyThrows
+ private void doCreateUpdateDelete(final PipelineContainerComposer containerComposer, final Object orderId) {
+ String updatedStatus = "updated" + System.currentTimeMillis();
+ insertOneOrder(containerComposer, orderId);
+ updateOneOrder(containerComposer, orderId, updatedStatus);
+ deleteOneOrder(containerComposer, orderId, updatedStatus);
+ }
+
private void insertOneOrder(final PipelineContainerComposer containerComposer, final Object uniqueKey) throws SQLException {
- try (PreparedStatement preparedStatement = containerComposer.getSourceDataSource().getConnection().prepareStatement("INSERT INTO t_order (order_id,user_id,status) VALUES (?,?,?)")) {
+ try (
+ Connection connection = containerComposer.getSourceDataSource().getConnection();
+ PreparedStatement preparedStatement = connection.prepareStatement("INSERT INTO t_order (order_id,user_id,status) VALUES (?,?,?)")) {
preparedStatement.setObject(1, uniqueKey);
preparedStatement.setObject(2, 1);
preparedStatement.setObject(3, "OK");
@@ -105,6 +122,33 @@ class IndexesMigrationE2EIT extends AbstractMigrationE2EIT {
}
}
+ private void updateOneOrder(final PipelineContainerComposer containerComposer, final Object uniqueKey, final String updatedStatus) throws SQLException {
+ try (
+ Connection connection = containerComposer.getSourceDataSource().getConnection();
+ PreparedStatement preparedStatement = connection
+ .prepareStatement("UPDATE t_order SET status=? WHERE order_id = ? AND user_id = ? AND status = ?")) {
+ preparedStatement.setObject(1, updatedStatus);
+ preparedStatement.setObject(2, uniqueKey);
+ preparedStatement.setObject(3, 1);
+ preparedStatement.setObject(4, "OK");
+ int actualCount = preparedStatement.executeUpdate();
+ assertThat(actualCount, is(1));
+ }
+ }
+
+ private void deleteOneOrder(final PipelineContainerComposer containerComposer, final Object uniqueKey, final String updatedStatus) throws SQLException {
+ try (
+ Connection connection = containerComposer.getSourceDataSource().getConnection();
+ PreparedStatement preparedStatement = connection
+ .prepareStatement("DELETE FROM t_order WHERE order_id = ? AND user_id = ? AND status = ?")) {
+ preparedStatement.setObject(1, uniqueKey);
+ preparedStatement.setObject(2, 1);
+ preparedStatement.setObject(3, updatedStatus);
+ int actualCount = preparedStatement.executeUpdate();
+ assertThat(actualCount, is(1));
+ }
+ }
+
@ParameterizedTest(name = "{0}")
@EnabledIf("isEnabled")
@ArgumentsSource(PipelineE2ETestCaseArgumentsProvider.class)
@@ -122,6 +166,7 @@ class IndexesMigrationE2EIT extends AbstractMigrationE2EIT {
Object uniqueKey = keyGenerateAlgorithm.generateKey();
assertMigrationSuccess(containerComposer, sql, "user_id", keyGenerateAlgorithm, consistencyCheckAlgorithmType, () -> {
insertOneOrder(containerComposer, uniqueKey);
+ doCreateUpdateDelete(containerComposer, keyGenerateAlgorithm.generateKey());
containerComposer.assertProxyOrderRecordExist("t_order", uniqueKey);
return null;
});
@@ -145,6 +190,7 @@ class IndexesMigrationE2EIT extends AbstractMigrationE2EIT {
Object uniqueKey = keyGenerateAlgorithm.generateKey();
assertMigrationSuccess(containerComposer, sql, "user_id", keyGenerateAlgorithm, consistencyCheckAlgorithmType, () -> {
insertOneOrder(containerComposer, uniqueKey);
+ doCreateUpdateDelete(containerComposer, keyGenerateAlgorithm.generateKey());
containerComposer.assertProxyOrderRecordExist("t_order", uniqueKey);
return null;
});
diff --git a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/E2EIncrementalTask.java b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/E2EIncrementalTask.java
index 19b9e2b09e9..0e211b4e7e6 100644
--- a/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/E2EIncrementalTask.java
+++ b/test/e2e/operation/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/E2EIncrementalTask.java
@@ -133,6 +133,7 @@ public final class E2EIncrementalTask extends BaseIncrementTask {
private void setNullToAllFields(final Object orderId) {
if (databaseType instanceof MySQLDatabaseType) {
String sql = SQLBuilderUtils.buildUpdateSQL(ignoreShardingColumns(MYSQL_COLUMN_NAMES), orderTableName, "null");
+ log.info("update sql: {}", sql);
DataSourceExecuteUtils.execute(dataSource, sql, new Object[]{orderId});
}
}
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/importer/DataSourceImporterTest.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/importer/DataSourceImporterTest.java
index dc0140be0d5..1f190e36cbc 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/importer/DataSourceImporterTest.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/importer/DataSourceImporterTest.java
@@ -122,7 +122,7 @@ class DataSourceImporterTest {
verify(preparedStatement).setObject(1, 20);
verify(preparedStatement).setObject(2, "UPDATE");
verify(preparedStatement).setObject(3, 1);
- verify(preparedStatement).setObject(4, 20);
+ verify(preparedStatement).setObject(4, 10);
verify(preparedStatement).executeUpdate();
}
@@ -137,7 +137,7 @@ class DataSourceImporterTest {
inOrder.verify(preparedStatement).setObject(2, 10);
inOrder.verify(preparedStatement).setObject(3, "UPDATE");
inOrder.verify(preparedStatement).setObject(4, 1);
- inOrder.verify(preparedStatement).setObject(5, 10);
+ inOrder.verify(preparedStatement).setObject(5, 0);
inOrder.verify(preparedStatement).executeUpdate();
}