Add VectorType support to CDC connector (#170)

* Add VectorType support to CDC connector

* Optimize imports

Author: aymkhalil

connector/src/main/java/com/datastax/oss/pulsar/source/converters/AbstractNativeConverter.java

@@ -22,6 +22,7 @@
 import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
+import com.datastax.oss.driver.api.core.type.CqlVectorType;
 import com.datastax.oss.driver.api.core.type.DataType;
 import com.datastax.oss.driver.api.core.type.ListType;
 import com.datastax.oss.driver.api.core.type.MapType;
@@ -74,6 +75,13 @@ public AbstractNativeConverter(KeyspaceMetadata ksm, TableMetadata tm, List<Colu
                             subSchemas.put(field.name(), collectionSchema);
                             log.info("Add collection schema {}={}", field.name(), collectionSchema);
                             break;
+                        case ProtocolConstants.DataType.CUSTOM:
+                            if (cm.getType() instanceof CqlVectorType) {
+                                Schema vectorSchema = dataTypeSchema(ksm, cm.getType());
+                                subSchemas.put(field.name(), vectorSchema);
+                                log.info("Add vector schema {}={}", field.name(), vectorSchema);
+                            }
+                            break;
                     }
                 }
             }
@@ -121,6 +129,8 @@ boolean isSupportedCqlType(DataType dataType) {
             case ProtocolConstants.DataType.SET:
             case ProtocolConstants.DataType.MAP:
                 return true;
+            case ProtocolConstants.DataType.CUSTOM:
+                return dataType instanceof CqlVectorType;
         }
         return false;
     }
@@ -189,6 +199,11 @@ Schema dataTypeSchema(KeyspaceMetadata ksm, DataType dataType) {
             case ProtocolConstants.DataType.MAP:
                 MapType mapType = (MapType) dataType;
                 return org.apache.avro.Schema.createMap(dataTypeSchema(ksm, mapType.getValueType()));
+            case ProtocolConstants.DataType.CUSTOM:
+                if (dataType instanceof CqlVectorType) {
+                    CqlVectorType vectorType = (CqlVectorType) dataType;
+                    return org.apache.avro.Schema.createArray(dataTypeSchema(ksm, vectorType.getSubtype()));
+                }
             default:
                 throw new UnsupportedOperationException("Ignoring unsupported type=" + dataType.asCql(false, true));
         }

connector/src/main/java/com/datastax/oss/pulsar/source/converters/NativeAvroConverter.java

@@ -20,10 +20,12 @@
 import com.datastax.oss.driver.api.core.cql.ColumnDefinition;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.data.CqlDuration;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 import com.datastax.oss.driver.api.core.data.UdtValue;
 import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
+import com.datastax.oss.driver.api.core.type.CqlVectorType;
 import com.datastax.oss.driver.api.core.type.DataType;
 import com.datastax.oss.driver.api.core.type.ListType;
 import com.datastax.oss.driver.api.core.type.MapType;
@@ -168,6 +170,17 @@ public byte[] toConnectData(Row row) {
                         genericRecordBuilder.put(fieldName, mapValue);
                     }
                     break;
+                    case ProtocolConstants.DataType.CUSTOM: {
+                        if (cm.getType() instanceof CqlVectorType) {
+                            Schema vectorSchema = subSchemas.get(fieldName);
+                            CqlVector<?> vector = row.getCqlVector(fieldName);
+                            log.debug("field={} listSchema={} listValue={}", fieldName, vectorSchema, vector);
+                            List<Object> vectorValue = new ArrayList<>();
+                            vector.getValues().forEach(vectorValue::add);
+                            genericRecordBuilder.put(fieldName, buildArrayValue(vectorSchema, vectorValue));
+                        }
+                    }
+                    break;
                     default:
                         log.debug("Ignoring unsupported column name={} type={}", cm.getName(), cm.getType().asCql(false, true));
                 }

gradle.properties

@@ -9,7 +9,7 @@ releasesRepoUrl=https://repo.datastax.com/artifactory/datastax-public-releases-l
 # deps version
 avroVersion=1.10.2
 lombokVersion=1.18.20
-ossDriverVersion=4.15.0
+ossDriverVersion=4.16.0
 cassandra3Version=3.11.10
 cassandra4Version=4.0.4
 dse4Version=6.8.23