Allow optional UDT attributes (#93)

* Allow optional UDT attributes

* Increase CI action timeout to 90 min

Author: Ayman Khalil

.github/workflows/ci.yaml

@@ -17,7 +17,7 @@ on:
 jobs:
   build:
     runs-on: ubuntu-latest
-    timeout-minutes: 60
+    timeout-minutes: 90
     name: Build
     steps:
       - uses: actions/checkout@v2
@@ -33,7 +33,7 @@ jobs:
     needs: build
     name: Test
     runs-on: ubuntu-latest
-    timeout-minutes: 60
+    timeout-minutes: 90
     strategy:
       fail-fast: false
       matrix:

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

@@ -170,7 +170,7 @@ Schema dataTypeSchema(KeyspaceMetadata ksm, DataType dataType) {
             case ProtocolConstants.DataType.TIME:
                 return CqlLogicalTypes.timeMicrosType;
             case ProtocolConstants.DataType.UDT:
-                return buildUDTSchema(ksm, dataType.asCql(false, true), false);
+                return buildUDTSchema(ksm, dataType.asCql(false, true), true);
             case ProtocolConstants.DataType.UUID:
             case ProtocolConstants.DataType.TIMEUUID:
                 return CqlLogicalTypes.uuidType;

connector/src/test/java/com/datastax/oss/pulsar/source/PulsarCassandraSourceTests.java

@@ -455,15 +455,19 @@ public void testSchema(String ksName) throws InterruptedException, IOException {
                         dataSpecMap.get("map").cqlValue
                 );
 
+                // force udt values to be null by populating 1 item, using zudt.newValue() without explicitly setting
+                // any field to non-null value will cause the udt column itself to be null in the C* table
+                UdtValue zudtOptionalValues = zudt.newValue(dataSpecMap.get("text").cqlValue);
+
                 cqlSession.execute("CREATE TABLE IF NOT EXISTS " + ksName + ".table3 (" +
                         "xtext text, xascii ascii, xboolean boolean, xblob blob, xtimestamp timestamp, xtime time, xdate date, xuuid uuid, xtimeuuid timeuuid, xtinyint tinyint, xsmallint smallint, xint int, xbigint bigint, xvarint varint, xdecimal decimal, xdouble double, xfloat float, xinet4 inet, xinet6 inet, " +
-                        "ytext text, yascii ascii, yboolean boolean, yblob blob, ytimestamp timestamp, ytime time, ydate date, yuuid uuid, ytimeuuid timeuuid, ytinyint tinyint, ysmallint smallint, yint int, ybigint bigint, yvarint varint, ydecimal decimal, ydouble double, yfloat float, yinet4 inet, yinet6 inet, yduration duration, yudt zudt, ylist list<text>, yset set<int>, ymap map<text, double>, ylistofmap list<frozen<map<text,double>>>, ysetofudt set<frozen<zudt>>," +
+                        "ytext text, yascii ascii, yboolean boolean, yblob blob, ytimestamp timestamp, ytime time, ydate date, yuuid uuid, ytimeuuid timeuuid, ytinyint tinyint, ysmallint smallint, yint int, ybigint bigint, yvarint varint, ydecimal decimal, ydouble double, yfloat float, yinet4 inet, yinet6 inet, yduration duration, yudt zudt, yudtoptional zudt, ylist list<text>, yset set<int>, ymap map<text, double>, ylistofmap list<frozen<map<text,double>>>, ysetofudt set<frozen<zudt>>," +
                         "primary key (xtext, xascii, xboolean, xblob, xtimestamp, xtime, xdate, xuuid, xtimeuuid, xtinyint, xsmallint, xint, xbigint, xvarint, xdecimal, xdouble, xfloat, xinet4, xinet6)) " +
                         "WITH CLUSTERING ORDER BY (xascii ASC, xboolean DESC, xblob ASC, xtimestamp DESC, xtime DESC, xdate ASC, xuuid DESC, xtimeuuid ASC, xtinyint DESC, xsmallint ASC, xint DESC, xbigint ASC, xvarint DESC, xdecimal ASC, xdouble DESC, xfloat ASC, xinet4 ASC, xinet6 DESC) AND cdc=true");
                 cqlSession.execute("INSERT INTO " + ksName + ".table3 (" +
                                 "xtext, xascii, xboolean, xblob, xtimestamp, xtime, xdate, xuuid, xtimeuuid, xtinyint, xsmallint, xint, xbigint, xvarint, xdecimal, xdouble, xfloat, xinet4, xinet6, " +
-                                "ytext, yascii, yboolean, yblob, ytimestamp, ytime, ydate, yuuid, ytimeuuid, ytinyint, ysmallint, yint, ybigint, yvarint, ydecimal, ydouble, yfloat, yinet4, yinet6, yduration, yudt, ylist, yset, ymap, ylistofmap, ysetofudt" +
-                                ") VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, ?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, ?,?, ?,?,?,?,?)",
+                                "ytext, yascii, yboolean, yblob, ytimestamp, ytime, ydate, yuuid, ytimeuuid, ytinyint, ysmallint, yint, ybigint, yvarint, ydecimal, ydouble, yfloat, yinet4, yinet6, yduration, yudt, yudtoptional, ylist, yset, ymap, ylistofmap, ysetofudt" +
+                                ") VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, ?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, ?,?,?, ?,?,?,?,?)",
                         dataSpecMap.get("text").cqlValue,
                         dataSpecMap.get("ascii").cqlValue,
                         dataSpecMap.get("boolean").cqlValue,
@@ -506,6 +510,7 @@ public void testSchema(String ksName) throws InterruptedException, IOException {
 
                         dataSpecMap.get("duration").cqlValue,
                         zudtValue,
+                        zudtOptionalValues,
 
                         dataSpecMap.get("list").cqlValue,
                         dataSpecMap.get("set").cqlValue,
@@ -615,7 +620,7 @@ void assertGenericArray(String field, GenericArray ga) {
 
     void assertField(String fieldName, Object value) {
         String vKey = fieldName.substring(1);
-        if (!vKey.equals("udt") && ! vKey.equals("setofudt")) {
+        if (!vKey.equals("udt") && !vKey.equals("udtoptional") && ! vKey.equals("setofudt")) {
             Assert.assertTrue("Unknown field " + vKey, dataSpecMap.containsKey(vKey));
         }
         if (value instanceof GenericRecord) {
@@ -672,6 +677,17 @@ void assertGenericRecords(String field, GenericRecord gr) {
                 }
             }
             return;
+            case "udtoptional": {
+                for (Field f : gr.getFields()) {
+                    if (f.getName().equals("ztext")){
+                        assertField(f.getName(), gr.getField(f.getName()));
+                    }
+                    else {
+                        assertNull(gr.getField(f.getName()));
+                    }
+                }
+            }
+            return;
         }
         Assert.assertTrue("Unexpected field="+field, false);
     }
@@ -781,6 +797,17 @@ void assertJsonNode(String field, JsonNode node) {
                 }
             }
             return;
+            case "udtoptional": {
+                for (Iterator<Map.Entry<String, JsonNode>> it = node.fields(); it.hasNext(); ) {
+                    Map.Entry<String, JsonNode> f = it.next();
+                    if (f.getKey().equals("ztext")) {
+                        assertField(f.getKey(), f.getValue());
+                    } else {
+                        assertNull(f.getValue());
+                    }
+                }
+            }
+            return;
         }
         Assert.assertTrue("Unexpected field="+field, false);
     }