Fix ARRAY NOT NULL JDBC issue (#189)

Author: jogrogan

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroConverter.java

@@ -157,7 +157,9 @@ public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory) {
 
   /** Converts Avro Schema to RelDataType.
    * Nullability is preserved except for array types, JDBC is incapable of interpreting e.g. "FLOAT NOT NULL ARRAY"
-   * causing "NOT NULL" arrays to get demoted to "ANY ARRAY" which is not desired.
+   * causing "NOT NULL" arrays to get demoted to "ANY ARRAY" which is not desired. See HoptimatorArraySqlType for
+   * more details.
+   *
    * TODO: default field values are lost when converting from Avro to RelDataType
    */
   public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory, boolean nullable) {
@@ -187,8 +189,7 @@ public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory, boo
       case BOOLEAN:
         return createRelType(typeFactory, SqlTypeName.BOOLEAN, nullable);
       case ARRAY:
-        return typeFactory.createTypeWithNullability(
-            typeFactory.createArrayType(rel(schema.getElementType(), typeFactory, true), -1), nullable);
+        return new HoptimatorArraySqlType(rel(schema.getElementType(), typeFactory, true), nullable);
       case MAP:
         return typeFactory.createTypeWithNullability(
             typeFactory.createMapType(typeFactory.createSqlType(SqlTypeName.VARCHAR), rel(schema.getValueType(), typeFactory, nullable)), nullable);

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/HoptimatorArraySqlType.java

@@ -0,0 +1,27 @@
+package com.linkedin.hoptimator.avro;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.type.ArraySqlType;
+import org.checkerframework.checker.initialization.qual.UnknownInitialization;
+
+
+// Custom ArraySqlType to fix digest compatibility issue with Calcite JdbcSchema.
+// JdbcSchema expects row types digest format to match something like "INTEGER ARRAY". JdbcSchema is incapable of parsing
+// the array type if the digest format is like "INTEGER NOT NULL ARRAY" or "INTEGER ARRAY NOT NULL".
+// Nullability for the inner element is not supported by Calcite at all but nullability for the array itself is supported
+// and obtained via the isNullable() method instead.
+// Ideally we should be doing this for all SqlTypes but Calcite JdbcSchema is currently only checking ArraySqlType digest format.
+public class HoptimatorArraySqlType extends ArraySqlType {
+
+  public HoptimatorArraySqlType(RelDataType elementType, boolean isNullable) {
+    super(elementType, isNullable);
+  }
+
+  @Override
+  @SuppressWarnings("method.invocation.invalid")
+  protected void computeDigest(@UnknownInitialization HoptimatorArraySqlType this) {
+    StringBuilder sb = new StringBuilder();
+    generateTypeString(sb, true);
+    digest = sb.toString();
+  }
+}

hoptimator-avro/src/test/java/com/linkedin/hoptimator/avro/AvroConverterTest.java

@@ -178,22 +178,18 @@ public void testAvroKeyPayloadSchemaPrimitiveKey() {
 
   @Test
   public void convertsNestedArray() {
-    // Create a RelDataType with an array of structs
-    RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
-
-    // Create a struct type for array elements
-    RelDataTypeFactory.Builder elementBuilder = new RelDataTypeFactory.Builder(typeFactory);
-    elementBuilder.add("field1", typeFactory.createSqlType(SqlTypeName.VARCHAR));
-    elementBuilder.add("field2", typeFactory.createSqlType(SqlTypeName.INTEGER));
-    RelDataType structType = elementBuilder.build();
-
-    // Create array of structs type
-    RelDataType arrayOfStructsType = typeFactory.createArrayType(structType, -1);
-
-    // Test with a struct containing an array of structs field
-    RelDataTypeFactory.Builder containerBuilder = new RelDataTypeFactory.Builder(typeFactory);
-    containerBuilder.add("arrayOfStructsField", arrayOfStructsType);
-    RelDataType containerType = containerBuilder.build();
+    String schemaString =
+        "{\"type\":\"record\",\"name\":\"record\",\"namespace\":\"ns\",\"fields\":["
+            + "{\"name\":\"arrayOfStructsField\",\"type\":{\"type\":\"array\",\"items\":{\"type\":\"record\",\"name\":\"record_event1\",\"fields\":["
+            + "{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}}}]}";
+    Schema avroSchema = (new Schema.Parser()).parse(schemaString);
+    RelDataType containerType = AvroConverter.rel(avroSchema); // Ensure parsing works
+    // Calcite does not support inner array fields as NOT NULL.
+    assertEquals(1, containerType.getFieldList().size());
+    assertFalse(containerType.getFieldList().get(0).getType().isNullable());
+    assertEquals(2, containerType.getFieldList().get(0).getType().getComponentType().getFieldList().size());
+    assertTrue(containerType.getFieldList().get(0).getType().getComponentType().getFieldList().get(0).getType().isNullable());
+    assertEquals("RecordType(RecordType(VARCHAR field1, INTEGER field2) ARRAY arrayOfStructsField) NOT NULL", containerType.getFullTypeString());
 
     Schema containerSchema = AvroConverter.avro("test", "Record", containerType);
     assertNotNull(containerSchema);
@@ -204,10 +200,21 @@ public void convertsNestedArray() {
     assertEquals(Schema.Type.ARRAY, arrayFieldSchema.getType());
 
     Schema structElementSchema = arrayFieldSchema.getElementType();
-    assertEquals(Schema.Type.RECORD, structElementSchema.getType());
-    assertEquals(2, structElementSchema.getFields().size());
-    assertEquals("field1", structElementSchema.getFields().get(0).name());
-    assertEquals("field2", structElementSchema.getFields().get(1).name());
+    assertEquals(Schema.Type.UNION, structElementSchema.getType());
+    assertEquals(2, structElementSchema.getTypes().size());
+    assertEquals(Schema.Type.NULL, structElementSchema.getTypes().get(0).getType());
+    assertEquals(Schema.Type.RECORD, structElementSchema.getTypes().get(1).getType());
+
+    Schema innerRecord = structElementSchema.getTypes().get(1);
+    assertEquals(2, innerRecord.getFields().size());
+    assertEquals("field1", innerRecord.getFields().get(0).name());
+    assertEquals("field2", innerRecord.getFields().get(1).name());
+    assertEquals(2, innerRecord.getFields().size());
+
+    Schema innermostField = innerRecord.getFields().get(0).schema();
+    assertEquals(2, innermostField.getTypes().size());
+    assertEquals(Schema.Type.NULL, innermostField.getTypes().get(0).getType());
+    assertEquals(Schema.Type.STRING, innermostField.getTypes().get(1).getType());
   }
 
   @Test