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 442e07c Fix "Sharding value must implements Comparabl" during
incremental migration of PostgreSQL (#14399)
442e07c is described below
commit 442e07c2175e68e7d3e9622069e6ce697a25007d
Author: ReyYang <[email protected]>
AuthorDate: Wed Dec 29 15:36:33 2021 +0800
Fix "Sharding value must implements Comparabl" during incremental migration
of PostgreSQL (#14399)
---
.../pipeline/opengauss/ingest/OpenGaussWalDumper.java | 19 -------------------
.../postgresql/ingest/wal/WalEventConverter.java | 4 +++-
.../data/pipeline/api/ingest/record/Column.java | 14 +++-----------
3 files changed, 6 insertions(+), 31 deletions(-)
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWalDumper.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWalDumper.java
index 0731aab..504da0e 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWalDumper.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWalDumper.java
@@ -24,11 +24,8 @@ import
org.apache.shardingsphere.data.pipeline.api.datasource.config.impl.Standa
import
org.apache.shardingsphere.data.pipeline.api.executor.AbstractLifecycleExecutor;
import org.apache.shardingsphere.data.pipeline.api.ingest.channel.Channel;
import
org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPosition;
-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.Record;
import
org.apache.shardingsphere.data.pipeline.core.datasource.creator.PipelineDataSourceCreatorFactory;
-import
org.apache.shardingsphere.data.pipeline.core.ingest.IngestDataChangeType;
import
org.apache.shardingsphere.data.pipeline.core.ingest.exception.IngestException;
import org.apache.shardingsphere.data.pipeline.core.util.ThreadUtil;
import
org.apache.shardingsphere.data.pipeline.opengauss.ingest.wal.OpenGaussLogicalReplication;
@@ -122,7 +119,6 @@ public final class OpenGaussWalDumper extends
AbstractLifecycleExecutor implemen
if (!(event instanceof PlaceholderEvent) &&
log.isDebugEnabled()) {
log.debug("dump, event={}, record={}", event, record);
}
- updateRecordOldValue(record);
pushRecord(record);
}
} catch (final SQLException ex) {
@@ -133,21 +129,6 @@ public final class OpenGaussWalDumper extends
AbstractLifecycleExecutor implemen
}
}
- private void updateRecordOldValue(final Record record) {
- if (!(record instanceof DataRecord)) {
- return;
- }
- DataRecord dataRecord = (DataRecord) record;
- if (!IngestDataChangeType.UPDATE.equals(dataRecord.getType())) {
- return;
- }
- for (Column each: dataRecord.getColumns()) {
- if (each.isPrimaryKey() && each.isUpdated()) {
- each.setOldValue(each.getValue());
- }
- }
- }
-
private void pushRecord(final Record record) {
try {
channel.pushRecord(record);
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/wal/WalEventConverter.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/wal/WalEventConverter.java
index befe2aa..38c3155 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/wal/WalEventConverter.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/wal/WalEventConverter.java
@@ -125,7 +125,9 @@ public final class WalEventConverter {
private void putColumnsIntoDataRecord(final DataRecord dataRecord, final
TableMetaData tableMetaData, final List<Object> values) {
for (int i = 0; i < values.size(); i++) {
- dataRecord.addColumn(new
Column(tableMetaData.getColumnMetaData(i).getName(), values.get(i), true,
tableMetaData.isPrimaryKey(i)));
+ Object primaryKeyOldValue = tableMetaData.isPrimaryKey(i) ?
values.get(i) : null;
+ Column column = new
Column(tableMetaData.getColumnMetaData(i).getName(), primaryKeyOldValue,
values.get(i), true, tableMetaData.isPrimaryKey(i));
+ dataRecord.addColumn(column);
}
}
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java
index 5c0cc17..b1417b4 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/ingest/record/Column.java
@@ -18,11 +18,12 @@
package org.apache.shardingsphere.data.pipeline.api.ingest.record;
import lombok.Getter;
-import lombok.Setter;
+import lombok.RequiredArgsConstructor;
/**
* Column.
*/
+@RequiredArgsConstructor
@Getter
public final class Column {
@@ -31,8 +32,7 @@ public final class Column {
/**
* Value are available only when the primary key column is updated.
*/
- @Setter
- private Object oldValue;
+ private final Object oldValue;
private final Object value;
@@ -44,14 +44,6 @@ public final class Column {
this(name, null, value, updated, primaryKey);
}
- public Column(final String name, final Object oldValue, final Object
value, final boolean updated, final boolean primaryKey) {
- this.name = name;
- this.oldValue = oldValue;
- this.value = value;
- this.updated = updated;
- this.primaryKey = primaryKey;
- }
-
@Override
public String toString() {
return String.format("%s=%s", name, value);
