Develop

connectors-common/connector-core/src/main/java/io/tapdata/kit/DbKit.java

@@ -96,6 +96,16 @@ public static List<String> getColumnTypesFromResultSet(ResultSet resultSet) thro
         return columnTypeNames;
     }
 
+    public static List<Integer> getColumnTypeNumbersFromResultSet(ResultSet resultSet) throws SQLException {
+        //get all column typeNames
+        List<Integer> columnTypeNumbers = new ArrayList<>();
+        ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+        for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+            columnTypeNumbers.add(resultSetMetaData.getColumnType(i));
+        }
+        return columnTypeNumbers;
+    }
+
     public static List<Object> getDataArrayByColumnName(ResultSet resultSet, String columnName) throws SQLException {
         List<Object> list = TapSimplify.list();
         while (resultSet.next()) {

connectors-common/connector-core/src/main/java/io/tapdata/util/NetUtil.java

@@ -3,8 +3,8 @@
 import io.tapdata.kit.EmptyKit;
 
 import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.Socket;
+import java.net.*;
+import java.util.Enumeration;
 
 /**
  * @author samuel
@@ -41,4 +41,27 @@ public static void validateHostPortWithSocket(String host, int port, int timeout
     public static void validateHostPortWithSocket(String host, int port) throws IOException, IllegalArgumentException {
         validateHostPortWithSocket(host, port, DEFAULT_SOCKET_TIMEOUT_MS);
     }
+
+    public static String getLocalIP() {
+        try {
+            Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
+            while (interfaces.hasMoreElements()) {
+                NetworkInterface ni = interfaces.nextElement();
+                // 跳过回环接口、未启用或虚拟网卡
+                if (ni.isLoopback() || !ni.isUp() || ni.isVirtual()) continue;
+
+                Enumeration<InetAddress> addresses = ni.getInetAddresses();
+                while (addresses.hasMoreElements()) {
+                    InetAddress addr = addresses.nextElement();
+                    // 只取 IPv4 地址，跳过 IPv6
+                    if (addr instanceof Inet4Address && !addr.isLoopbackAddress()) {
+                        return addr.getHostAddress();
+                    }
+                }
+            }
+            return InetAddress.getLocalHost().getHostAddress();
+        } catch (Exception e) {
+            return "127.0.0.1";
+        }
+    }
 }

connectors-common/connector-core/src/test/java/io/tapdata/base/StringKitTest.java

@@ -16,4 +16,14 @@ void testRemoveParentheses() {
         result = StringKit.removeParentheses("TIMESTAMP(6) WITH TIME ZONE");
         assertEquals("TIMESTAMP WITH TIME ZONE", result);
     }
+
+    @Test
+    void testRemoveSqlNote() {
+        String result = StringKit.removeSqlNote("-----注释\n" +
+                "alter table\n" +
+                "    ----加注释\n" +
+                "    xxx add /*又来注释*/ aaa int;\n");
+        assertEquals("alter table\n" +
+                "        xxx add  aaa int;", result);
+    }
 }

connectors-common/debezium-bucket/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/RowDeserializers.java

@@ -322,33 +322,34 @@ protected static Serializable deserializeTimeV2(int meta, ByteArrayInputStream i
          * (3 bytes in total)
          *
          * + fractional-seconds storage (size depends on meta)
+         *  * The fractional part:
+         * read 1 byte, if meta is 1 or 2
+         * read 2 bytes, if meta is 3 or 4
+         * read 3 bytes, if meta is 5 or 6
          */
-        long time = bigEndianLong(inputStream.read(3), 0, 3);
-        boolean is_negative = bitSlice(time, 0, 1, 24) == 0;
-        int hours = bitSlice(time, 2, 10, 24);
-        int minutes = bitSlice(time, 12, 6, 24);
-        int seconds = bitSlice(time, 18, 6, 24);
-        int nanoSeconds;
-        if (is_negative) { // mysql binary arithmetic for negative encoded values
-            hours = ~hours & MASK_10_BITS;
-            hours = hours & ~(1 << 10); // unset sign bit
-            minutes = ~minutes & MASK_6_BITS;
-            minutes = minutes & ~(1 << 6); // unset sign bit
-            seconds = ~seconds & MASK_6_BITS;
-            seconds = seconds & ~(1 << 6); // unset sign bit
-            nanoSeconds = deserializeFractionalSecondsInNanosNegative(meta, inputStream);
-            if (nanoSeconds == 0 && seconds < 59) { // weird java Duration behavior
-                ++seconds;
-            }
-            hours = -hours;
-            minutes = -minutes;
-            seconds = -seconds;
-            nanoSeconds = -nanoSeconds;
-        }
-        else {
-            nanoSeconds = deserializeFractionalSecondsInNanos(meta, inputStream);
-        }
-        return Duration.ofHours(hours).plusMinutes(minutes).plusSeconds(seconds).plusNanos(nanoSeconds);
+        int fractionBytes = (meta + 1) / 2;
+        int payloadBytes = 3 + fractionBytes;
+        int payloadBits = payloadBytes * 8;
+        long time = bigEndianLong(inputStream.read(payloadBytes), 0, payloadBytes);
+        boolean is_negative = bitSlice(time, 0, 1, payloadBits) == 0;
+
+        if (is_negative) {
+            /*
+             * Negative numbers are stored in two's complement form.
+             * To get the positive value of a negative number in two's complement form,
+             * we should invert the bits of the number and add 1 to the result.
+             * Then we can take the number from the corresponding bits on the final result.
+             */
+            time = ~time + 1;
+        }
+
+        int hours = bitSlice(time, 2, 10, payloadBits);
+        int minutes = bitSlice(time, 12, 6, payloadBits);
+        int seconds = bitSlice(time, 18, 6, payloadBits);
+        int fraction = bitSlice(time, 24, fractionBytes * 8, payloadBits);
+        long nanoSeconds = (long) (fraction / (0.0000001 * Math.pow(100, fractionBytes - 1)));
+        final Duration duration = Duration.ofHours(hours).plusMinutes(minutes).plusSeconds(seconds).plusNanos(nanoSeconds);
+        return is_negative && !duration.isNegative() ? duration.negated() : duration;
     }
 
     /**

connectors-common/sql-core/pom.xml

@@ -151,6 +151,26 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
+                <executions>
+                    <execution>
+                        <id>generate-log4jdbc-properties</id>
+                        <phase>process-resources</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <target>
+                                <echo file="${project.build.outputDirectory}/log4jdbc.log4j2.properties">
+                                    log4jdbc.spylogdelegator.name = io.tapdata.common.log.CustomLogDelegator</echo>
+                            </target>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 </project>