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 2024/08/24 07:52:13 UTC
(shardingsphere) branch master updated: Refactor DialectJDBCStreamQueryBuilder (#32655)
This is an automated email from the ASF dual-hosted git repository.
zhangliang 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 345a5bba796 Refactor DialectJDBCStreamQueryBuilder (#32655)
345a5bba796 is described below
commit 345a5bba796d2715f162e2706b592b895ec9f266
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Sat Aug 24 15:52:06 2024 +0800
Refactor DialectJDBCStreamQueryBuilder (#32655)
* Refactor DialectJDBCStreamQueryBuilder
* Refactor DialectJDBCStreamQueryBuilder
---
.../calculator/RecordSingleTableInventoryCalculator.java | 6 +-----
.../core/ingest/dumper/inventory/InventoryDumper.java | 12 ++----------
.../pipeline/core/query/DialectJDBCStreamQueryBuilder.java | 5 ++---
.../data/pipeline/core/query/JDBCStreamQueryBuilder.java | 5 +++--
.../pipeline/mysql/query/MySQLJDBCStreamQueryBuilder.java | 3 +--
.../opengauss/query/OpenGaussJDBCStreamQueryBuilder.java | 4 ++--
.../postgresql/query/PostgreSQLJDBCStreamQueryBuilder.java | 4 ++--
.../core/fixture/h2/query/H2JDBCStreamQueryBuilder.java | 3 +--
8 files changed, 14 insertions(+), 28 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java
index 79e9afec6e3..3051ebfb6e4 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/calculator/RecordSingleTableInventoryCalculator.java
@@ -29,7 +29,6 @@ import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.inventory.colu
import org.apache.shardingsphere.data.pipeline.core.query.JDBCStreamQueryBuilder;
import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql.PipelineDataConsistencyCalculateSQLBuilder;
import org.apache.shardingsphere.infra.annotation.HighFrequencyInvocation;
-import org.apache.shardingsphere.infra.database.mysql.type.MySQLDatabaseType;
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.exception.core.external.sql.type.kernel.category.PipelineSQLException;
import org.apache.shardingsphere.infra.util.close.QuietlyCloser;
@@ -142,11 +141,8 @@ public final class RecordSingleTableInventoryCalculator extends AbstractStreamin
private void fulfillCalculationContext(final CalculationContext calculationContext, final SingleTableInventoryCalculateParameter param) throws SQLException {
String sql = getQuerySQL(param);
- PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(param.getDatabaseType(), calculationContext.getConnection(), sql);
+ PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(param.getDatabaseType(), calculationContext.getConnection(), sql, chunkSize);
setCurrentStatement(preparedStatement);
- if (!(param.getDatabaseType() instanceof MySQLDatabaseType)) {
- preparedStatement.setFetchSize(chunkSize);
- }
calculationContext.setPreparedStatement(preparedStatement);
setParameters(preparedStatement, param);
ResultSet resultSet = preparedStatement.executeQuery();
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java
index b21809b6e73..eed116908f7 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java
@@ -47,9 +47,7 @@ import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql.PipelineInven
import org.apache.shardingsphere.data.pipeline.core.util.PipelineJdbcUtils;
import org.apache.shardingsphere.infra.annotation.HighFrequencyInvocation;
import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
-import org.apache.shardingsphere.infra.database.mysql.type.MySQLDatabaseType;
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
-import org.apache.shardingsphere.infra.util.DatabaseTypeUtils;
import javax.sql.DataSource;
import java.sql.Connection;
@@ -168,11 +166,8 @@ public class InventoryDumper extends AbstractPipelineLifecycleRunnable implement
private List<Record> dumpPageByPage(final Connection connection, final InventoryQueryParameter queryParam, final AtomicLong rowCount) throws SQLException {
DatabaseType databaseType = dumperContext.getCommonContext().getDataSourceConfig().getDatabaseType();
int batchSize = dumperContext.getBatchSize();
- try (PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(databaseType, connection, buildInventoryDumpPageByPageSQL(queryParam))) {
+ try (PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(databaseType, connection, buildInventoryDumpPageByPageSQL(queryParam), batchSize)) {
runningStatement.set(preparedStatement);
- if (!(DatabaseTypeUtils.getTrunkDatabaseType(databaseType) instanceof MySQLDatabaseType)) {
- preparedStatement.setFetchSize(batchSize);
- }
setParameters(preparedStatement, queryParam, false);
try (ResultSet resultSet = preparedStatement.executeQuery()) {
JobRateLimitAlgorithm rateLimitAlgorithm = dumperContext.getRateLimitAlgorithm();
@@ -287,11 +282,8 @@ public class InventoryDumper extends AbstractPipelineLifecycleRunnable implement
if (null != dumperContext.getTransactionIsolation()) {
connection.setTransactionIsolation(dumperContext.getTransactionIsolation());
}
- try (PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(databaseType, connection, buildInventoryDumpSQLWithStreamingQuery())) {
+ try (PreparedStatement preparedStatement = JDBCStreamQueryBuilder.build(databaseType, connection, buildInventoryDumpSQLWithStreamingQuery(), batchSize)) {
runningStatement.set(preparedStatement);
- if (!(DatabaseTypeUtils.getTrunkDatabaseType(databaseType) instanceof MySQLDatabaseType)) {
- preparedStatement.setFetchSize(batchSize);
- }
PrimaryKeyIngestPosition<?> primaryPosition = (PrimaryKeyIngestPosition<?>) dumperContext.getCommonContext().getPosition();
InventoryQueryParameter queryParam = InventoryQueryParameter.buildForRangeQuery(new QueryRange(primaryPosition.getBeginValue(), true, primaryPosition.getEndValue()));
setParameters(preparedStatement, queryParam, true);
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/DialectJDBCStreamQueryBuilder.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/DialectJDBCStreamQueryBuilder.java
index c93b79a76ea..b3868a1aaac 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/DialectJDBCStreamQueryBuilder.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/DialectJDBCStreamQueryBuilder.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.data.pipeline.core.query;
import org.apache.shardingsphere.infra.database.core.spi.DatabaseTypedSPI;
-import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import java.sql.Connection;
@@ -35,10 +34,10 @@ public interface DialectJDBCStreamQueryBuilder extends DatabaseTypedSPI {
* Build streamed prepared statement.
*
* @param connection connection
- * @param databaseType database type
* @param sql SQL to be queried
+ * @param batchSize batch size
* @return built prepared statement
* @throws SQLException SQL exception
*/
- PreparedStatement build(DatabaseType databaseType, Connection connection, String sql) throws SQLException;
+ PreparedStatement build(Connection connection, String sql, int batchSize) throws SQLException;
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/JDBCStreamQueryBuilder.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/JDBCStreamQueryBuilder.java
index 6a945f440c3..2ce95036d9d 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/JDBCStreamQueryBuilder.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/query/JDBCStreamQueryBuilder.java
@@ -42,13 +42,14 @@ public final class JDBCStreamQueryBuilder {
* @param connection connection
* @param databaseType database type
* @param sql SQL to be queried
+ * @param batchSize batch size
* @return built prepared statement
* @throws SQLException SQL exception
*/
- public static PreparedStatement build(final DatabaseType databaseType, final Connection connection, final String sql) throws SQLException {
+ public static PreparedStatement build(final DatabaseType databaseType, final Connection connection, final String sql, final int batchSize) throws SQLException {
Optional<DialectJDBCStreamQueryBuilder> dialectBuilder = DatabaseTypedSPILoader.findService(DialectJDBCStreamQueryBuilder.class, databaseType);
if (dialectBuilder.isPresent()) {
- return dialectBuilder.get().build(databaseType, connection, sql);
+ return dialectBuilder.get().build(connection, sql, batchSize);
}
log.warn("not support {} streaming query now, pay attention to memory usage", databaseType.getType());
return connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/query/MySQLJDBCStreamQueryBuilder.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/query/MySQLJDBCStreamQueryBuilder.java
index f5bb8eb012a..1842b95f1a8 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/query/MySQLJDBCStreamQueryBuilder.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/query/MySQLJDBCStreamQueryBuilder.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.data.pipeline.mysql.query;
import org.apache.shardingsphere.data.pipeline.core.query.DialectJDBCStreamQueryBuilder;
-import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -31,7 +30,7 @@ import java.sql.SQLException;
public final class MySQLJDBCStreamQueryBuilder implements DialectJDBCStreamQueryBuilder {
@Override
- public PreparedStatement build(final DatabaseType databaseType, final Connection connection, final String sql) throws SQLException {
+ public PreparedStatement build(final Connection connection, final String sql, final int batchSize) throws SQLException {
PreparedStatement result = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
result.setFetchSize(Integer.MIN_VALUE);
return result;
diff --git a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/query/OpenGaussJDBCStreamQueryBuilder.java b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/query/OpenGaussJDBCStreamQueryBuilder.java
index 9435621ca66..edcacf4a784 100644
--- a/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/query/OpenGaussJDBCStreamQueryBuilder.java
+++ b/kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/query/OpenGaussJDBCStreamQueryBuilder.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.data.pipeline.opengauss.query;
import org.apache.shardingsphere.data.pipeline.core.query.DialectJDBCStreamQueryBuilder;
-import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -31,9 +30,10 @@ import java.sql.SQLException;
public final class OpenGaussJDBCStreamQueryBuilder implements DialectJDBCStreamQueryBuilder {
@Override
- public PreparedStatement build(final DatabaseType databaseType, final Connection connection, final String sql) throws SQLException {
+ public PreparedStatement build(final Connection connection, final String sql, final int batchSize) throws SQLException {
PreparedStatement result = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY, ResultSet.CLOSE_CURSORS_AT_COMMIT);
connection.setAutoCommit(false);
+ result.setFetchSize(batchSize);
return result;
}
diff --git a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/query/PostgreSQLJDBCStreamQueryBuilder.java b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/query/PostgreSQLJDBCStreamQueryBuilder.java
index 174f80dd7b3..b32c0219be6 100644
--- a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/query/PostgreSQLJDBCStreamQueryBuilder.java
+++ b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/query/PostgreSQLJDBCStreamQueryBuilder.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.data.pipeline.postgresql.query;
import org.apache.shardingsphere.data.pipeline.core.query.DialectJDBCStreamQueryBuilder;
-import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -31,9 +30,10 @@ import java.sql.SQLException;
public final class PostgreSQLJDBCStreamQueryBuilder implements DialectJDBCStreamQueryBuilder {
@Override
- public PreparedStatement build(final DatabaseType databaseType, final Connection connection, final String sql) throws SQLException {
+ public PreparedStatement build(final Connection connection, final String sql, final int batchSize) throws SQLException {
PreparedStatement result = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY, ResultSet.CLOSE_CURSORS_AT_COMMIT);
connection.setAutoCommit(false);
+ result.setFetchSize(batchSize);
return result;
}
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/h2/query/H2JDBCStreamQueryBuilder.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/h2/query/H2JDBCStreamQueryBuilder.java
index 04f2d6e6228..128a33a0a66 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/h2/query/H2JDBCStreamQueryBuilder.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/fixture/h2/query/H2JDBCStreamQueryBuilder.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.test.it.data.pipeline.core.fixture.h2.query;
import org.apache.shardingsphere.data.pipeline.core.query.DialectJDBCStreamQueryBuilder;
-import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -31,7 +30,7 @@ import java.sql.SQLException;
public final class H2JDBCStreamQueryBuilder implements DialectJDBCStreamQueryBuilder {
@Override
- public PreparedStatement build(final DatabaseType databaseType, final Connection connection, final String sql) throws SQLException {
+ public PreparedStatement build(final Connection connection, final String sql, final int batchSize) throws SQLException {
return connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
}