SAMZA-2320: Samza-sql: Refactor validation to cover more cases and make it more extensible.

samza-sql/src/main/java/org/apache/samza/sql/avro/AvroTypeFactoryImpl.java

@@ -1,21 +1,21 @@
 /*
-* Licensed to the Apache Software Foundation (ASF) under one
-* or more contributor license agreements.  See the NOTICE file
-* distributed with this work for additional information
-* regarding copyright ownership.  The ASF licenses this file
-* to you 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.
-*/
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.samza.sql.avro;
 
@@ -31,6 +31,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 /**
  * Factory that creates {@link SqlSchema} from the Avro Schema. This is used by the
  * {@link AvroRelConverter} to convert Avro schema to Samza Sql schema.
@@ -44,37 +45,47 @@ public AvroTypeFactoryImpl() {
   }
 
   public SqlSchema createType(Schema schema) {
+    validateTopLevelAvroType(schema);
+    return convertSchema(schema.getFields(), true);
+  }
+
+  /**
+   * Given a schema field, determine if it is an optional field. There could be cases where a field
+   * is considered as optional even if it is marked as required in the schema. The producer could be filling in this
+   * field and hence need not be specified in the query and hence is optional. Typically, such fields are
+   * the top level fields in the schema.
+   * @param field schema field
+   * @param isTopLevelField if it is top level field in the schema
+   * @return if the field is optional
+   */
+  protected boolean isOptional(Schema.Field field, boolean isTopLevelField) {
+    return field.defaultValue() != null;
+  }
+
+  private void validateTopLevelAvroType(Schema schema) {
     Schema.Type type = schema.getType();
     if (type != Schema.Type.RECORD) {
       String msg =
-          String.format("System supports only RECORD as top level avro type, But the Schema's type is %s", type);
+          String.format("Samza Sql supports only RECORD as top level avro type, But the Schema's type is %s", type);
       LOG.error(msg);
       throw new SamzaException(msg);
     }
-
-    return convertSchema(schema.getFields());
   }
 
-  private SqlSchema convertSchema(List<Schema.Field> fields) {
-
+  private SqlSchema convertSchema(List<Schema.Field> fields, boolean isTopLevelField) {
     SqlSchemaBuilder schemaBuilder = SqlSchemaBuilder.builder();
     for (Schema.Field field : fields) {
-      boolean isOptional = (field.defaultValue() != null);
-      SqlFieldSchema fieldSchema = convertField(field.schema(), false, isOptional);
+      SqlFieldSchema fieldSchema = convertField(field.schema(), false, isOptional(field, isTopLevelField));
       schemaBuilder.addField(field.name(), fieldSchema);
     }
 
     return schemaBuilder.build();
   }
 
-  private SqlFieldSchema convertField(Schema fieldSchema) {
-    return convertField(fieldSchema, false, false);
-  }
-
   private SqlFieldSchema convertField(Schema fieldSchema, boolean isNullable, boolean isOptional) {
     switch (fieldSchema.getType()) {
       case ARRAY:
-        SqlFieldSchema elementSchema = convertField(fieldSchema.getElementType());
+        SqlFieldSchema elementSchema = convertField(fieldSchema.getElementType(), false, false);
         return SqlFieldSchema.createArraySchema(elementSchema, isNullable, isOptional);
       case BOOLEAN:
         return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BOOLEAN, isNullable, isOptional);
@@ -98,7 +109,7 @@ private SqlFieldSchema convertField(Schema fieldSchema, boolean isNullable, bool
       case LONG:
         return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.INT64, isNullable, isOptional);
       case RECORD:
-        SqlSchema rowSchema = convertSchema(fieldSchema.getFields());
+        SqlSchema rowSchema = convertSchema(fieldSchema.getFields(), false);
         return SqlFieldSchema.createRowFieldSchema(rowSchema, isNullable, isOptional);
       case MAP:
         // Can the value type be nullable and have default values ? Guess not!

samza-sql/src/main/java/org/apache/samza/sql/interfaces/SqlIOConfig.java

@@ -81,13 +81,13 @@ public SqlIOConfig(String systemName, String streamName, List<String> sourcePart
     this.streamId = String.format("%s-%s", systemName, streamName);
 
     samzaRelConverterName = streamConfigs.get(CFG_SAMZA_REL_CONVERTER);
-    Validate.notEmpty(samzaRelConverterName,
-        String.format("%s is not set or empty for system %s", CFG_SAMZA_REL_CONVERTER, systemName));
+    Validate.notEmpty(samzaRelConverterName, String.format("System %s is unknown. %s is not set or empty for this"
+        + " system", systemName, CFG_SAMZA_REL_CONVERTER));
 
     if (isRemoteTable()) {
       samzaRelTableKeyConverterName = streamConfigs.get(CFG_SAMZA_REL_TABLE_KEY_CONVERTER);
-      Validate.notEmpty(samzaRelTableKeyConverterName,
-          String.format("%s is not set or empty for system %s", CFG_SAMZA_REL_CONVERTER, systemName));
+      Validate.notEmpty(samzaRelTableKeyConverterName, String.format("System %s is unknown. %s is not set or empty for"
+          + " this system", systemName, CFG_SAMZA_REL_CONVERTER));
     } else {
       samzaRelTableKeyConverterName = "";
     }