DBZ-6637 Support Geo types without srid

debezium · Jul 12, 2023 · 37fcf91 · 37fcf91
1 parent 38c3296
commit 37fcf91
Show file tree

Hide file tree

Showing 32 changed files with 114 additions and 64 deletions.
diff --git a/src/main/java/io/debezium/connector/jdbc/JdbcChangeEventSink.java b/src/main/java/io/debezium/connector/jdbc/JdbcChangeEventSink.java
@@ -9,6 +9,7 @@
 
 import java.sql.SQLException;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
 
 import org.apache.kafka.connect.data.Struct;
@@ -310,18 +311,17 @@ private void writeDelete(String sql, SinkRecordDescriptor record) throws SQLExce
     }
 
     private int bindKeyValuesToQuery(SinkRecordDescriptor record, NativeQuery<?> query, int index) {
-        switch (config.getPrimaryKeyMode()) {
-            case KAFKA:
-                query.setParameter(index++, record.getTopicName());
-                query.setParameter(index++, record.getPartition());
-                query.setParameter(index++, record.getOffset());
-                break;
-            default:
-                final Struct keySource = record.getKeyStruct(config.getPrimaryKeyMode());
-                if (keySource != null) {
-                    index = bindFieldValuesToQuery(record, query, index, keySource, record.getKeyFieldNames());
-                }
-                break;
+
+        if (Objects.requireNonNull(config.getPrimaryKeyMode()) == JdbcSinkConnectorConfig.PrimaryKeyMode.KAFKA) {
+            query.setParameter(index++, record.getTopicName());
+            query.setParameter(index++, record.getPartition());
+            query.setParameter(index++, record.getOffset());
+        }
+        else {
+            final Struct keySource = record.getKeyStruct(config.getPrimaryKeyMode());
+            if (keySource != null) {
+                index = bindFieldValuesToQuery(record, query, index, keySource, record.getKeyFieldNames());
+            }
         }
         return index;
     }

diff --git a/src/main/java/io/debezium/connector/jdbc/SinkRecordDescriptor.java b/src/main/java/io/debezium/connector/jdbc/SinkRecordDescriptor.java
@@ -208,9 +208,9 @@ public String getTypeName() {
             return typeName;
         }
 
-        public String getQueryBinding(ColumnDescriptor column) {
+        public String getQueryBinding(ColumnDescriptor column, Object value) {
             if (queryBinding == null) {
-                queryBinding = type.getQueryBinding(column, schema);
+                queryBinding = type.getQueryBinding(column, schema, value);
             }
             return queryBinding;
         }

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/GeneralDatabaseDialect.java b/src/main/java/io/debezium/connector/jdbc/dialect/GeneralDatabaseDialect.java
@@ -24,6 +24,7 @@
 import java.util.Set;
 
 import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.hibernate.SessionFactory;
 import org.hibernate.StatelessSession;
@@ -621,7 +622,41 @@ protected void addColumnDefaultValue(FieldDescriptor field, StringBuilder column
 
     protected String columnQueryBindingFromField(String fieldName, TableDescriptor table, SinkRecordDescriptor record) {
         final String columnName = columnNameFromField(fieldName, record);
-        return record.getFields().get(fieldName).getQueryBinding(table.getColumnByName(columnName));
+
+        if (record.getNonKeyFieldNames().contains(columnName)) {
+            Struct source = record.getAfterStruct();
+            return record.getFields().get(fieldName).getQueryBinding(table.getColumnByName(columnName), source.get(fieldName));
+        }
+
+        Object value = getValueFromKeyField(fieldName, record, columnName);
+        return record.getFields().get(fieldName).getQueryBinding(table.getColumnByName(columnName), value);
+    }
+
+    private Object getValueFromKeyField(String fieldName, SinkRecordDescriptor record, String columnName) {
+
+        Object value;
+        if (connectorConfig.getPrimaryKeyMode() == JdbcSinkConnectorConfig.PrimaryKeyMode.KAFKA) {
+            value = getValueForColumn(columnName, record);
+        }
+        else {
+            final Struct source = record.getKeyStruct(connectorConfig.getPrimaryKeyMode());
+            value = source.get(fieldName);
+        }
+        return value;
+    }
+
+    private Object getValueForColumn(String columnName, SinkRecordDescriptor record) {
+
+        switch (columnName) {
+            case "__connect_topic":
+                return record.getTopicName();
+            case "__connect_partition":
+                return record.getPartition();
+            case "__connect_offset":
+                return record.getOffset();
+            default:
+                return null;
+        }
     }
 
     protected String columnNameFromField(String fieldName, SinkRecordDescriptor record) {
@@ -673,7 +708,7 @@ protected String getQualifiedTableName(TableId tableId) {
     private String columnNameEqualsBinding(String fieldName, TableDescriptor table, SinkRecordDescriptor record) {
         final ColumnDescriptor column = table.getColumnByName(columnNameFromField(fieldName, record));
         final FieldDescriptor field = record.getFields().get(fieldName);
-        return toIdentifier(columnNamingStrategy.resolveColumnName(fieldName)) + "=" + field.getQueryBinding(column);
+        return toIdentifier(columnNamingStrategy.resolveColumnName(fieldName)) + "=" + field.getQueryBinding(column, record.getAfterStruct());
     }
 
     private static boolean isColumnNullable(String columnName, Collection<String> primaryKeyColumnNames, int nullability) {

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/mysql/GeometryType.java b/src/main/java/io/debezium/connector/jdbc/dialect/mysql/GeometryType.java
@@ -18,8 +18,8 @@ public class GeometryType extends AbstractGeoType {
     public static final Type INSTANCE = new GeometryType();
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
-        return "ST_GeomFromWKB(?, ?)";
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
+        return value == null ? "?" : "ST_GeomFromWKB(?, ?)";
     }
 
     @Override

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/mysql/JsonType.java b/src/main/java/io/debezium/connector/jdbc/dialect/mysql/JsonType.java
@@ -28,7 +28,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as json)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/mysql/MapToJsonType.java b/src/main/java/io/debezium/connector/jdbc/dialect/mysql/MapToJsonType.java
@@ -32,8 +32,8 @@ class MapToJsonType extends AbstractConnectMapType {
     public static final MapToJsonType INSTANCE = new MapToJsonType();
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
-        return JsonType.INSTANCE.getQueryBinding(column, schema);
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
+        return JsonType.INSTANCE.getQueryBinding(column, schema, value);
     }
 
     @Override

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/BitType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/BitType.java
@@ -34,7 +34,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         if (isBitOne(schema)) {
             final Optional<String> columnType = getSourceColumnType(schema);
             if (columnType.isPresent() && "BIT".equals(columnType.get())) {

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/CaseInsensitiveTextType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/CaseInsensitiveTextType.java
@@ -27,7 +27,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as citext)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/CidrType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/CidrType.java
@@ -27,7 +27,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as cidr)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/GeometryType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/GeometryType.java
@@ -30,8 +30,8 @@ public void configure(JdbcSinkConnectorConfig config, DatabaseDialect dialect) {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
-        return String.format(GEO_FROM_WKB_FUNCTION, postgisSchema);
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
+        return value == null ? "?" : String.format(GEO_FROM_WKB_FUNCTION, postgisSchema);
     }
 
     @Override

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/InetType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/InetType.java
@@ -27,7 +27,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as inet)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/IntervalType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/IntervalType.java
@@ -29,7 +29,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as interval)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/JsonType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/JsonType.java
@@ -29,7 +29,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         if (isHstore(schema)) {
             return "cast(? as hstore)";
             // return super.getQueryBinding(schema);

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/LtreeType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/LtreeType.java
@@ -27,7 +27,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as ltree)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/MacAddressType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/MacAddressType.java
@@ -27,7 +27,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as macaddr)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/MapToHstoreType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/MapToHstoreType.java
@@ -26,7 +26,7 @@ class MapToHstoreType extends AbstractConnectMapType {
     public static final MapToHstoreType INSTANCE = new MapToHstoreType();
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as hstore)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/MoneyType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/MoneyType.java
@@ -27,7 +27,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as money)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/PointType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/PointType.java
@@ -26,8 +26,8 @@ class PointType extends GeometryType {
     private static final String GEO_FROM_WKB_FUNCTION_AS_POINT = "cast(" + GEO_FROM_WKB_FUNCTION + " as point)";
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
-        return String.format(GEO_FROM_WKB_FUNCTION_AS_POINT, postgisSchema);
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
+        return value == null ? "?" : String.format(GEO_FROM_WKB_FUNCTION_AS_POINT, postgisSchema);
     }
 
     @Override

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/RangeType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/RangeType.java
@@ -29,7 +29,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as " + getSourceColumnType(schema).orElseThrow() + ")";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/UuidType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/UuidType.java
@@ -28,7 +28,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as uuid)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/postgres/XmlType.java b/src/main/java/io/debezium/connector/jdbc/dialect/postgres/XmlType.java
@@ -28,7 +28,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as xml)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/sqlserver/BitType.java b/src/main/java/io/debezium/connector/jdbc/dialect/sqlserver/BitType.java
@@ -34,7 +34,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         if (Bits.LOGICAL_NAME.equals(schema.name())) {
             final int bitSize = Integer.parseInt(schema.parameters().get(Bits.LENGTH_FIELD));
             return String.format("cast(? as %s)", bitSize > 1 ? String.format("varbinary(%d)", bitSize) : "bit");

diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/sqlserver/XmlType.java b/src/main/java/io/debezium/connector/jdbc/dialect/sqlserver/XmlType.java
@@ -28,7 +28,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return "cast(? as xml)";
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/type/AbstractGeoType.java b/src/main/java/io/debezium/connector/jdbc/type/AbstractGeoType.java
@@ -5,7 +5,7 @@
  */
 package io.debezium.connector.jdbc.type;
 
-import java.util.Base64;
+import java.util.Optional;
 
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Struct;
@@ -24,10 +24,13 @@ public int bind(Query<?> query, int index, Schema schema, Object value) {
         }
 
         if (value instanceof Struct) {
-            final int srid = ((Struct) value).getInt32(SRID);
+            // Default srid is 0 for both
+            // MySQL https://dev.mysql.com/doc/refman/8.0/en/spatial-reference-systems.html#:~:text=The%20SRS%20denoted%20in%20MySQL,for%20spatial%20data%20in%20MySQL.
+            // PostgreSQL https://postgis.net/docs/using_postgis_dbmanagement.html#spatial_ref_sys_table
+            final Integer srid = Optional.ofNullable(((Struct) value).getInt32(SRID)).orElse(0);
             final byte[] wkb = ((Struct) value).getBytes(WKB);
 
-            query.setParameter(index, Base64.getDecoder().decode(wkb));
+            query.setParameter(index, wkb);
             query.setParameter(index + 1, srid);
             return 2;
         }

diff --git a/src/main/java/io/debezium/connector/jdbc/type/AbstractType.java b/src/main/java/io/debezium/connector/jdbc/type/AbstractType.java
@@ -35,7 +35,7 @@ public void configure(JdbcSinkConnectorConfig config, DatabaseDialect dialect) {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return getDialect().getQueryBindingWithValueCast(column, schema, this);
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/type/Type.java b/src/main/java/io/debezium/connector/jdbc/type/Type.java
@@ -55,9 +55,10 @@ public interface Type {
      *
      * @param column column descriptor in the table relational model, never {@code null}
      * @param schema field schema, never {@code null}
+     * @param value value to be bound, may be {@code null}
      * @return query parameter argument binding SQL fragment
      */
-    String getQueryBinding(ColumnDescriptor column, Schema schema);
+    String getQueryBinding(ColumnDescriptor column, Schema schema, Object value);
 
     /**
      * Resolve the default value clause value.

diff --git a/src/main/java/io/debezium/connector/jdbc/type/connect/ConnectMapToConnectStringType.java b/src/main/java/io/debezium/connector/jdbc/type/connect/ConnectMapToConnectStringType.java
@@ -25,8 +25,8 @@ public class ConnectMapToConnectStringType extends AbstractConnectMapType {
     public static final ConnectMapToConnectStringType INSTANCE = new ConnectMapToConnectStringType();
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
-        return ConnectStringType.INSTANCE.getQueryBinding(column, schema);
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
+        return ConnectStringType.INSTANCE.getQueryBinding(column, schema, value);
     }
 
     @Override

diff --git a/src/main/java/io/debezium/connector/jdbc/type/connect/ConnectTimeType.java b/src/main/java/io/debezium/connector/jdbc/type/connect/ConnectTimeType.java
@@ -35,7 +35,7 @@ public String[] getRegistrationKeys() {
     }
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return getDialect().getTimeQueryBinding();
     }
 

diff --git a/src/main/java/io/debezium/connector/jdbc/type/debezium/AbstractDebeziumTimeType.java b/src/main/java/io/debezium/connector/jdbc/type/debezium/AbstractDebeziumTimeType.java
@@ -24,7 +24,7 @@
 public abstract class AbstractDebeziumTimeType extends AbstractTimeType {
 
     @Override
-    public String getQueryBinding(ColumnDescriptor column, Schema schema) {
+    public String getQueryBinding(ColumnDescriptor column, Schema schema, Object value) {
         return getDialect().getTimeQueryBinding();
     }