Refactor InventoryDumper

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/DatabaseTypeUtils.java → infra/common/src/main/java/org/apache/shardingsphere/infra/util/DatabaseTypeUtils.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.core.util;
+package org.apache.shardingsphere.infra.util;
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;

infra/common/src/test/java/org/apache/shardingsphere/infra/util/DatabaseTypeUtilsTest.java

@@ -0,0 +1,38 @@
+/*
+ * 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.infra.util;
+
+import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
+import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class DatabaseTypeUtilsTest {
+
+    @Test
+    void assertIsMySQL() {
+        DatabaseType actual = TypedSPILoader.getService(DatabaseType.class, "MySQL");
+        assertTrue(DatabaseTypeUtils.isMySQL(actual));
+        actual = TypedSPILoader.getService(DatabaseType.class, "MariaDB");
+        assertTrue(DatabaseTypeUtils.isMySQL(actual));
+        actual = TypedSPILoader.getService(DatabaseType.class, "PostgreSQL");
+        assertFalse(DatabaseTypeUtils.isMySQL(actual));
+    }
+}

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java

@@ -44,7 +44,7 @@
 import org.apache.shardingsphere.data.pipeline.core.query.JDBCStreamQueryBuilder;
 import org.apache.shardingsphere.data.pipeline.core.ratelimit.JobRateLimitAlgorithm;
 import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql.PipelineInventoryDumpSQLBuilder;
-import org.apache.shardingsphere.data.pipeline.core.util.DatabaseTypeUtils;
+import org.apache.shardingsphere.infra.util.DatabaseTypeUtils;
 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;
@@ -108,10 +108,10 @@ protected void runBlocking() {
         }
         init();
         try (Connection connection = dataSource.getConnection()) {
-            if (!Strings.isNullOrEmpty(dumperContext.getQuerySQL()) || !dumperContext.hasUniqueKey() || isPrimaryKeyWithoutRanged(position)) {
-                dumpWithStreamingQuery(connection);
-            } else {
+            if (Strings.isNullOrEmpty(dumperContext.getQuerySQL()) && dumperContext.hasUniqueKey() && !isPrimaryKeyWithoutRange(position)) {
                 dumpPageByPage(connection);
+            } else {
+                dumpWithStreamingQuery(connection);
             }
             // CHECKSTYLE:OFF
         } catch (final SQLException | RuntimeException ex) {
@@ -121,20 +121,17 @@ protected void runBlocking() {
         }
     }
 
-    private boolean isPrimaryKeyWithoutRanged(final IngestPosition position) {
-        return position instanceof PrimaryKeyIngestPosition && null == ((PrimaryKeyIngestPosition<?>) position).getBeginValue() && null == ((PrimaryKeyIngestPosition<?>) position).getEndValue();
-    }
-
-    /**
-     * Initialize.
-     */
-    public void init() {
+    private void init() {
         if (null == tableMetaData) {
             tableMetaData = metaDataLoader.getTableMetaData(
                     dumperContext.getCommonContext().getTableAndSchemaNameMapper().getSchemaName(dumperContext.getLogicTableName()), dumperContext.getActualTableName());
         }
     }
 
+    private boolean isPrimaryKeyWithoutRange(final IngestPosition position) {
+        return position instanceof PrimaryKeyIngestPosition && null == ((PrimaryKeyIngestPosition<?>) position).getBeginValue() && null == ((PrimaryKeyIngestPosition<?>) position).getEndValue();
+    }
+
     @SuppressWarnings("MagicConstant")
     private void dumpWithStreamingQuery(final Connection connection) throws SQLException {
         int batchSize = dumperContext.getBatchSize();