This is an automated email from the ASF dual-hosted git repository.
terrymanu 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 5cfc82b43df Address Sonar findings in pipeline modules (#40345)
5cfc82b43df is described below
commit 5cfc82b43df42893e713b0abfcca18ac0a10adb6
Author: Liang Zhang <[email protected]>
AuthorDate: Thu Oct 22 20:30:41 2026 +0800
Address Sonar findings in pipeline modules (#40345)
---
.../column/InventoryColumnValueReaderEngine.java | 26 +++++++------
.../AbstractRecordTableInventoryCalculator.java | 10 +++--
.../metadata/generator/PipelineDDLDecorator.java | 3 +-
.../segment/PipelineSQLSegmentBuilder.java | 4 +-
.../incremental/client/MySQLBinlogClient.java | 12 +++---
.../netty/MySQLBinlogEventPacketDecoder.java | 15 +++-----
.../incremental/dumper/MySQLIncrementalDumper.java | 12 +++---
.../incremental/client/MySQLBinlogClientTest.java | 45 +++++++++++++++++++++-
.../wal/decode/MppdbDecodingPlugin.java | 28 +++-----------
.../dumper/PostgreSQLIncrementalDumper.java | 4 +-
.../ingest/incremental/wal/WALEventConverter.java | 12 +++---
.../incremental/wal/decode/TestDecodingPlugin.java | 3 +-
.../ddl/PostgreSQLDDLTemplateExecutor.java | 7 ++--
.../column/PostgreSQLColumnPropertiesAppender.java | 18 ++++-----
.../PostgreSQLConstraintsPropertiesAppender.java | 13 ++++---
.../ddl/index/PostgreSQLIndexSQLGenerator.java | 3 +-
.../ddl/table/PostgreSQLTablePropertiesLoader.java | 2 +-
.../RecordTableInventoryCheckCalculatorTest.java | 25 ++++++++++++
.../InventoryColumnValueReaderEngineTest.java | 6 +++
19 files changed, 152 insertions(+), 96 deletions(-)
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngine.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngine.java
index 8e67cb02d52..ba2968c7b54 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngine.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngine.java
@@ -20,6 +20,7 @@ package
org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.col
import lombok.RequiredArgsConstructor;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import org.jspecify.annotations.Nullable;
import java.math.BigDecimal;
import java.sql.Blob;
@@ -67,28 +68,20 @@ public final class InventoryColumnValueReaderEngine {
case Types.TINYINT:
if (isSigned(metaData, columnIndex)) {
return resultSet.getByte(columnIndex);
- } else {
- return resultSet.getShort(columnIndex);
}
+ return resultSet.getShort(columnIndex);
case Types.SMALLINT:
if (isSigned(metaData, columnIndex)) {
return resultSet.getShort(columnIndex);
- } else {
- return resultSet.getInt(columnIndex);
}
+ return resultSet.getInt(columnIndex);
case Types.INTEGER:
if (isSigned(metaData, columnIndex)) {
return resultSet.getInt(columnIndex);
- } else {
- return resultSet.getLong(columnIndex);
}
+ return resultSet.getLong(columnIndex);
case Types.BIGINT:
- if (isSigned(metaData, columnIndex)) {
- return resultSet.getLong(columnIndex);
- } else {
- BigDecimal decimal = resultSet.getBigDecimal(columnIndex);
- return null == decimal ? null : decimal.toBigInteger();
- }
+ return readBigIntValue(resultSet, metaData, columnIndex);
case Types.NUMERIC:
case Types.DECIMAL:
return resultSet.getBigDecimal(columnIndex);
@@ -133,4 +126,13 @@ public final class InventoryColumnValueReaderEngine {
private boolean isSigned(final ResultSetMetaData metaData, final int
columnIndex) throws SQLException {
return metaData.isSigned(columnIndex);
}
+
+ @Nullable
+ private Object readBigIntValue(final ResultSet resultSet, final
ResultSetMetaData metaData, final int columnIndex) throws SQLException {
+ if (isSigned(metaData, columnIndex)) {
+ return resultSet.getLong(columnIndex);
+ }
+ BigDecimal result = resultSet.getBigDecimal(columnIndex);
+ return null == result ? null : result.toBigInteger();
+ }
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/query/calculator/AbstractRecordTableInventoryCalculator.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/query/calculator/AbstractRecordTableInventoryCalculator.java
index 6133e6f4eaa..e2e57bfbf6e 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/query/calculator/AbstractRecordTableInventoryCalculator.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/query/calculator/AbstractRecordTableInventoryCalculator.java
@@ -281,7 +281,7 @@ public abstract class
AbstractRecordTableInventoryCalculator<S, C> extends Abstr
private void setParameters(final PreparedStatement preparedStatement,
final TableInventoryCalculateParameter param) throws SQLException {
QueryType queryType = param.getQueryType();
- if (queryType == QueryType.RANGE_QUERY) {
+ if (QueryType.RANGE_QUERY == queryType) {
Range<?> range = param.getRange();
ShardingSpherePreconditions.checkNotNull(range,
() -> new
PipelineTableDataConsistencyCheckLoadingFailedException(param.getTable(), new
RuntimeException("Unique keys values range is null.")));
@@ -295,7 +295,9 @@ public abstract class
AbstractRecordTableInventoryCalculator<S, C> extends Abstr
if (StreamingRangeType.SMALL == streamingRangeType) {
setParameter(preparedStatement, parameterIndex, chunkSize *
streamingChunkCount);
}
- } else if (queryType == QueryType.POINT_QUERY) {
+ return;
+ }
+ if (QueryType.POINT_QUERY == queryType) {
Collection<Object> uniqueKeysValues = param.getUniqueKeysValues();
ShardingSpherePreconditions.checkNotNull(uniqueKeysValues,
() -> new
PipelineTableDataConsistencyCheckLoadingFailedException(param.getTable(), new
RuntimeException("Unique keys values is null.")));
@@ -311,9 +313,9 @@ public abstract class
AbstractRecordTableInventoryCalculator<S, C> extends Abstr
setParameter(preparedStatement, parameterIndex++, each);
}
}
- } else {
- throw new UnsupportedOperationException("Query type: " +
queryType);
+ return;
}
+ throw new UnsupportedOperationException("Query type: " + queryType);
}
private void setParameter(final PreparedStatement preparedStatement, final
int parameterIndex, final Object value) throws SQLException {
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLDecorator.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLDecorator.java
index a5101fdfbfa..fe3c39271d5 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLDecorator.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/generator/PipelineDDLDecorator.java
@@ -185,8 +185,7 @@ public final class PipelineDDLDecorator {
|| sqlStatement instanceof AlterTableStatement || sqlStatement
instanceof CommentStatement) {
TableExtractor tableExtractor = new TableExtractor();
tableExtractor.extractTablesFromSQLStatement(sqlStatement);
- Optional<SimpleTableSegment> simpleTableSegment =
tableExtractor.getTableContext().stream()
- .filter(each -> each instanceof
SimpleTableSegment).map(each -> (SimpleTableSegment) each).findFirst();
+ Optional<SimpleTableSegment> simpleTableSegment =
tableExtractor.getTableContext().stream().filter(SimpleTableSegment.class::isInstance).map(SimpleTableSegment.class::cast).findFirst();
if (!simpleTableSegment.isPresent()) {
return sql;
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
index 86c5364188f..4dd5ab4608a 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/sqlbuilder/segment/PipelineSQLSegmentBuilder.java
@@ -28,7 +28,7 @@ import
org.apache.shardingsphere.infra.metadata.identifier.DatabaseIdentifierCon
import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
import org.jspecify.annotations.Nullable;
-import java.util.function.Function;
+import java.util.function.UnaryOperator;
/**
* Pipeline SQL segment builder.
@@ -131,7 +131,7 @@ public final class PipelineSQLSegmentBuilder {
return buildQualifiedTableName(schemaName, tableName,
this::getEscapedActualIdentifier);
}
- private String buildQualifiedTableName(final String schemaName, final
String tableName, final Function<String, String> identifierEscaper) {
+ private String buildQualifiedTableName(final String schemaName, final
String tableName, final UnaryOperator<String> identifierEscaper) {
StringBuilder result = new StringBuilder();
if (dialectDatabaseMetaData.getSchemaOption().isSchemaAvailable() &&
!Strings.isNullOrEmpty(schemaName)) {
result.append(identifierEscaper.apply(schemaName)).append('.');
diff --git
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
index ff7dd63f5b7..258fc777da0 100644
---
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
+++
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
@@ -33,6 +33,7 @@ import io.netty.util.concurrent.Promise;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ThreadUtils;
import
org.apache.shardingsphere.data.pipeline.core.exception.PipelineInternalException;
import
org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.MySQLBaseBinlogEvent;
import
org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.PlaceholderBinlogEvent;
@@ -55,6 +56,7 @@ import org.apache.shardingsphere.infra.util.json.JsonEngine;
import org.apache.shardingsphere.proxy.frontend.netty.ChannelAttrInitializer;
import java.net.InetSocketAddress;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
@@ -346,8 +348,8 @@ public final class MySQLBinlogClient {
blockingEventQueue.put(records);
return;
}
- if (msg instanceof MySQLBaseBinlogEvent) {
- lastBinlogEvent.set((MySQLBaseBinlogEvent) msg);
+ if (msg instanceof MySQLBaseBinlogEvent binlogEvent) {
+ lastBinlogEvent.set(binlogEvent);
blockingEventQueue.put(Collections.singletonList(lastBinlogEvent.get()));
}
}
@@ -375,14 +377,14 @@ public final class MySQLBinlogClient {
}
@SneakyThrows(InterruptedException.class)
- private synchronized void reconnect() {
+ private void reconnect() {
log.info("Reconnect failure count: {}",
continuousFailureCount.get());
if (continuousFailureCount.get() >= 5) {
log.error("MySQL binlog client failed permanently,
lastBinlogEvent: {}", JsonEngine.marshal(lastBinlogEvent.get()));
closeChannel(true);
return;
}
- wait(2000L * (continuousFailureCount.get() + 1));
+ ThreadUtils.sleep(Duration.ofMillis(2000L *
(continuousFailureCount.get() + 1)));
try {
reconnectWithRetry();
// CHECKSTYLE:OFF
@@ -407,7 +409,7 @@ public final class MySQLBinlogClient {
if (reconnectTimes > 3) {
throw ex;
}
- wait(1000L << reconnectTimes);
+ ThreadUtils.sleep(Duration.ofMillis(1000L <<
reconnectTimes));
}
}
}
diff --git
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/netty/MySQLBinlogEventPacketDecoder.java
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/netty/MySQLBinlogEventPacketDecoder.java
index e86cf171e28..b52daa5499d 100644
---
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/netty/MySQLBinlogEventPacketDecoder.java
+++
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/netty/MySQLBinlogEventPacketDecoder.java
@@ -120,8 +120,7 @@ public final class MySQLBinlogEventPacketDecoder extends
ByteToMessageDecoder {
}
private void processEventWithTX(final MySQLBaseBinlogEvent binlogEvent,
final List<Object> out) {
- if (binlogEvent instanceof MySQLQueryBinlogEvent) {
- MySQLQueryBinlogEvent queryEvent = (MySQLQueryBinlogEvent)
binlogEvent;
+ if (binlogEvent instanceof MySQLQueryBinlogEvent queryEvent) {
if (TX_BEGIN_SQL.equals(queryEvent.getSql())) {
records = new LinkedList<>();
} else {
@@ -136,8 +135,7 @@ public final class MySQLBinlogEventPacketDecoder extends
ByteToMessageDecoder {
}
private void processEventIgnoreTX(final MySQLBaseBinlogEvent binlogEvent,
final List<Object> out) {
- if (binlogEvent instanceof MySQLQueryBinlogEvent) {
- MySQLQueryBinlogEvent queryEvent = (MySQLQueryBinlogEvent)
binlogEvent;
+ if (binlogEvent instanceof MySQLQueryBinlogEvent queryEvent) {
if (TX_BEGIN_SQL.equals(queryEvent.getSql())) {
return;
}
@@ -156,14 +154,11 @@ public final class MySQLBinlogEventPacketDecoder extends
ByteToMessageDecoder {
case TABLE_MAP_EVENT:
decodeTableMapEvent(binlogEventHeader, payload);
return Optional.empty();
- case WRITE_ROWS_EVENT_V1:
- case WRITE_ROWS_EVENT_V2:
+ case WRITE_ROWS_EVENT_V1, WRITE_ROWS_EVENT_V2:
return Optional.of(decodeWriteRowsEventV2(binlogEventHeader,
payload));
- case UPDATE_ROWS_EVENT_V1:
- case UPDATE_ROWS_EVENT_V2:
+ case UPDATE_ROWS_EVENT_V1, UPDATE_ROWS_EVENT_V2:
return Optional.of(decodeUpdateRowsEventV2(binlogEventHeader,
payload));
- case DELETE_ROWS_EVENT_V1:
- case DELETE_ROWS_EVENT_V2:
+ case DELETE_ROWS_EVENT_V1, DELETE_ROWS_EVENT_V2:
return Optional.of(decodeDeleteRowsEventV2(binlogEventHeader,
payload));
case QUERY_EVENT:
return Optional.of(decodeQueryEvent(binlogEventHeader,
payload));
diff --git
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/dumper/MySQLIncrementalDumper.java
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/dumper/MySQLIncrementalDumper.java
index 27c64213cc2..d07ac20fdd6 100644
---
a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/dumper/MySQLIncrementalDumper.java
+++
b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/dumper/MySQLIncrementalDumper.java
@@ -126,14 +126,14 @@ public final class MySQLIncrementalDumper extends
AbstractPipelineLifecycleRunna
return Collections.singletonList(createPlaceholderRecord(event));
}
PipelineTableMetaData tableMetaData =
getPipelineTableMetaData(rowsEvent.getTableName());
- if (event instanceof MySQLWriteRowsBinlogEvent) {
- return handleWriteRowsEvent((MySQLWriteRowsBinlogEvent) event,
tableMetaData);
+ if (event instanceof MySQLWriteRowsBinlogEvent writeRowsEvent) {
+ return handleWriteRowsEvent(writeRowsEvent, tableMetaData);
}
- if (event instanceof MySQLUpdateRowsBinlogEvent) {
- return handleUpdateRowsEvent((MySQLUpdateRowsBinlogEvent) event,
tableMetaData);
+ if (event instanceof MySQLUpdateRowsBinlogEvent updateRowsEvent) {
+ return handleUpdateRowsEvent(updateRowsEvent, tableMetaData);
}
- if (event instanceof MySQLDeleteRowsBinlogEvent) {
- return handleDeleteRowsEvent((MySQLDeleteRowsBinlogEvent) event,
tableMetaData);
+ if (event instanceof MySQLDeleteRowsBinlogEvent deleteRowsEvent) {
+ return handleDeleteRowsEvent(deleteRowsEvent, tableMetaData);
}
return Collections.emptyList();
}
diff --git
a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java
b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java
index 794dfa1f574..12ce76ef50c 100644
---
a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java
+++
b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java
@@ -32,6 +32,7 @@ import io.netty.util.Attribute;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.SneakyThrows;
+import org.apache.commons.lang3.ThreadUtils;
import
org.apache.shardingsphere.data.pipeline.core.exception.PipelineInternalException;
import
org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.MySQLBinlogContext;
import
org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.MySQLBaseBinlogEvent;
@@ -48,6 +49,8 @@ import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperation
import
org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureAssertion;
import
org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureExtension;
import
org.apache.shardingsphere.test.infra.framework.extension.log.LogCaptureSettings;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -69,6 +72,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -88,7 +92,9 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.times;
@@ -97,9 +103,10 @@ import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
@SuppressWarnings("ProhibitedExceptionDeclared")
-@ExtendWith({MockitoExtension.class, LogCaptureExtension.class})
+@ExtendWith({MockitoExtension.class, LogCaptureExtension.class,
AutoMockExtension.class})
@LogCaptureSettings(suppressOutput = true)
@MockitoSettings(strictness = Strictness.LENIENT)
+@StaticMockSettings({CompletableFuture.class, ThreadUtils.class})
class MySQLBinlogClientTest {
@Mock
@@ -316,6 +323,42 @@ class MySQLBinlogClientTest {
assertThat(getChecksumLength(decoder), is(0));
}
+ @Test
+ void assertReconnectAfterTransientFailure() throws Exception {
+ client = createClientMock();
+ prepareClientChannel();
+ setServerVersion("5.5.0");
+ doReturn(true).when(client).execute(anyString());
+
when(channel.writeAndFlush(any(MySQLComRegisterSlaveCommandPacket.class))).thenAnswer(invocation
-> {
+ Promise<Object> callback = new
DefaultPromise<>(eventLoopGroup.next());
+ callback.setSuccess(new MySQLOKPacket(0));
+ setResponseCallback(callback);
+ return null;
+ });
+ client.subscribe("binlog-000004", 16L);
+ ArgumentCaptor<ChannelHandler> handlerCaptor =
ArgumentCaptor.forClass(ChannelHandler.class);
+ verify(pipeline, times(2)).addLast(handlerCaptor.capture());
+ ChannelInboundHandlerAdapter handler =
handlerCaptor.getAllValues().stream().filter(each ->
"MySQLBinlogEventHandler".equals(each.getClass().getSimpleName()))
+ .map(each -> (ChannelInboundHandlerAdapter)
each).findFirst().orElseThrow(IllegalStateException::new);
+ ChannelHandlerContext context = mock(ChannelHandlerContext.class);
+ handler.channelRead(context, new
PlaceholderBinlogEvent("binlog-000005", 32L, 1L));
+ doThrow(new PipelineInternalException("connect
failed")).doNothing().when(client).connect();
+ doNothing().when(client).subscribe("binlog-000005", 32L);
+ CompletableFuture<Void> reconnectFuture = new CompletableFuture<>();
+ ArgumentCaptor<Runnable> reconnectCaptor =
ArgumentCaptor.forClass(Runnable.class);
+
when(CompletableFuture.runAsync(reconnectCaptor.capture())).thenReturn(reconnectFuture);
+ handler.channelInactive(context);
+ handler.channelInactive(context);
+ assertThat(reconnectCaptor.getAllValues().size(), is(1));
+ reconnectCaptor.getValue().run();
+ verify(client, times(2)).connect();
+ verify(channel).close();
+ verify(client).subscribe("binlog-000005", 32L);
+ reconnectFuture.complete(null);
+ handler.channelInactive(context);
+ assertThat(reconnectCaptor.getAllValues().size(), is(2));
+ }
+
@Test
void assertPollOnNotRunning() throws ReflectiveOperationException {
Plugins.getMemberAccessor().set(MySQLBinlogClient.class.getDeclaredField("channel"),
client, channel);
diff --git
a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/incremental/wal/decode/MppdbDecodingPlugin.java
b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/incremental/wal/decode/MppdbDecodingPlugin.java
index e7f70ed29d9..d9456f741c1 100644
---
a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/incremental/wal/decode/MppdbDecodingPlugin.java
+++
b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/incremental/wal/decode/MppdbDecodingPlugin.java
@@ -192,9 +192,7 @@ public final class MppdbDecodingPlugin implements
DecodingPlugin {
return decodeString(data.substring(1));
}
switch (columnType) {
- case "tinyint":
- case "smallint":
- case "integer":
+ case "tinyint", "smallint", "integer":
return Integer.parseInt(data);
case "bigint":
return Long.parseLong(data);
@@ -204,8 +202,7 @@ public final class MppdbDecodingPlugin implements
DecodingPlugin {
return Double.parseDouble(data);
case "boolean":
return Boolean.parseBoolean(data);
- case "time without time zone":
- case "time with time zone":
+ case "time without time zone", "time with time zone":
try {
return timestampUtils.toTime(null, decodeString(data));
} catch (final SQLException ex) {
@@ -213,35 +210,20 @@ public final class MppdbDecodingPlugin implements
DecodingPlugin {
}
case "date":
return Date.valueOf(decodeString(data));
- case "timestamp without time zone":
- case "timestamp with time zone":
- case "smalldatetime":
+ case "timestamp without time zone", "timestamp with time zone",
"smalldatetime":
try {
return timestampUtils.toTimestamp(null,
decodeString(data));
} catch (final SQLException ex) {
throw new DecodingException(ex);
}
- case "bytea":
- case "blob":
+ case "bytea", "blob":
return decodeBytea(data);
- case "raw":
- case "reltime":
- case "int4range":
- case "int8range":
- case "numrange":
- case "tsrange":
- case "tstzrange":
- case "daterange":
+ case "raw", "reltime", "int4range", "int8range", "numrange",
"tsrange", "tstzrange", "daterange":
return decodePgObject(data, columnType);
case "money":
return decodeMoney(data);
case "interval":
return decodeInterval(data);
- case "character varying":
- case "text":
- case "character":
- case "nvarchar2":
- case "tsquery":
default:
return decodeString(data).replace("''", "'");
}
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/dumper/PostgreSQLIncrementalDumper.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/dumper/PostgreSQLIncrementalDumper.java
index 8ae7fac21ca..375ceccf8ea 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/dumper/PostgreSQLIncrementalDumper.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/dumper/PostgreSQLIncrementalDumper.java
@@ -149,8 +149,8 @@ public final class PostgreSQLIncrementalDumper extends
AbstractPipelineLifecycle
channel.push(records);
return;
}
- if (event instanceof AbstractRowEvent) {
- rowEvents.add((AbstractRowEvent) event);
+ if (event instanceof AbstractRowEvent rowEvent) {
+ rowEvents.add(rowEvent);
}
}
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
index 4f7209fa27a..33edff7bd13 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/WALEventConverter.java
@@ -110,14 +110,14 @@ public final class WALEventConverter {
return createPlaceholderRecord(event);
}
PipelineTableMetaData tableMetaData =
metaDataLoader.getTableMetaData(expectedSchemaName, actualTableName);
- if (event instanceof WriteRowEvent) {
- return handleWriteRowEvent((WriteRowEvent) event, tableMetaData,
logicTableName);
+ if (event instanceof WriteRowEvent writeRowEvent) {
+ return handleWriteRowEvent(writeRowEvent, tableMetaData,
logicTableName);
}
- if (event instanceof UpdateRowEvent) {
- return handleUpdateRowEvent((UpdateRowEvent) event, tableMetaData,
logicTableName);
+ if (event instanceof UpdateRowEvent updateRowEvent) {
+ return handleUpdateRowEvent(updateRowEvent, tableMetaData,
logicTableName);
}
- if (event instanceof DeleteRowEvent) {
- return handleDeleteRowEvent((DeleteRowEvent) event, tableMetaData,
logicTableName);
+ if (event instanceof DeleteRowEvent deleteRowEvent) {
+ return handleDeleteRowEvent(deleteRowEvent, tableMetaData,
logicTableName);
}
throw new UnsupportedSQLOperationException("");
}
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/decode/TestDecodingPlugin.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/decode/TestDecodingPlugin.java
index 0769caaeba0..e687f0496e6 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/decode/TestDecodingPlugin.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/incremental/wal/decode/TestDecodingPlugin.java
@@ -207,8 +207,7 @@ public final class TestDecodingPlugin implements
DecodingPlugin {
}
case "bytea":
return decodeHex(readNextString(data).substring(2));
- case "json":
- case "jsonb":
+ case "json", "jsonb":
return readNextJson(data);
default:
return readNextString(data);
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/PostgreSQLDDLTemplateExecutor.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/PostgreSQLDDLTemplateExecutor.java
index b5455416df0..de5cc75e29c 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/PostgreSQLDDLTemplateExecutor.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/PostgreSQLDDLTemplateExecutor.java
@@ -33,7 +33,6 @@ import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.Map;
-import java.util.stream.Collectors;
/**
* Abstract ddl adapter for PostgreSQL.
@@ -73,7 +72,7 @@ public final class PostgreSQLDDLTemplateExecutor {
Collection<Map<String, Object>> result = new LinkedList<>();
while (resultSet.next()) {
int columnCount = metaData.getColumnCount();
- Map<String, Object> row = new LinkedHashMap<>(columnCount);
+ Map<String, Object> row =
LinkedHashMap.newLinkedHashMap(columnCount);
for (int i = 1; i <= columnCount; i++) {
row.put(metaData.getColumnName(i), resultSet.getObject(i));
}
@@ -103,7 +102,7 @@ public final class PostgreSQLDDLTemplateExecutor {
private Map<String, Object> getSingleRow(final ResultSet resultSet) throws
SQLException {
ResultSetMetaData metaData = resultSet.getMetaData();
int columnCount = metaData.getColumnCount();
- Map<String, Object> result = new LinkedHashMap<>(columnCount);
+ Map<String, Object> result =
LinkedHashMap.newLinkedHashMap(columnCount);
if (resultSet.next()) {
for (int i = 1; i <= columnCount; i++) {
result.put(metaData.getColumnName(i), resultSet.getObject(i));
@@ -123,7 +122,7 @@ public final class PostgreSQLDDLTemplateExecutor {
return;
}
Collection<Map<String, String>> formatLabels = new LinkedList<>();
- Collection<String> securityLabels = Arrays.stream((String[]) ((Array)
data.get("seclabels")).getArray()).collect(Collectors.toList());
+ Collection<String> securityLabels = Arrays.stream((String[]) ((Array)
data.get("seclabels")).getArray()).toList();
for (String each : securityLabels) {
Map<String, String> securityLabel = new LinkedHashMap<>(2, 1F);
securityLabel.put("provider", each.substring(0,
each.indexOf(SECURITY_LABEL_SPLIT)));
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/column/PostgreSQLColumnPropertiesAppender.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/column/PostgreSQLColumnPropertiesAppender.java
index c2a35b58a0a..1d63edfeb05 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/column/PostgreSQLColumnPropertiesAppender.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/column/PostgreSQLColumnPropertiesAppender.java
@@ -23,6 +23,7 @@ import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.Postgre
import java.sql.Array;
import java.sql.Connection;
import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -50,9 +51,9 @@ public final class PostgreSQLColumnPropertiesAppender {
private static final Pattern BRACKETS_PATTERN =
Pattern.compile("(\\(\\d+\\))");
- private static final Pattern SIGNED_NUMBER_WITH_GROUPING_PATTERN =
Pattern.compile("^([+-])?\\s*[0-9][0-9,\\s]*$");
+ private static final Pattern SIGNED_NUMBER_WITH_GROUPING_PATTERN =
Pattern.compile("^([+-])?\\s*\\d[\\d,\\s]*$");
- private static final Pattern NON_DIGIT_PATTERN = Pattern.compile("[^0-9]");
+ private static final Pattern NON_DIGIT_PATTERN = Pattern.compile("\\D");
private static final String ATT_OPTION_SPLIT = "=";
@@ -106,7 +107,7 @@ public final class PostgreSQLColumnPropertiesAppender {
}
private Collection<String> toCollection(final Array array) throws
SQLException {
- return Arrays.stream((String[])
array.getArray()).collect(Collectors.toList());
+ return Arrays.stream((String[])
array.getArray()).collect(Collectors.toCollection(ArrayList::new));
}
private Collection<Map<String, Object>> getColumnFromInherits(final
Collection<String> collInherits) {
@@ -138,7 +139,7 @@ public final class PostgreSQLColumnPropertiesAppender {
}
private Collection<String> toCollectionAndSort(final Array editTypes)
throws SQLException {
- return Arrays.stream((String[])
editTypes.getArray()).sorted(String::compareTo).collect(Collectors.toList());
+ return Arrays.stream((String[])
editTypes.getArray()).sorted(String::compareTo).collect(Collectors.toCollection(ArrayList::new));
}
private void columnFormatter(final Map<String, Object> column, final
Collection<String> editTypes) throws SQLException {
@@ -148,7 +149,7 @@ public final class PostgreSQLColumnPropertiesAppender {
formatColumnVariables(column);
templateExecutor.formatSecurityLabels(column);
editTypes.add(column.get("cltype").toString());
- column.put("edit_types",
editTypes.stream().sorted().collect(Collectors.toList()));
+ column.put("edit_types",
editTypes.stream().sorted().collect(Collectors.toCollection(ArrayList::new)));
column.put("cltype", parseTypeName(column.get("cltype").toString()));
}
@@ -192,7 +193,7 @@ public final class PostgreSQLColumnPropertiesAppender {
if (null == column.get("attnum") || null == column.get("indkey")) {
return;
}
- if (Arrays.stream(column.get("indkey").toString().split("
")).collect(Collectors.toList()).contains(column.get("attnum").toString())) {
+ if (Arrays.stream(column.get("indkey").toString().split("
")).toList().contains(column.get("attnum").toString())) {
column.put("is_pk", true);
column.put("is_primary_key", true);
} else {
@@ -213,8 +214,7 @@ public final class PostgreSQLColumnPropertiesAppender {
case NUMERIC:
setColumnPrecision(column, fullType);
break;
- case DATE:
- case VARCHAR:
+ case DATE, VARCHAR:
setColumnLength(column, fullType);
break;
default:
@@ -323,7 +323,7 @@ public final class PostgreSQLColumnPropertiesAppender {
return;
}
Collection<Map<String, String>> attOptions = new LinkedList<>();
- Collection<String> columnVariables = Arrays.stream((String[]) ((Array)
column.get("attoptions")).getArray()).collect(Collectors.toList());
+ Collection<String> columnVariables = Arrays.stream((String[]) ((Array)
column.get("attoptions")).getArray()).toList();
for (String each : columnVariables) {
Map<String, String> columnVariable = new LinkedHashMap<>();
columnVariable.put("name", each.substring(0,
each.indexOf(ATT_OPTION_SPLIT)));
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/constraints/PostgreSQLConstraintsPropertiesAppender.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/constraints/PostgreSQLConstraintsPropertiesAppender.java
index 23325e23a87..7295c987972 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/constraints/PostgreSQLConstraintsPropertiesAppender.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/constraints/PostgreSQLConstraintsPropertiesAppender.java
@@ -20,6 +20,7 @@ package
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.constr
import
org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.PostgreSQLDDLTemplateExecutor;
import java.sql.Connection;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -59,7 +60,7 @@ public final class PostgreSQLConstraintsPropertiesAppender {
private void loadPrimaryOrUniqueConstraint(final Map<String, Object>
context, final String name, final String type) {
Collection<Map<String, Object>> constraintsProps =
fetchConstraintsProperties(context, type);
fetchConstraintsColumns(constraintsProps);
- context.put(name, constraintsProps.stream().filter(each ->
!isPartitionAndConstraintInherited(each,
context)).collect(Collectors.toList()));
+ context.put(name, constraintsProps.stream().filter(each ->
!isPartitionAndConstraintInherited(each,
context)).collect(Collectors.toCollection(ArrayList::new)));
}
private Collection<Map<String, Object>> fetchConstraintsProperties(final
Map<String, Object> context, final String constraintType) {
@@ -73,8 +74,8 @@ public final class PostgreSQLConstraintsPropertiesAppender {
private void fetchConstraintsColumns(final Collection<Map<String, Object>>
constraintsProps) {
for (Map<String, Object> each : constraintsProps) {
- each.put("columns",
- fetchConstraintsCols(each).stream().<Map<String,
Object>>map(col -> Collections.singletonMap("column", stripQuote((String)
col.get("column")))).collect(Collectors.toList()));
+ each.put("columns",
fetchConstraintsCols(each).stream().<Map<String, Object>>map(col ->
Collections.singletonMap("column", stripQuote((String) col.get("column"))))
+ .collect(Collectors.toCollection(ArrayList::new)));
appendConstraintsInclude(each);
}
}
@@ -82,7 +83,7 @@ public final class PostgreSQLConstraintsPropertiesAppender {
private void appendConstraintsInclude(final Map<String, Object>
constraintsProp) {
Collection<Object> includes = templateExecutor.getMajorVersion() >=
PG_CONSTRAINTS_INCLUDE_VERSION
?
templateExecutor.executeByTemplate(Collections.singletonMap("cid",
constraintsProp.get("oid")),
-
"component/index_constraint/%s/get_constraint_include.ftl").stream().map(each
-> each.get("colname")).collect(Collectors.toList())
+
"component/index_constraint/%s/get_constraint_include.ftl").stream().map(each
-> each.get("colname")).collect(Collectors.toCollection(ArrayList::new))
: Collections.emptyList();
constraintsProp.put("include", includes);
}
@@ -106,11 +107,11 @@ public final class
PostgreSQLConstraintsPropertiesAppender {
}
private Collection<Map<String, Object>> fetchForeignKeys(final Map<String,
Object> context) {
- return getForeignKeys((Long) context.get("tid")).stream().filter(each
-> !isPartitionAndConstraintInherited(each,
context)).collect(Collectors.toList());
+ return getForeignKeys((Long) context.get("tid")).stream().filter(each
-> !isPartitionAndConstraintInherited(each,
context)).collect(Collectors.toCollection(ArrayList::new));
}
private Collection<Map<String, Object>> fetchCheckConstraints(final
Map<String, Object> context) {
- return getCheckConstraints((Long)
context.get("tid")).stream().filter(each ->
!isPartitionAndConstraintInherited(each, context)).collect(Collectors.toList());
+ return getCheckConstraints((Long)
context.get("tid")).stream().filter(each ->
!isPartitionAndConstraintInherited(each,
context)).collect(Collectors.toCollection(ArrayList::new));
}
private Collection<Map<String, Object>> getExclusionConstraints(final
Map<String, Object> context) {
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/index/PostgreSQLIndexSQLGenerator.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/index/PostgreSQLIndexSQLGenerator.java
index c6621285be5..b42492bed17 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/index/PostgreSQLIndexSQLGenerator.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/index/PostgreSQLIndexSQLGenerator.java
@@ -23,6 +23,7 @@ import org.postgresql.jdbc.PgArray;
import java.sql.Connection;
import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
@@ -69,7 +70,7 @@ public final class PostgreSQLIndexSQLGenerator {
appendColumnDetails(indexData, (Long) indexNode.get("oid"));
if (templateExecutor.getMajorVersion() >= PG_INDEX_INCLUDE_VERSION) {
Collection<Map<String, Object>> includeDetails =
templateExecutor.executeByTemplate(Collections.singletonMap("idx",
indexNode.get("oid")), "component/indexes/%s/include_details.ftl");
- indexData.put("include", includeDetails.stream().map(each ->
each.get("colname")).collect(Collectors.toList()));
+ indexData.put("include", includeDetails.stream().map(each ->
each.get("colname")).collect(Collectors.toCollection(ArrayList::new)));
}
return doGenerateIndexSQL(indexData);
}
diff --git
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/table/PostgreSQLTablePropertiesLoader.java
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/table/PostgreSQLTablePropertiesLoader.java
index ea6aeb180a4..255958a1d13 100644
---
a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/table/PostgreSQLTablePropertiesLoader.java
+++
b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/ddl/table/PostgreSQLTablePropertiesLoader.java
@@ -118,7 +118,7 @@ public final class PostgreSQLTablePropertiesLoader {
}
private boolean anyIsTrue(final Collection<Object> collection) {
- return collection.stream().anyMatch(each -> each instanceof Boolean &&
(Boolean) each);
+ return collection.stream().anyMatch(each -> each instanceof Boolean
value && value);
}
private void updateRlspolicySupport(final Map<String, Object> context) {
diff --git
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordTableInventoryCheckCalculatorTest.java
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordTableInventoryCheckCalculatorTest.java
index bc1d20d9ffd..2dd8fe4ba75 100644
---
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordTableInventoryCheckCalculatorTest.java
+++
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordTableInventoryCheckCalculatorTest.java
@@ -33,6 +33,7 @@ import
org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.util.close.QuietlyCloser;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
@@ -111,6 +112,14 @@ class RecordTableInventoryCheckCalculatorTest {
assertRangeQueryCalculatedResult(calculator, param,
Range.openClosed(4, null), 4, 8);
}
+ @Test
+ void assertCalculateOfRangeQueryWithOnlyUpperBound() {
+ RecordTableInventoryCheckCalculator calculator = new
RecordTableInventoryCheckCalculator(10, 1, StreamingRangeType.SMALL);
+ TableInventoryCalculateParameter param = new
TableInventoryCalculateParameter(dataSource, new QualifiedTable(null,
"t_order"), Collections.emptyList(), buildUserIdUniqueKey(),
+ QueryType.RANGE_QUERY, null);
+ assertRangeQueryCalculatedResult(calculator, param,
Range.openClosed(null, 3), 7, 3);
+ }
+
@ParameterizedTest
@CsvSource({"SMALL", "LARGE"})
void assertCalculateOfRangeQueryWithMultiColumnUniqueKeys(final String
streamingRangeType) {
@@ -359,6 +368,22 @@ class RecordTableInventoryCheckCalculatorTest {
assertThat(actual.getMaxUniqueKeyValue().get(), is(3));
}
+ @Test
+ void assertCalculateOfPointQueryWithShardingColumn() {
+ RecordTableInventoryCheckCalculator calculator = new
RecordTableInventoryCheckCalculator(3, StreamingRangeType.SMALL);
+ TableInventoryCalculateParameter param = new
TableInventoryCalculateParameter(dataSource, new QualifiedTable(null,
"t_order"), Collections.emptyList(), buildUserIdUniqueKey(),
+ QueryType.POINT_QUERY, null);
+ param.setUniqueKeysValues(Collections.singleton(3));
+ param.setShardingColumnsNames(Collections.singletonList("order_id"));
+ param.setShardingColumnsValues(Collections.singletonList(5));
+ Optional<TableInventoryCheckCalculatedResult> actualResult =
calculator.calculateChunk(param);
+ QuietlyCloser.close(param.getCalculationContext());
+ assertTrue(actualResult.isPresent());
+ RecordTableInventoryCheckCalculatedResult actual =
(RecordTableInventoryCheckCalculatedResult) actualResult.get();
+ assertThat(actual.getRecordsCount(), is(1));
+ assertRecord(actual.getRecords().get(0), 3, 5);
+ }
+
@ParameterizedTest
@CsvSource({"SMALL", "LARGE"})
void assertCalculateOfPointRangeQuery(final String streamingRangeType) {
diff --git
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngineTest.java
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngineTest.java
index 319f0ec7ece..93db895d7d1 100644
---
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngineTest.java
+++
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/column/InventoryColumnValueReaderEngineTest.java
@@ -137,6 +137,12 @@ class InventoryColumnValueReaderEngineTest {
assertThat(engine.read(resultSet, metaData, 1), is(new
BigInteger("1")));
}
+ @Test
+ void assertReadWithNullUnsignedBigIntValue() throws SQLException {
+ when(metaData.getColumnType(1)).thenReturn(Types.BIGINT);
+ assertNull(engine.read(resultSet, metaData, 1));
+ }
+
@Test
void assertReadWithNumericValue() throws SQLException {
when(metaData.getColumnType(1)).thenReturn(Types.NUMERIC);
