Refactor MySQLBinlogNumberDataTypeHandlerEngine (#32546)

* Add MySQLBinlogBinaryDataTypeHandler

* Refactor MySQLBinlogNumberDataTypeHandlerEngine

* Refactor MySQLBinlogNumberDataTypeHandlerEngine

Author: terrymanu

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/MySQLBinlogDataTypeHandler.java

@@ -20,9 +20,8 @@
 import lombok.NoArgsConstructor;
 import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColumnMetaData;
 import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.binary.MySQLBinlogBinaryDataTypeHandler;
-import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number.MySQLBinlogNumberDataTypeHandler;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number.MySQLBinlogNumberDataTypeHandlerEngine;
 import org.apache.shardingsphere.db.protocol.mysql.packet.binlog.row.column.value.string.MySQLBinaryString;
-import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
 
 import java.io.Serializable;
 import java.util.Optional;
@@ -41,10 +40,13 @@ public final class MySQLBinlogDataTypeHandler {
      * @return handled column value
      */
     public static Serializable handle(final PipelineColumnMetaData columnMetaData, final Serializable value) {
+        if (null == value) {
+            return null;
+        }
         if (value instanceof MySQLBinaryString) {
             return MySQLBinlogBinaryDataTypeHandler.handle(columnMetaData, value);
         }
-        Optional<MySQLBinlogNumberDataTypeHandler> dataTypeHandler = TypedSPILoader.findService(MySQLBinlogNumberDataTypeHandler.class, columnMetaData.getDataTypeName());
-        return dataTypeHandler.isPresent() ? dataTypeHandler.get().handle(value) : value;
+        Optional<Serializable> result = MySQLBinlogNumberDataTypeHandlerEngine.handle(columnMetaData, value);
+        return result.orElse(value);
     }
 }

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/number/MySQLBinlogNumberDataTypeHandler.java

@@ -17,16 +17,12 @@
 
 package org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number;
 
-import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
-import org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;
-
 import java.io.Serializable;
 
 /**
  * MySQL binlog number data type handler.
  */
-@SingletonSPI
-public interface MySQLBinlogNumberDataTypeHandler extends TypedSPI {
+public interface MySQLBinlogNumberDataTypeHandler {
 
     /**
      * Handle column value.

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/number/MySQLBinlogNumberDataTypeHandlerEngine.java

@@ -0,0 +1,60 @@
+/*
+ * 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.mysql.ingest.dumper.type.number;
+
+import lombok.NoArgsConstructor;
+import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColumnMetaData;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number.impl.MySQLBinlogUnsignedBigintHandler;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number.impl.MySQLBinlogUnsignedIntHandler;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number.impl.MySQLBinlogUnsignedMediumintHandler;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number.impl.MySQLBinlogUnsignedSmallintHandler;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.dumper.type.number.impl.MySQLBinlogUnsignedTinyintHandler;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * MySQL binlog number data type handler engine.
+ */
+@NoArgsConstructor(access = lombok.AccessLevel.PRIVATE)
+public final class MySQLBinlogNumberDataTypeHandlerEngine {
+    
+    private static final Map<String, MySQLBinlogNumberDataTypeHandler> HANDLERS = new HashMap<>();
+    
+    static {
+        HANDLERS.put("TINYINT UNSIGNED", new MySQLBinlogUnsignedTinyintHandler());
+        HANDLERS.put("SMALLINT UNSIGNED", new MySQLBinlogUnsignedSmallintHandler());
+        HANDLERS.put("MEDIUMINT UNSIGNED", new MySQLBinlogUnsignedMediumintHandler());
+        HANDLERS.put("INT UNSIGNED", new MySQLBinlogUnsignedIntHandler());
+        HANDLERS.put("BIGINT UNSIGNED", new MySQLBinlogUnsignedBigintHandler());
+    }
+    
+    /**
+     * Handle column value.
+     *
+     * @param columnMetaData column meta data
+     * @param value column value
+     * @return handled column value
+     */
+    public static Optional<Serializable> handle(final PipelineColumnMetaData columnMetaData, final Serializable value) {
+        String dataTypeName = columnMetaData.getDataTypeName();
+        return HANDLERS.containsKey(dataTypeName) ? Optional.of(HANDLERS.get(dataTypeName).handle(value)) : Optional.empty();
+    }
+}

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/number/impl/MySQLBinlogUnsignedBigintHandler.java

@@ -31,15 +31,7 @@ public final class MySQLBinlogUnsignedBigintHandler implements MySQLBinlogNumber
 
     @Override
     public Serializable handle(final Serializable value) {
-        if (null == value) {
-            return null;
-        }
         long longValue = (long) value;
         return longValue < 0L ? BIGINT_MODULO.add(BigInteger.valueOf(longValue)) : longValue;
     }
-    
-    @Override
-    public String getType() {
-        return "BIGINT UNSIGNED";
-    }
 }

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/number/impl/MySQLBinlogUnsignedIntHandler.java

@@ -30,15 +30,7 @@ public final class MySQLBinlogUnsignedIntHandler implements MySQLBinlogNumberDat
 
     @Override
     public Serializable handle(final Serializable value) {
-        if (null == value) {
-            return null;
-        }
         int intValue = (int) value;
         return intValue < 0 ? INT_MODULO + intValue : intValue;
     }
-    
-    @Override
-    public String getType() {
-        return "INT UNSIGNED";
-    }
 }

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/number/impl/MySQLBinlogUnsignedMediumintHandler.java

@@ -30,15 +30,7 @@ public final class MySQLBinlogUnsignedMediumintHandler implements MySQLBinlogNum
 
     @Override
     public Serializable handle(final Serializable value) {
-        if (null == value) {
-            return null;
-        }
         int intValue = (int) value;
         return intValue < 0 ? MEDIUMINT_MODULO + intValue : intValue;
     }
-    
-    @Override
-    public String getType() {
-        return "MEDIUMINT UNSIGNED";
-    }
 }

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/number/impl/MySQLBinlogUnsignedSmallintHandler.java

@@ -30,15 +30,7 @@ public final class MySQLBinlogUnsignedSmallintHandler implements MySQLBinlogNumb
 
     @Override
     public Serializable handle(final Serializable value) {
-        if (null == value) {
-            return null;
-        }
         short shortValue = (short) value;
         return shortValue < 0 ? SMALLINT_MODULO + shortValue : shortValue;
     }
-    
-    @Override
-    public String getType() {
-        return "SMALLINT UNSIGNED";
-    }
 }

kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/dumper/type/number/impl/MySQLBinlogUnsignedTinyintHandler.java

@@ -30,15 +30,7 @@ public final class MySQLBinlogUnsignedTinyintHandler implements MySQLBinlogNumbe
 
     @Override
     public Serializable handle(final Serializable value) {
-        if (null == value) {
-            return null;
-        }
         byte byteValue = (byte) value;
         return byteValue < 0 ? TINYINT_MODULO + byteValue : byteValue;
     }
-    
-    @Override
-    public String getType() {
-        return "TINYINT UNSIGNED";
-    }
 }