Add DefaultDataSourcePrepareOption (#29424)

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/PipelineJobPreparerUtils.java

@@ -76,12 +76,9 @@ public static boolean isIncrementalSupported(final DatabaseType databaseType) {
      * @throws SQLException if prepare target schema fail
      */
     public static void prepareTargetSchema(final DatabaseType databaseType, final PrepareTargetSchemasParameter prepareTargetSchemasParam) throws SQLException {
-        Optional<DataSourcePrepareOption> option = DatabaseTypedSPILoader.findService(DataSourcePrepareOption.class, databaseType);
-        if (!option.isPresent()) {
-            log.info("Data source preparer option null, ignore prepare target");
-            return;
-        }
-        new DataSourcePreparer(option.get()).prepareTargetSchemas(prepareTargetSchemasParam);
+        DataSourcePrepareOption option = DatabaseTypedSPILoader.findService(DataSourcePrepareOption.class, databaseType)
+                .orElseGet(() -> DatabaseTypedSPILoader.getService(DataSourcePrepareOption.class, null));
+        new DataSourcePreparer(option).prepareTargetSchemas(prepareTargetSchemasParam);
     }
 
     /**
@@ -105,13 +102,10 @@ public static SQLParserEngine getSQLParserEngine(final ShardingSphereMetaData me
      * @throws SQLException SQL exception
      */
     public static void prepareTargetTables(final DatabaseType databaseType, final PrepareTargetTablesParameter prepareTargetTablesParam) throws SQLException {
-        Optional<DataSourcePrepareOption> option = DatabaseTypedSPILoader.findService(DataSourcePrepareOption.class, databaseType);
-        if (!option.isPresent()) {
-            log.info("Data source preparer option null, ignore prepare target");
-            return;
-        }
+        DataSourcePrepareOption option = DatabaseTypedSPILoader.findService(DataSourcePrepareOption.class, databaseType)
+                .orElseGet(() -> DatabaseTypedSPILoader.getService(DataSourcePrepareOption.class, null));
         long startTimeMillis = System.currentTimeMillis();
-        new DataSourcePreparer(option.get()).prepareTargetTables(prepareTargetTablesParam);
+        new DataSourcePreparer(option).prepareTargetTables(prepareTargetTablesParam);
         log.info("prepareTargetTables cost {} ms", System.currentTimeMillis() - startTimeMillis);
     }
 

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePrepareOption.java → kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/datasource/DefaultDataSourcePrepareOption.java

@@ -15,17 +15,15 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.mysql.prepare.datasource;
-
-import org.apache.shardingsphere.data.pipeline.core.preparer.datasource.DataSourcePrepareOption;
+package org.apache.shardingsphere.data.pipeline.core.preparer.datasource;
 
 /**
- * Data source prepare option for MySQL.
+ * Default data source prepare option.
  */
-public final class MySQLDataSourcePrepareOption implements DataSourcePrepareOption {
+public final class DefaultDataSourcePrepareOption implements DataSourcePrepareOption {
 
     @Override
-    public String getDatabaseType() {
-        return "MySQL";
+    public boolean isDefault() {
+        return true;
     }
 }

test/it/pipeline/src/test/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.preparer.datasource.DataSourcePrepareOption → kernel/data-pipeline/core/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.preparer.datasource.DataSourcePrepareOption

@@ -15,4 +15,4 @@
 # limitations under the License.
 #
 
-org.apache.shardingsphere.test.it.data.pipeline.core.fixture.H2DataSourcePrepareOption
+org.apache.shardingsphere.data.pipeline.core.preparer.datasource.DefaultDataSourcePrepareOption