Remove DialectDataSourceChecker in pipeline module (#32152)

* Remove DialectDataSourceChecker in pipeline module

* Remove DialectDataSourceChecker in pipeline module

kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/check/datasource/PostgreSQLDataSourceCheckerTest.java → infra/database/type/postgresql/src/test/java/org/apache/shardingsphere/infra/database/postgresql/checker/PostgreSQLDatabaseEnvironmentCheckerTest.java

@@ -15,9 +15,10 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.postgresql.check.datasource;
+package org.apache.shardingsphere.infra.database.postgresql.checker;
 
-import org.apache.shardingsphere.data.pipeline.core.exception.job.PrepareJobWithoutEnoughPrivilegeException;
+import org.apache.shardingsphere.infra.database.core.checker.PrivilegeCheckType;
+import org.apache.shardingsphere.infra.database.core.exception.MissingRequiredPrivilegeException;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -38,7 +39,7 @@
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
-class PostgreSQLDataSourceCheckerTest {
+class PostgreSQLDatabaseEnvironmentCheckerTest {
 
     @Mock
     private DataSource dataSource;
@@ -67,28 +68,28 @@ void setUp() throws SQLException {
 
     @Test
     void assertCheckRolReplication() throws SQLException {
-        PostgreSQLDataSourceChecker dataSourceChecker = new PostgreSQLDataSourceChecker();
+        PostgreSQLDatabaseEnvironmentChecker dataSourceChecker = new PostgreSQLDatabaseEnvironmentChecker();
         when(resultSet.getString("rolreplication")).thenReturn("t");
         when(resultSet.getString("rolsuper")).thenReturn("f");
-        dataSourceChecker.checkPrivilege(dataSource);
+        dataSourceChecker.checkPrivilege(dataSource, PrivilegeCheckType.PIPELINE);
         verify(resultSet, atLeastOnce()).getString("rolsuper");
     }
 
     @Test
     void assertCheckRolSuper() throws SQLException {
-        PostgreSQLDataSourceChecker dataSourceChecker = new PostgreSQLDataSourceChecker();
+        PostgreSQLDatabaseEnvironmentChecker dataSourceChecker = new PostgreSQLDatabaseEnvironmentChecker();
         when(resultSet.getString("rolsuper")).thenReturn("t");
         when(resultSet.getString("rolreplication")).thenReturn("f");
-        dataSourceChecker.checkPrivilege(dataSource);
+        dataSourceChecker.checkPrivilege(dataSource, PrivilegeCheckType.PIPELINE);
         verify(resultSet, atLeastOnce()).getString("rolreplication");
     }
 
     @Test
     void assertCheckNoPrivilege() throws SQLException {
-        PostgreSQLDataSourceChecker dataSourceChecker = new PostgreSQLDataSourceChecker();
+        PostgreSQLDatabaseEnvironmentChecker dataSourceChecker = new PostgreSQLDatabaseEnvironmentChecker();
         when(resultSet.getString("rolsuper")).thenReturn("f");
         when(resultSet.getString("rolreplication")).thenReturn("f");
-        assertThrows(PrepareJobWithoutEnoughPrivilegeException.class, () -> dataSourceChecker.checkPrivilege(dataSource));
+        assertThrows(MissingRequiredPrivilegeException.class, () -> dataSourceChecker.checkPrivilege(dataSource, PrivilegeCheckType.PIPELINE));
         verify(resultSet, atLeastOnce()).getString("rolreplication");
     }
 }

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/checker/DataSourceCheckEngine.java

@@ -20,6 +20,8 @@
 import org.apache.shardingsphere.data.pipeline.core.exception.job.PrepareJobWithTargetTableNotEmptyException;
 import org.apache.shardingsphere.data.pipeline.core.importer.ImporterConfiguration;
 import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.sql.PipelinePrepareSQLBuilder;
+import org.apache.shardingsphere.infra.database.core.checker.DialectDatabaseEnvironmentChecker;
+import org.apache.shardingsphere.infra.database.core.checker.PrivilegeCheckType;
 import org.apache.shardingsphere.infra.database.core.spi.DatabaseTypedSPILoader;
 import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
 import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
@@ -38,12 +40,12 @@
  */
 public final class DataSourceCheckEngine {
 
-    private final DialectDataSourceChecker checker;
+    private final DialectDatabaseEnvironmentChecker checker;
 
     private final PipelinePrepareSQLBuilder sqlBuilder;
 
     public DataSourceCheckEngine(final DatabaseType databaseType) {
-        checker = DatabaseTypedSPILoader.findService(DialectDataSourceChecker.class, databaseType).orElse(null);
+        checker = DatabaseTypedSPILoader.findService(DialectDatabaseEnvironmentChecker.class, databaseType).orElse(null);
         sqlBuilder = new PipelinePrepareSQLBuilder(databaseType);
     }
 
@@ -73,7 +75,7 @@ public void checkSourceDataSources(final Collection<DataSource> dataSources) {
         if (null == checker) {
             return;
         }
-        dataSources.forEach(checker::checkPrivilege);
+        dataSources.forEach(each -> checker.checkPrivilege(each, PrivilegeCheckType.PIPELINE));
         dataSources.forEach(checker::checkVariable);
     }