modify bitType value process for flink pipeline synchronize

Author: baisui1981

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/plugins/incr/flink/cdc/FlinkCol.java

@@ -200,6 +200,7 @@ public static BiFunction Byte() {
         return new ByteProcess();
     }
 
+
     public static BiFunction DateTime() {
         return new DateTimeProcess();
     }
@@ -244,6 +245,13 @@ public Object deApply(Object o) {
         }
     }
 
+    public static class PipelineBooleanProcess extends ByteProcess {
+        @Override
+        public Object apply(Object o) {
+            return (Boolean) (((Byte) super.apply(o)) > 0);
+        }
+    }
+
     public static class BoolProcess extends BiFunction {
         @Override
         public Object apply(Object o) {

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/BasicFlinkDataMapper.java

@@ -25,6 +25,7 @@
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCDCPipelineEventProcess.FlinkPipelineStringConvert;
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCol.DTOConvertTo;
+import com.qlangtech.plugins.incr.flink.cdc.FlinkCol.PipelineBooleanProcess;
 import com.qlangtech.plugins.incr.flink.cdc.RowFieldGetterFactory;
 import com.qlangtech.plugins.incr.flink.cdc.RowFieldGetterFactory.ByteGetter;
 import com.qlangtech.tis.async.message.client.consumer.IFlinkColCreator;
@@ -271,7 +272,7 @@ public FlinkCol bitType(DataType type) {
             return new FlinkCol(meta, type, DataTypes.TINYINT()
                     , FlinkCol.Byte()
                     , FlinkCol.NoOp()
-                    , new FlinkCDCPipelineEventProcess(org.apache.flink.cdc.common.types.DataTypes.BINARY(1), FlinkCol.Byte())
+                    , new FlinkCDCPipelineEventProcess(org.apache.flink.cdc.common.types.DataTypes.BOOLEAN(), new PipelineBooleanProcess())
                     , new ByteGetter(meta.getName(), colIndex));
         }