Fix mishandling of null types in output Avro schemas (#146)

Author: ryannedolan

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

@@ -69,7 +69,9 @@ public static Schema avro(String namespace, String name, RelDataType dataType) {
               dataType.isNullable());
         case UNKNOWN:
         case NULL:
-          return createAvroSchemaWithNullability(Schema.createUnion(Schema.create(Schema.Type.NULL)), true);
+          // We can't have a union of null and null ["null", "null"], nor a nested union ["null",["null"]],
+          // so we ignore nullability and just use ["null"] here.
+          return Schema.createUnion(Schema.create(Schema.Type.NULL));
         default:
           throw new UnsupportedOperationException("No support yet for " + dataType.getSqlTypeName().toString());
       }

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

@@ -81,6 +81,16 @@ public void convertsNestedUnionSchemas() {
     assertEquals(avroSchema4.toString(), avroSchema4.getFields().size(), rel1.getFieldCount());
   }
 
+  @Test
+  public void supportsNullTypes() {
+    RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+    RelDataType rel = typeFactory.createStructType(Collections.singletonList(typeFactory.createSqlType(SqlTypeName.NULL)),
+        Collections.singletonList("field1"));
+
+    Schema avroSchema = AvroConverter.avro("NS", "R", rel);
+    assertEquals(avroSchema.toString(), avroSchema.getFields().size(), rel.getFieldCount());
+  }
+
   @Test
   public void testAvroKeyPayloadSchemaNoKeyOptions() {
     RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);