STREAM-770 Add support for Tuple CQL data type (#69)

* dependencies update

* STREAM-770: Add support for Tuple cql data type

* Remove provided and test scopes for org.apache.pulsar:pulsar-client-api to avoid warnings

* Revert dsbulk upgrade to 1.11.0 to see if it would run tests okay

* Redo dsbulk version upgrade

* Attempt to add more unit tests

* Tweak the nested tuple type unit tests

* Add support for nested tuple types scenario

---------

Co-authored-by: Madhavan <cxo@ibm.com>

Author: msmygit

.github/dependabot.yml

@@ -0,0 +1,10 @@
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"
+    groups:
+      github-actions:
+        patterns:
+          - "*"

.github/workflows/ci-integration.yml

@@ -18,32 +18,33 @@ jobs:
           - "datastax/lunastreaming:2.10_2.3"
           - "apachepulsar/pulsar:2.10.2"
         cassandra:
-          - cassandra_version: 3.0.27
-            ccm_version: 3.0.27
+          - cassandra_version: 3.0.32
+            ccm_version: 3.0.32
             is_dse: false
-          - cassandra_version: 3.11.13
-            ccm_version: 3.11.13
+          - cassandra_version: 3.11.19
+            ccm_version: 3.11.19
             is_dse: false
-          - cassandra_version: 4.0.4
-            ccm_version: 4.0.4
+          - cassandra_version: 4.0.19
+            ccm_version: 4.0.19
             is_dse: false
           - cassandra_version: dse-5.1
-            ccm_version: 5.1.31
-            is_dse: true
-          - cassandra_version: dse-6.0
-            ccm_version: 6.0.18
-            is_dse: true
-          - cassandra_version: dse-6.7
-            ccm_version: 6.7.17
+            ccm_version: 5.1.48
             is_dse: true
+          # DSE 6.0 and 6.7 are commented out because they are EOL and no longer supported by DataStax/IBM
+          # - cassandra_version: dse-6.0
+          #   ccm_version: 6.0.18
+          #   is_dse: true
+          # - cassandra_version: dse-6.7
+          #   ccm_version: 6.7.17
+          #   is_dse: true
           - cassandra_version: dse-6.8
-            ccm_version: 6.8.24
+            ccm_version: 6.8.61
             is_dse: true
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v6
       - name: Set up JDK 8
-        uses: actions/setup-java@v2
+        uses: actions/setup-java@v5
         with:
           java-version: 8
           distribution: 'temurin'
@@ -52,7 +53,7 @@ jobs:
           export JDK8_PATH=$JAVA_HOME
           echo "JDK8_PATH=$JDK8_PATH" >> $GITHUB_ENV
       - name: Set up JDK 11
-        uses: actions/setup-java@v2
+        uses: actions/setup-java@v5
         with:
           java-version: 11
           distribution: 'temurin'
@@ -67,7 +68,7 @@ jobs:
           psutil
           """ > requirements.txt
       - name: Set up Python
-        uses: MatteoH2O1999/setup-python@v1
+        uses: MatteoH2O1999/setup-python@v6
         with:
           python-version: '2.7'
           cache: 'pip'
@@ -91,7 +92,7 @@ jobs:
           #!/bin/bash
           pip install ccm
       - name: Cache local Maven repository
-        uses: actions/cache@v3
+        uses: actions/cache@v5
         with:
           path: ~/.m2/repository
           key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}

pom.xml

@@ -44,7 +44,7 @@
     <java.release.version>8</java.release.version>
     <caffeine.version>2.6.2</caffeine.version>
     <oss.driver.version>4.16.0</oss.driver.version>
-    <dsbulk.version>1.10.0</dsbulk.version>
+    <dsbulk.version>1.11.0</dsbulk.version>
     <reactive-streams.version>1.0.3</reactive-streams.version>
     <pulsar.version>2.8.2</pulsar.version>
     <slf4j.version>1.7.25</slf4j.version>

pulsar-impl/pom.xml

@@ -121,16 +121,6 @@
       <artifactId>pulsar-functions-api</artifactId>
       <scope>provided</scope>
     </dependency>
-    <dependency>
-      <groupId>org.apache.pulsar</groupId>
-      <artifactId>pulsar-client-api</artifactId>
-      <scope>provided</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.pulsar</groupId>
-      <artifactId>pulsar-client-api</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.apache.pulsar</groupId>
       <artifactId>pulsar-common</artifactId>
@@ -206,7 +196,7 @@
                 <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                   <resource>reference.conf</resource>
                 </transformer>
-                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
               </transformers>
               <relocations>
                 <relocation>

pulsar-impl/src/main/java/com/datastax/oss/pulsar/sink/codecs/PulsarCodecProvider.java

@@ -17,6 +17,7 @@
 
 import com.datastax.oss.common.sink.AbstractStruct;
 import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.TupleType;
 import com.datastax.oss.driver.api.core.type.UserDefinedType;
 import com.datastax.oss.driver.api.core.type.reflect.GenericType;
 import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
@@ -36,6 +37,14 @@ public class PulsarCodecProvider implements ConvertingCodecProvider {
       @NonNull GenericType<?> externalJavaType,
       @NonNull ConvertingCodecFactory codecFactory,
       boolean rootCodec) {
+    // Handle Tuple types
+    if (cqlType instanceof TupleType
+        && (externalJavaType.equals(GenericType.of(PulsarStruct.class))
+            || externalJavaType.equals(GenericType.of(AbstractStruct.class)))) {
+      return Optional.of(new StructToTupleCodec(codecFactory, (TupleType) cqlType));
+    }
+
+    // Handle UDT types
     if (cqlType instanceof UserDefinedType
         && (externalJavaType.equals(GenericType.of(PulsarStruct.class))
             || externalJavaType.equals(GenericType.of(AbstractStruct.class)))) {

pulsar-impl/src/main/java/com/datastax/oss/pulsar/sink/codecs/StructToTupleCodec.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.pulsar.sink.codecs;
+
+import com.datastax.oss.common.sink.AbstractField;
+import com.datastax.oss.common.sink.AbstractSchema;
+import com.datastax.oss.common.sink.record.StructDataMetadata;
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.TupleType;
+import com.datastax.oss.driver.api.core.type.reflect.GenericType;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodec;
+import com.datastax.oss.dsbulk.codecs.api.ConvertingCodecFactory;
+import com.datastax.oss.sink.pulsar.PulsarStruct;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Codec to convert a Pulsar Struct to a CQL Tuple. */
+public class StructToTupleCodec extends ConvertingCodec<PulsarStruct, TupleValue> {
+
+  private final ConvertingCodecFactory codecFactory;
+  private final TupleType tupleType;
+
+  StructToTupleCodec(ConvertingCodecFactory codecFactory, TupleType cqlType) {
+    super(codecFactory.getCodecRegistry().codecFor(cqlType), PulsarStruct.class);
+    this.codecFactory = codecFactory;
+    this.tupleType = cqlType;
+  }
+
+  @Override
+  public TupleValue externalToInternal(PulsarStruct external) {
+    if (external == null) {
+      return null;
+    }
+
+    List<DataType> componentTypes = tupleType.getComponentTypes();
+    int size = componentTypes.size();
+    AbstractSchema schema = external.schema();
+    StructDataMetadata structMetadata = new StructDataMetadata(schema);
+
+    Set<String> structFieldNames =
+        schema.fields().stream().map(AbstractField::name).collect(Collectors.toSet());
+
+    if (structFieldNames.size() != size) {
+      throw new IllegalArgumentException(
+          String.format("Expecting %d tuple fields, got %d", size, structFieldNames.size()));
+    }
+
+    TupleValue tupleValue = tupleType.newValue();
+
+    // Tuple fields are named index_0, index_1, etc. following CDC convention
+    for (int i = 0; i < size; i++) {
+      String fieldName = "index_" + i;
+      DataType componentType = componentTypes.get(i);
+
+      // Check if field exists in the struct
+      if (!schema.fields().stream().anyMatch(f -> f.name().equals(fieldName))) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Field %s not found in input struct for tuple component %d", fieldName, i));
+      }
+
+      @SuppressWarnings("unchecked")
+      GenericType<Object> fieldType =
+          (GenericType<Object>) structMetadata.getFieldType(fieldName, componentType);
+      ConvertingCodec<Object, Object> fieldCodec =
+          codecFactory.createConvertingCodec(componentType, fieldType, false);
+      Object fieldValue = external.get(fieldName);
+      Object convertedValue = fieldCodec.externalToInternal(fieldValue);
+      tupleValue = tupleValue.set(i, convertedValue, fieldCodec.getInternalJavaType());
+    }
+
+    return tupleValue;
+  }
+
+  @Override
+  public PulsarStruct internalToExternal(TupleValue internal) {
+    if (internal == null) {
+      return null;
+    }
+    throw new UnsupportedOperationException(
+        "This codec does not support converting from TupleValue to Struct");
+  }
+}

pulsar-impl/src/main/java/com/datastax/oss/sink/pulsar/AvroTypeUtil.java

@@ -17,6 +17,7 @@
 
 import static com.datastax.oss.sink.pulsar.CqlLogicalTypes.CQL_DECIMAL;
 import static com.datastax.oss.sink.pulsar.CqlLogicalTypes.CQL_DURATION;
+import static com.datastax.oss.sink.pulsar.CqlLogicalTypes.CQL_TUPLE;
 import static com.datastax.oss.sink.pulsar.CqlLogicalTypes.CQL_VARINT;
 import static com.datastax.oss.sink.pulsar.CqlLogicalTypes.DATE;
 import static com.datastax.oss.sink.pulsar.CqlLogicalTypes.TIMESTAMP_MILLIS;
@@ -220,11 +221,13 @@ public static void enableDecodeCDCDataTypes(boolean decodeCDCDataTypes) {
       LogicalTypes.register(CQL_DECIMAL, schema -> new CqlLogicalTypes.CqlDecimalLogicalType());
       LogicalTypes.register(CQL_DURATION, schema -> new CqlLogicalTypes.CqlDurationLogicalType());
       LogicalTypes.register(CQL_VARINT, schema -> new CqlLogicalTypes.CqlVarintLogicalType());
+      LogicalTypes.register(CQL_TUPLE, schema -> new CqlLogicalTypes.CqlTupleLogicalType());
 
       // Register logical type converters
       logicalTypeConverters.put(CQL_DECIMAL, new CqlLogicalTypes.CqlDecimalConversion());
       logicalTypeConverters.put(CQL_DURATION, new CqlLogicalTypes.CqlDurationConversion());
       logicalTypeConverters.put(CQL_VARINT, new CqlLogicalTypes.CqlVarintConversion());
+      logicalTypeConverters.put(CQL_TUPLE, new CqlLogicalTypes.CqlTupleConversion());
       logicalTypeConverters.put(DATE, new CqlLogicalTypes.DateConversion());
       logicalTypeConverters.put(TIME_MICROS, new CqlLogicalTypes.TimeConversion());
       logicalTypeConverters.put(TIMESTAMP_MILLIS, new CqlLogicalTypes.TimestampConversion());
@@ -235,11 +238,68 @@ public static void enableDecodeCDCDataTypes(boolean decodeCDCDataTypes) {
       logicalTypeConverters.remove(CQL_DECIMAL);
       logicalTypeConverters.remove(CQL_DURATION);
       logicalTypeConverters.remove(CQL_VARINT);
+      logicalTypeConverters.remove(CQL_TUPLE);
       logicalTypeConverters.remove(DATE);
       logicalTypeConverters.remove(TIME_MICROS);
       logicalTypeConverters.remove(TIMESTAMP_MILLIS);
     }
 
     AvroTypeUtil.decodeCDCDataTypes = decodeCDCDataTypes;
   }
+
+  /**
+   * Check if a GenericRecord represents a tuple by examining its schema name. Tuples follow the CDC
+   * naming convention: Tuple_<hashcode>
+   */
+  public static boolean isTupleRecord(GenericRecord record) {
+    if (record == null) {
+      return false;
+    }
+    Object nativeObject = record.getNativeObject();
+    if (nativeObject instanceof org.apache.avro.generic.GenericRecord) {
+      org.apache.avro.generic.GenericRecord avroRecord =
+          (org.apache.avro.generic.GenericRecord) nativeObject;
+      return avroRecord.getSchema().getName().startsWith("Tuple_");
+    }
+    return false;
+  }
+
+  /** Check if an Avro schema represents a tuple type. */
+  public static boolean isTupleSchema(Schema schema) {
+    if (schema == null) {
+      return false;
+    }
+    // Handle union types
+    if (schema.isUnion()) {
+      return schema
+          .getTypes()
+          .stream()
+          .anyMatch(s -> s.getType() == Schema.Type.RECORD && s.getName().startsWith("Tuple_"));
+    }
+    // Handle direct record types
+    return schema.getType() == Schema.Type.RECORD && schema.getName().startsWith("Tuple_");
+  }
+
+  /**
+   * Check if a schema contains tuple types in its structure. This checks maps, arrays, and nested
+   * structures.
+   */
+  public static boolean containsTuples(Schema schema) {
+    if (schema == null) {
+      return false;
+    }
+
+    switch (schema.getType()) {
+      case RECORD:
+        return schema.getName().startsWith("Tuple_");
+      case ARRAY:
+        return containsTuples(schema.getElementType());
+      case MAP:
+        return containsTuples(schema.getValueType());
+      case UNION:
+        return schema.getTypes().stream().anyMatch(AvroTypeUtil::containsTuples);
+      default:
+        return false;
+    }
+  }
 }

pulsar-impl/src/main/java/com/datastax/oss/sink/pulsar/CqlLogicalTypes.java

@@ -32,11 +32,31 @@ public class CqlLogicalTypes {
   public static final String CQL_VARINT = "cql_varint";
   public static final String CQL_DECIMAL = "cql_decimal";
   public static final String CQL_DURATION = "cql_duration";
+  public static final String CQL_TUPLE = "cql_tuple";
 
   public static final String DATE = LogicalTypes.date().getName();
   public static final String TIME_MICROS = LogicalTypes.timeMicros().getName();
   public static final String TIMESTAMP_MILLIS = LogicalTypes.timestampMillis().getName();
 
+  public static class CqlTupleConversion extends Conversion<IndexedRecord> {
+    @Override
+    public Class<IndexedRecord> getConvertedType() {
+      return IndexedRecord.class;
+    }
+
+    @Override
+    public String getLogicalTypeName() {
+      return CQL_TUPLE;
+    }
+
+    @Override
+    public IndexedRecord fromRecord(IndexedRecord value, Schema schema, LogicalType type) {
+      // For tuple support, we return the IndexedRecord as-is
+      // The actual conversion to TupleValue will be handled by StructToTupleCodec
+      return value;
+    }
+  }
+
   public static class CqlDurationConversion extends Conversion<CqlDuration> {
     @Override
     public Class<CqlDuration> getConvertedType() {
@@ -165,4 +185,10 @@ public CqlVarintLogicalType() {
       super(CQL_VARINT);
     }
   }
+
+  static class CqlTupleLogicalType extends LogicalType {
+    public CqlTupleLogicalType() {
+      super(CQL_TUPLE);
+    }
+  }
 }