apache · zhaojinchao95 · Nov 4, 2023 · Nov 4, 2023 · Nov 4, 2023 · Nov 4, 2023
diff --git a/...rg/apache/shardingsphere/data/pipeline/api/ingest/dumper/context/DumperCommonContext.java b/...rg/apache/shardingsphere/data/pipeline/api/ingest/dumper/context/DumperCommonContext.java
@@ -20,16 +20,13 @@
 import lombok.Getter;
 import lombok.Setter;
 import lombok.ToString;
-import org.apache.shardingsphere.data.pipeline.api.context.TableAndSchemaNameMapper;
 import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
+import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.mapper.ActualAndLogicTableNameMapper;
+import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.mapper.TableAndSchemaNameMapper;
 import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPosition;
-import org.apache.shardingsphere.data.pipeline.api.metadata.ActualTableName;
-import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
-
-import java.util.Map;
 
 /**
- * Base dumper context.
+ * Dumper common context.
  */
 @Getter
 @Setter
@@ -40,53 +37,9 @@ public abstract class DumperCommonContext {
 
     private PipelineDataSourceConfiguration dataSourceConfig;
 
-    private Map<ActualTableName, LogicTableName> tableNameMap;
+    private ActualAndLogicTableNameMapper tableNameMapper;
 
     private TableAndSchemaNameMapper tableAndSchemaNameMapper;
 
     private IngestPosition position;
-
-    /**
-     * Get logic table name.
-     *
-     * @param actualTableName actual table name
-     * @return logic table name
-     */
-    public LogicTableName getLogicTableName(final String actualTableName) {
-        return tableNameMap.get(new ActualTableName(actualTableName));
-    }
-
-    private LogicTableName getLogicTableName(final ActualTableName actualTableName) {
-        return tableNameMap.get(actualTableName);
-    }
-
-    /**
-     * Whether contains table.
-     *
-     * @param actualTableName actual table name
-     * @return contains or not
-     */
-    public boolean containsTable(final String actualTableName) {
-        return tableNameMap.containsKey(new ActualTableName(actualTableName));
-    }
-
-    /**
-     * Get schema name.
-     *
-     * @param logicTableName logic table name
-     * @return schema name. nullable
-     */
-    public String getSchemaName(final LogicTableName logicTableName) {
-        return tableAndSchemaNameMapper.getSchemaName(logicTableName);
-    }
-
-    /**
-     * Get schema name.
-     *
-     * @param actualTableName actual table name
-     * @return schema name, can be nullable 
-     */
-    public String getSchemaName(final ActualTableName actualTableName) {
-        return tableAndSchemaNameMapper.getSchemaName(getLogicTableName(actualTableName));
-    }
 }
diff --git a/...apache/shardingsphere/data/pipeline/api/ingest/dumper/context/InventoryDumperContext.java b/...apache/shardingsphere/data/pipeline/api/ingest/dumper/context/InventoryDumperContext.java
@@ -54,7 +54,7 @@ public final class InventoryDumperContext extends DumperCommonContext {
     public InventoryDumperContext(final DumperCommonContext dumperContext) {
         setDataSourceName(dumperContext.getDataSourceName());
         setDataSourceConfig(dumperContext.getDataSourceConfig());
-        setTableNameMap(dumperContext.getTableNameMap());
+        setTableNameMapper(dumperContext.getTableNameMapper());
         setTableAndSchemaNameMapper(dumperContext.getTableAndSchemaNameMapper());
     }
 

diff --git a/...gsphere/data/pipeline/api/ingest/dumper/context/mapper/ActualAndLogicTableNameMapper.java b/...gsphere/data/pipeline/api/ingest/dumper/context/mapper/ActualAndLogicTableNameMapper.java
@@ -0,0 +1,57 @@
+/*
+ * 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.data.pipeline.api.ingest.dumper.context.mapper;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import org.apache.shardingsphere.data.pipeline.api.metadata.ActualTableName;
+import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
+
+import java.util.Map;
+
+/**
+ * Actual table name and logic table name mapper.
+ */
+@RequiredArgsConstructor
+@Getter
+@ToString
+public final class ActualAndLogicTableNameMapper {
+
+    private final Map<ActualTableName, LogicTableName> tableNameMap;
+
+    /**
+     * Get logic table name.
+     *
+     * @param actualTableName actual table name
+     * @return logic table name
+     */
+    public LogicTableName getLogicTableName(final String actualTableName) {
+        return tableNameMap.get(new ActualTableName(actualTableName));
+    }
+
+    /**
+     * Whether contains table.
+     *
+     * @param actualTableName actual table name
+     * @return contains or not
+     */
+    public boolean containsTable(final String actualTableName) {
+        return tableNameMap.containsKey(new ActualTableName(actualTableName));
+    }
+}
diff --git a/...api/context/TableAndSchemaNameMapper.java → ...text/mapper/TableAndSchemaNameMapper.java b/...api/context/TableAndSchemaNameMapper.java → ...text/mapper/TableAndSchemaNameMapper.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.api.context;
+package org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.mapper;
 
 import lombok.ToString;
 import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;

diff --git a/...ntext/TableNameSchemaNameMappingTest.java → .../mapper/TableAndSchemaNameMapperTest.java b/...ntext/TableNameSchemaNameMappingTest.java → .../mapper/TableAndSchemaNameMapperTest.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.api.context;
+package org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.mapper;
 
 import org.junit.jupiter.api.Test;
 
@@ -28,7 +28,7 @@
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertNull;
 
-class TableNameSchemaNameMappingTest {
+class TableAndSchemaNameMapperTest {
 
     @Test
     void assertConstructFromNull() {

diff --git a/...ain/java/org/apache/shardingsphere/data/pipeline/common/config/ImporterConfiguration.java b/...ain/java/org/apache/shardingsphere/data/pipeline/common/config/ImporterConfiguration.java
@@ -20,7 +20,7 @@
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
-import org.apache.shardingsphere.data.pipeline.api.context.TableAndSchemaNameMapper;
+import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.mapper.TableAndSchemaNameMapper;
 import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfiguration;
 import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
 import org.apache.shardingsphere.data.pipeline.spi.ratelimit.JobRateLimitAlgorithm;

diff --git a/...re/src/main/java/org/apache/shardingsphere/data/pipeline/core/dumper/InventoryDumper.java b/...re/src/main/java/org/apache/shardingsphere/data/pipeline/core/dumper/InventoryDumper.java
@@ -21,17 +21,16 @@
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.InventoryDumperContext;
 import org.apache.shardingsphere.data.pipeline.api.executor.AbstractLifecycleExecutor;
 import org.apache.shardingsphere.data.pipeline.api.ingest.channel.PipelineChannel;
 import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.Dumper;
+import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.InventoryDumperContext;
 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.FinishedRecord;
 import org.apache.shardingsphere.data.pipeline.api.ingest.record.Record;
 import org.apache.shardingsphere.data.pipeline.api.job.JobOperationType;
-import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
 import org.apache.shardingsphere.data.pipeline.api.metadata.loader.PipelineTableMetaDataLoader;
 import org.apache.shardingsphere.data.pipeline.api.metadata.model.PipelineColumnMetaData;
 import org.apache.shardingsphere.data.pipeline.api.metadata.model.PipelineTableMetaData;
@@ -46,8 +45,8 @@
 import org.apache.shardingsphere.data.pipeline.core.exception.IngestException;
 import org.apache.shardingsphere.data.pipeline.core.exception.param.PipelineInvalidParameterException;
 import org.apache.shardingsphere.data.pipeline.spi.ratelimit.JobRateLimitAlgorithm;
-import org.apache.shardingsphere.infra.database.mysql.type.MySQLDatabaseType;
 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 javax.sql.DataSource;
@@ -102,7 +101,8 @@ protected void runBlocking() {
             log.info("Ignored because of already finished.");
             return;
         }
-        PipelineTableMetaData tableMetaData = metaDataLoader.getTableMetaData(dumperContext.getSchemaName(new LogicTableName(dumperContext.getLogicTableName())), dumperContext.getActualTableName());
+        PipelineTableMetaData tableMetaData = metaDataLoader.getTableMetaData(
+                dumperContext.getTableAndSchemaNameMapper().getSchemaName(dumperContext.getLogicTableName()), dumperContext.getActualTableName());
         try (Connection connection = dataSource.getConnection()) {
             dump(tableMetaData, connection);
         } catch (final SQLException ex) {
@@ -156,8 +156,7 @@ private String buildInventoryDumpSQL() {
         if (!Strings.isNullOrEmpty(dumperContext.getQuerySQL())) {
             return dumperContext.getQuerySQL();
         }
-        LogicTableName logicTableName = new LogicTableName(dumperContext.getLogicTableName());
-        String schemaName = dumperContext.getSchemaName(logicTableName);
+        String schemaName = dumperContext.getTableAndSchemaNameMapper().getSchemaName(dumperContext.getLogicTableName());
         if (!dumperContext.hasUniqueKey()) {
             return inventoryDumpSQLBuilder.buildFetchAllSQL(schemaName, dumperContext.getActualTableName());
         }

diff --git a/...rg/apache/shardingsphere/data/pipeline/core/preparer/InventoryRecordsCountCalculator.java b/...rg/apache/shardingsphere/data/pipeline/core/preparer/InventoryRecordsCountCalculator.java
@@ -21,7 +21,6 @@
 import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.InventoryDumperContext;
-import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
 import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceWrapper;
 import org.apache.shardingsphere.data.pipeline.common.sqlbuilder.PipelineCommonSQLBuilder;
 import org.apache.shardingsphere.data.pipeline.core.exception.job.SplitPipelineJobByUniqueKeyException;
@@ -52,7 +51,7 @@ public final class InventoryRecordsCountCalculator {
      * @throws SplitPipelineJobByUniqueKeyException if there's exception from database
      */
     public static long getTableRecordsCount(final InventoryDumperContext dumperContext, final PipelineDataSourceWrapper dataSource) {
-        String schemaName = dumperContext.getSchemaName(new LogicTableName(dumperContext.getLogicTableName()));
+        String schemaName = dumperContext.getTableAndSchemaNameMapper().getSchemaName(dumperContext.getLogicTableName());
         String actualTableName = dumperContext.getActualTableName();
         PipelineCommonSQLBuilder pipelineSQLBuilder = new PipelineCommonSQLBuilder(dataSource.getDatabaseType());
         Optional<String> sql = pipelineSQLBuilder.buildEstimatedCountSQL(schemaName, actualTableName);

diff --git a/...ain/java/org/apache/shardingsphere/data/pipeline/core/preparer/InventoryTaskSplitter.java b/...ain/java/org/apache/shardingsphere/data/pipeline/core/preparer/InventoryTaskSplitter.java
@@ -20,19 +20,18 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.Range;
-import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.InventoryDumperContext;
 import org.apache.shardingsphere.data.pipeline.api.ingest.channel.PipelineChannel;
 import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.Dumper;
+import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.InventoryDumperContext;
 import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPosition;
-import org.apache.shardingsphere.data.pipeline.api.metadata.LogicTableName;
 import org.apache.shardingsphere.data.pipeline.api.metadata.model.PipelineColumnMetaData;
 import org.apache.shardingsphere.data.pipeline.common.config.ImporterConfiguration;
 import org.apache.shardingsphere.data.pipeline.common.config.process.PipelineReadConfiguration;
 import org.apache.shardingsphere.data.pipeline.common.context.InventoryIncrementalJobItemContext;
 import org.apache.shardingsphere.data.pipeline.common.context.InventoryIncrementalProcessContext;
 import org.apache.shardingsphere.data.pipeline.common.datasource.PipelineDataSourceWrapper;
-import org.apache.shardingsphere.data.pipeline.common.ingest.position.pk.type.IntegerPrimaryKeyPosition;
 import org.apache.shardingsphere.data.pipeline.common.ingest.position.PlaceholderPosition;
+import org.apache.shardingsphere.data.pipeline.common.ingest.position.pk.type.IntegerPrimaryKeyPosition;
 import org.apache.shardingsphere.data.pipeline.common.ingest.position.pk.type.StringPrimaryKeyPosition;
 import org.apache.shardingsphere.data.pipeline.common.ingest.position.pk.type.UnsupportedKeyPosition;
 import org.apache.shardingsphere.data.pipeline.common.job.progress.InventoryIncrementalJobItemProgress;
@@ -111,7 +110,7 @@ public Collection<InventoryDumperContext> splitInventoryDumperContext(final Inve
 
     private Collection<InventoryDumperContext> splitByTable(final InventoryDumperContext dumperContext) {
         Collection<InventoryDumperContext> result = new LinkedList<>();
-        dumperContext.getTableNameMap().forEach((key, value) -> {
+        dumperContext.getTableNameMapper().getTableNameMap().forEach((key, value) -> {
             InventoryDumperContext inventoryDumperContext = new InventoryDumperContext(dumperContext);
             // use original table name, for metadata loader, since some database table name case-sensitive
             inventoryDumperContext.setActualTableName(key.getOriginal());
@@ -127,7 +126,7 @@ private Collection<InventoryDumperContext> splitByTable(final InventoryDumperCon
     private Collection<InventoryDumperContext> splitByPrimaryKey(final InventoryDumperContext dumperContext, final InventoryIncrementalJobItemContext jobItemContext,
                                                                  final PipelineDataSourceWrapper dataSource) {
         if (null == dumperContext.getUniqueKeyColumns()) {
-            String schemaName = dumperContext.getSchemaName(new LogicTableName(dumperContext.getLogicTableName()));
+            String schemaName = dumperContext.getTableAndSchemaNameMapper().getSchemaName(dumperContext.getLogicTableName());
             String actualTableName = dumperContext.getActualTableName();
             List<PipelineColumnMetaData> uniqueKeyColumns = PipelineTableMetaDataUtils.getUniqueKeyColumns(schemaName, actualTableName, jobItemContext.getSourceMetaDataLoader());
             dumperContext.setUniqueKeyColumns(uniqueKeyColumns);
@@ -205,7 +204,7 @@ private Range<Long> getUniqueKeyValuesRange(final InventoryIncrementalJobItemCon
         String uniqueKey = dumperContext.getUniqueKeyColumns().get(0).getName();
         PipelineCommonSQLBuilder pipelineSQLBuilder = new PipelineCommonSQLBuilder(jobItemContext.getJobConfig().getSourceDatabaseType());
         String sql = pipelineSQLBuilder.buildUniqueKeyMinMaxValuesSQL(
-                dumperContext.getSchemaName(new LogicTableName(dumperContext.getLogicTableName())), dumperContext.getActualTableName(), uniqueKey);
+                dumperContext.getTableAndSchemaNameMapper().getSchemaName(dumperContext.getLogicTableName()), dumperContext.getActualTableName(), uniqueKey);
         try (
                 Connection connection = dataSource.getConnection();
                 Statement statement = connection.createStatement();

diff --git a/...g/apache/shardingsphere/data/pipeline/core/preparer/datasource/DataSourceCheckEngine.java b/...g/apache/shardingsphere/data/pipeline/core/preparer/datasource/DataSourceCheckEngine.java
@@ -17,7 +17,7 @@
 
 package org.apache.shardingsphere.data.pipeline.core.preparer.datasource;
 
-import org.apache.shardingsphere.data.pipeline.api.context.TableAndSchemaNameMapper;
+import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.mapper.TableAndSchemaNameMapper;
 import org.apache.shardingsphere.data.pipeline.common.sqlbuilder.PipelineCommonSQLBuilder;
 import org.apache.shardingsphere.data.pipeline.core.exception.job.PrepareJobWithInvalidConnectionException;
 import org.apache.shardingsphere.data.pipeline.core.exception.job.PrepareJobWithTargetTableNotEmptyException;

diff --git a/...ache/shardingsphere/data/pipeline/core/preparer/datasource/DataSourceCheckEngineTest.java b/...ache/shardingsphere/data/pipeline/core/preparer/datasource/DataSourceCheckEngineTest.java
@@ -17,7 +17,7 @@
 
 package org.apache.shardingsphere.data.pipeline.core.preparer.datasource;
 
-import org.apache.shardingsphere.data.pipeline.api.context.TableAndSchemaNameMapper;
+import org.apache.shardingsphere.data.pipeline.api.ingest.dumper.context.mapper.TableAndSchemaNameMapper;
 import org.apache.shardingsphere.data.pipeline.core.exception.job.PrepareJobWithInvalidConnectionException;
 import org.apache.shardingsphere.data.pipeline.core.exception.job.PrepareJobWithTargetTableNotEmptyException;
 import org.apache.shardingsphere.infra.database.core.type.DatabaseType;