fix(source-postgres): Handle numeric interval values for PostgreSQL 16.x CDC compatibility

Co-Authored-By: sophie.cui@airbyte.io <sophie.cui@airbyte.io>

Author: devin-ai-integration[bot]

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/cdc/PostgresConverter.java

@@ -292,7 +292,14 @@ private void registerDate(final RelationalColumn field, final ConverterRegistrat
         case "TIME":
           return resolveTime(field, x);
         case "INTERVAL":
-          return convertInterval((PGInterval) x);
+          if (x instanceof PGInterval) {
+            return convertInterval((PGInterval) x);
+          } else if (x instanceof Number) {
+            return convertIntervalFromMicros(((Number) x).longValue());
+          } else {
+            LOGGER.warn("Unexpected INTERVAL value type: {}", x.getClass());
+            return x.toString();
+          }
         default:
           throw new IllegalArgumentException("Unknown field type  " + fieldType.toUpperCase(Locale.ROOT));
       }
@@ -323,6 +330,25 @@ private String convertInterval(final PGInterval pgInterval) {
     return resultInterval.toString();
   }
 
+  private String convertIntervalFromMicros(final long micros) {
+    final boolean negative = micros < 0;
+    final long absMicros = Math.abs(micros);
+
+    final long totalSeconds = absMicros / 1_000_000L;
+    final int days = (int) (totalSeconds / 86_400L);
+    final int hours = (int) ((totalSeconds % 86_400L) / 3_600L);
+    final int minutes = (int) ((totalSeconds % 3_600L) / 60L);
+    final int seconds = (int) (totalSeconds % 60L);
+
+    final PGInterval interval = new PGInterval(0, 0,
+        negative ? -days : days,
+        negative ? -hours : hours,
+        negative ? -minutes : minutes,
+        negative ? -seconds : seconds);
+
+    return convertInterval(interval);
+  }
+
   private void registerMoney(final RelationalColumn field, final ConverterRegistration<SchemaBuilder> registration) {
     registration.register(SchemaBuilder.string().optional(), x -> {
       if (x == null) {