diff --git a/samza-api/src/main/java/org/apache/samza/sql/schema/SqlFieldSchema.java b/samza-api/src/main/java/org/apache/samza/sql/schema/SqlFieldSchema.java index b944011709..d3cec05b13 100644 --- a/samza-api/src/main/java/org/apache/samza/sql/schema/SqlFieldSchema.java +++ b/samza-api/src/main/java/org/apache/samza/sql/schema/SqlFieldSchema.java @@ -24,44 +24,59 @@ */ public class SqlFieldSchema { - private SamzaSqlFieldType fieldType; - private SqlFieldSchema elementType; - private SqlFieldSchema valueType; - private SqlSchema rowSchema; + private final SamzaSqlFieldType fieldType; + private final SqlFieldSchema elementType; + private final SqlFieldSchema valueType; + private final SqlSchema rowSchema; + // A field is considered nullable when the field could have a null value. Please note that nullable field + // needs to be explicitly set while writing and is expected to be set while reading. A non-nullable field + // cannot have a null value. + private final Boolean isNullable; + // A field is considered optional when the field has a default value. Such a field need not be set while writing + // but is expected to be set while reading. + // Please note that nullable field is also optional field if a default value is set but the value for + // nullable non-optional field need to be explicitly set. + private final Boolean isOptional; - private SqlFieldSchema(SamzaSqlFieldType fieldType, SqlFieldSchema elementType, SqlFieldSchema valueType, SqlSchema rowSchema) { + private SqlFieldSchema(SamzaSqlFieldType fieldType, SqlFieldSchema elementType, SqlFieldSchema valueType, + SqlSchema rowSchema, boolean isNullable, boolean isOptional) { this.fieldType = fieldType; this.elementType = elementType; this.valueType = valueType; this.rowSchema = rowSchema; + this.isNullable = isNullable; + this.isOptional = isOptional; } /** - * Create a primitive fi + * Create a primitive field schema. * @param typeName * @return */ - public static SqlFieldSchema createPrimitiveSchema(SamzaSqlFieldType typeName) { - return new SqlFieldSchema(typeName, null, null, null); + public static SqlFieldSchema createPrimitiveSchema(SamzaSqlFieldType typeName, boolean isNullable, + boolean isOptional) { + return new SqlFieldSchema(typeName, null, null, null, isNullable, isOptional); } - public static SqlFieldSchema createArraySchema(SqlFieldSchema elementType) { - return new SqlFieldSchema(SamzaSqlFieldType.ARRAY, elementType, null, null); + public static SqlFieldSchema createArraySchema(SqlFieldSchema elementType, boolean isNullable, + boolean isOptional) { + return new SqlFieldSchema(SamzaSqlFieldType.ARRAY, elementType, null, null, isNullable, isOptional); } - public static SqlFieldSchema createMapSchema(SqlFieldSchema valueType) { - return new SqlFieldSchema(SamzaSqlFieldType.MAP, null, valueType, null); + public static SqlFieldSchema createMapSchema(SqlFieldSchema valueType, boolean isNullable, boolean isOptional) { + return new SqlFieldSchema(SamzaSqlFieldType.MAP, null, valueType, null, isNullable, isOptional); } - public static SqlFieldSchema createRowFieldSchema(SqlSchema rowSchema) { - return new SqlFieldSchema(SamzaSqlFieldType.ROW, null, null, rowSchema); + public static SqlFieldSchema createRowFieldSchema(SqlSchema rowSchema, boolean isNullable, boolean isOptional) { + return new SqlFieldSchema(SamzaSqlFieldType.ROW, null, null, rowSchema, isNullable, isOptional); } /** * @return whether the field is a primitive field type or not. */ public boolean isPrimitiveField() { - return fieldType != SamzaSqlFieldType.ARRAY && fieldType != SamzaSqlFieldType.MAP && fieldType != SamzaSqlFieldType.ROW; + return fieldType != SamzaSqlFieldType.ARRAY && fieldType != SamzaSqlFieldType.MAP && + fieldType != SamzaSqlFieldType.ROW; } /** @@ -93,5 +108,17 @@ public SqlSchema getRowSchema() { return rowSchema; } + /** + * Get if the field type is nullable. + */ + public boolean isNullable() { + return isNullable; + } + /** + * Get if the field type is optional. + */ + public boolean isOptional() { + return isOptional; + } } diff --git a/samza-sql-shell/src/main/java/org/apache/samza/sql/client/impl/SamzaExecutor.java b/samza-sql-shell/src/main/java/org/apache/samza/sql/client/impl/SamzaExecutor.java index db97516ed4..5939adb0ee 100755 --- a/samza-sql-shell/src/main/java/org/apache/samza/sql/client/impl/SamzaExecutor.java +++ b/samza-sql-shell/src/main/java/org/apache/samza/sql/client/impl/SamzaExecutor.java @@ -293,9 +293,9 @@ public List listFunctions(ExecutionContext context) { */ List udfs = new ArrayList<>(); udfs.add(new SamzaSqlUdfDisplayInfo("RegexMatch", "Matches the string to the regex", - Arrays.asList(SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING), - SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING)), - SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BOOLEAN))); + Arrays.asList(SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING, false, false), + SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING, false, false)), + SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BOOLEAN, false, false))); return udfs; } diff --git a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroTypeFactoryImpl.java b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroTypeFactoryImpl.java index 68116b6df1..4c069384b2 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroTypeFactoryImpl.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroTypeFactoryImpl.java @@ -23,7 +23,6 @@ import org.apache.avro.Schema; import org.apache.calcite.rel.type.RelDataTypeSystem; import org.apache.calcite.sql.type.SqlTypeFactoryImpl; -import org.apache.commons.lang3.Validate; import org.apache.samza.SamzaException; import org.apache.samza.sql.schema.SamzaSqlFieldType; import org.apache.samza.sql.schema.SqlFieldSchema; @@ -33,8 +32,8 @@ import org.slf4j.LoggerFactory; /** - * Factory that creates the Calcite relational types from the Avro Schema. This is used by the - * AvroRelConverter to convert the Avro schema to calcite relational schema. + * Factory that creates {@link SqlSchema} from the Avro Schema. This is used by the + * {@link AvroRelConverter} to convert Avro schema to Samza Sql schema. */ public class AvroTypeFactoryImpl extends SqlTypeFactoryImpl { @@ -60,7 +59,8 @@ private SqlSchema convertSchema(List fields) { SqlSchemaBuilder schemaBuilder = SqlSchemaBuilder.builder(); for (Schema.Field field : fields) { - SqlFieldSchema fieldSchema = convertField(field.schema()); + boolean isOptional = (field.defaultValue() != null); + SqlFieldSchema fieldSchema = convertField(field.schema(), false, isOptional); schemaBuilder.addField(field.name(), fieldSchema); } @@ -68,36 +68,41 @@ private SqlSchema convertSchema(List fields) { } 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()); - return SqlFieldSchema.createArraySchema(elementSchema); + return SqlFieldSchema.createArraySchema(elementSchema, isNullable, isOptional); case BOOLEAN: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BOOLEAN); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BOOLEAN, isNullable, isOptional); case DOUBLE: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.DOUBLE); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.DOUBLE, isNullable, isOptional); case FLOAT: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.FLOAT); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.FLOAT, isNullable, isOptional); case ENUM: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING, isNullable, isOptional); case UNION: - return getSqlTypeFromUnionTypes(fieldSchema.getTypes()); + return getSqlTypeFromUnionTypes(fieldSchema.getTypes(), isNullable, isOptional); case FIXED: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BYTES); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BYTES, isNullable, isOptional); case STRING: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING, isNullable, isOptional); case BYTES: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BYTES); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.BYTES, isNullable, isOptional); case INT: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.INT32); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.INT32, isNullable, isOptional); case LONG: - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.INT64); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.INT64, isNullable, isOptional); case RECORD: SqlSchema rowSchema = convertSchema(fieldSchema.getFields()); - return SqlFieldSchema.createRowFieldSchema(rowSchema); + return SqlFieldSchema.createRowFieldSchema(rowSchema, isNullable, isOptional); case MAP: - SqlFieldSchema valueType = convertField(fieldSchema.getValueType()); - return SqlFieldSchema.createMapSchema(valueType); + // Can the value type be nullable and have default values ? Guess not! + SqlFieldSchema valueType = convertField(fieldSchema.getValueType(), false, false); + return SqlFieldSchema.createMapSchema(valueType, isNullable, isOptional); default: String msg = String.format("Field Type %s is not supported", fieldSchema.getType()); LOG.error(msg); @@ -105,17 +110,17 @@ private SqlFieldSchema convertField(Schema fieldSchema) { } } - private SqlFieldSchema getSqlTypeFromUnionTypes(List types) { + private SqlFieldSchema getSqlTypeFromUnionTypes(List types, boolean isNullable, boolean isOptional) { // Typically a nullable field's schema is configured as an union of Null and a Type. // This is to check whether the Union is a Nullable field if (types.size() == 2) { if (types.get(0).getType() == Schema.Type.NULL) { - return convertField(types.get(1)); + return convertField(types.get(1), true, isOptional); } else if ((types.get(1).getType() == Schema.Type.NULL)) { - return convertField(types.get(0)); + return convertField(types.get(0), true, isOptional); } } - return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.ANY); + return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.ANY, isNullable, isOptional); } } diff --git a/samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java b/samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java index bdf03f7e61..83ccea0c5e 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/planner/QueryPlanner.java @@ -152,16 +152,15 @@ public RelRoot plan(String query) { } } - public static RelDataType getSourceRelSchema(RelSchemaProvider relSchemaProvider, - RelSchemaConverter relSchemaConverter) { - // If the source part is the last one, then fetch the schema corresponding to the stream and register. + public static SqlSchema getSourceSqlSchema(RelSchemaProvider relSchemaProvider) { SqlSchema sqlSchema = relSchemaProvider.getSqlSchema(); List fieldNames = new ArrayList<>(); List fieldTypes = new ArrayList<>(); if (!sqlSchema.containsField(SamzaSqlRelMessage.KEY_NAME)) { fieldNames.add(SamzaSqlRelMessage.KEY_NAME); - fieldTypes.add(SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.ANY)); + // Key is a nullable and optional field. It is defaulted to null in SamzaSqlRelMessage. + fieldTypes.add(SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.ANY, true, true)); } fieldNames.addAll( @@ -169,8 +168,13 @@ public static RelDataType getSourceRelSchema(RelSchemaProvider relSchemaProvider fieldTypes.addAll( sqlSchema.getFields().stream().map(SqlSchema.SqlField::getFieldSchema).collect(Collectors.toList())); - SqlSchema newSchema = new SqlSchema(fieldNames, fieldTypes); - return relSchemaConverter.convertToRelSchema(newSchema); + return new SqlSchema(fieldNames, fieldTypes); + } + + public static RelDataType getSourceRelSchema(RelSchemaProvider relSchemaProvider, + RelSchemaConverter relSchemaConverter) { + // If the source part is the last one, then fetch the schema corresponding to the stream and register. + return relSchemaConverter.convertToRelSchema(getSourceSqlSchema(relSchemaProvider)); } private static Table createTableFromRelSchema(RelDataType relationalSchema) { diff --git a/samza-sql/src/main/java/org/apache/samza/sql/planner/RelSchemaConverter.java b/samza-sql/src/main/java/org/apache/samza/sql/planner/RelSchemaConverter.java index 6634f5ad04..dbd74b6464 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/planner/RelSchemaConverter.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/planner/RelSchemaConverter.java @@ -27,6 +27,7 @@ import org.apache.calcite.rel.type.RelDataTypeSystem; import org.apache.calcite.rel.type.RelRecordType; import org.apache.calcite.sql.type.ArraySqlType; +import org.apache.calcite.sql.type.MapSqlType; import org.apache.calcite.sql.type.SqlTypeFactoryImpl; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.samza.SamzaException; @@ -73,30 +74,29 @@ private RelDataType getRelDataType(SqlFieldSchema fieldSchema) { switch (fieldSchema.getFieldType()) { case ARRAY: RelDataType elementType = getRelDataType(fieldSchema.getElementSchema()); - return new ArraySqlType(elementType, true); + return new ArraySqlType(elementType, fieldSchema.isNullable()); case BOOLEAN: - return createTypeWithNullability(createSqlType(SqlTypeName.BOOLEAN), true); + return createTypeWithNullability(createSqlType(SqlTypeName.BOOLEAN), fieldSchema.isNullable()); case DOUBLE: - return createTypeWithNullability(createSqlType(SqlTypeName.DOUBLE), true); + return createTypeWithNullability(createSqlType(SqlTypeName.DOUBLE), fieldSchema.isNullable()); case FLOAT: - return createTypeWithNullability(createSqlType(SqlTypeName.FLOAT), true); + return createTypeWithNullability(createSqlType(SqlTypeName.FLOAT), fieldSchema.isNullable()); case STRING: - return createTypeWithNullability(createSqlType(SqlTypeName.VARCHAR), true); + return createTypeWithNullability(createSqlType(SqlTypeName.VARCHAR), fieldSchema.isNullable()); case BYTES: - return createTypeWithNullability(createSqlType(SqlTypeName.VARBINARY), true); + return createTypeWithNullability(createSqlType(SqlTypeName.VARBINARY), fieldSchema.isNullable()); case INT16: case INT32: - return createTypeWithNullability(createSqlType(SqlTypeName.INTEGER), true); + return createTypeWithNullability(createSqlType(SqlTypeName.INTEGER), fieldSchema.isNullable()); case INT64: - return createTypeWithNullability(createSqlType(SqlTypeName.BIGINT), true); + return createTypeWithNullability(createSqlType(SqlTypeName.BIGINT), fieldSchema.isNullable()); case ROW: case ANY: // TODO Calcite execution engine doesn't support record type yet. - return createTypeWithNullability(createSqlType(SqlTypeName.ANY), true); + return createTypeWithNullability(createSqlType(SqlTypeName.ANY), fieldSchema.isNullable()); case MAP: RelDataType valueType = getRelDataType(fieldSchema.getValueScehma()); - return super.createMapType(createTypeWithNullability(createSqlType(SqlTypeName.VARCHAR), true), - createTypeWithNullability(valueType, true)); + return new MapSqlType(createSqlType(SqlTypeName.VARCHAR), valueType, fieldSchema.isNullable()); default: String msg = String.format("Field Type %s is not supported", fieldSchema.getFieldType()); LOG.error(msg); diff --git a/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlValidator.java b/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlValidator.java index 08d4497403..9482a75d6c 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlValidator.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/planner/SamzaSqlValidator.java @@ -40,6 +40,8 @@ import org.apache.samza.sql.interfaces.RelSchemaProvider; import org.apache.samza.sql.interfaces.SamzaSqlJavaTypeFactoryImpl; import org.apache.samza.sql.runner.SamzaSqlApplicationConfig; +import org.apache.samza.sql.schema.SqlFieldSchema; +import org.apache.samza.sql.schema.SqlSchema; import org.apache.samza.sql.util.SamzaSqlQueryParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -98,24 +100,59 @@ protected void validate(RelRoot relRoot, SamzaSqlQueryParser.QueryInfo qinfo, Sa protected void validateOutput(RelRoot relRoot, RelSchemaProvider relSchemaProvider) throws SamzaSqlValidatorException { RelRecordType outputRecord = (RelRecordType) QueryPlanner.getSourceRelSchema(relSchemaProvider, new RelSchemaConverter()); + // Get Samza Sql schema along with Calcite schema. The reason is that the Calcite schema does not have a way + // to represent optional fields while Samza Sql schema can represent optional fields. This is the only reason that + // we use SqlSchema in validating output. + SqlSchema outputSqlSchema = QueryPlanner.getSourceSqlSchema(relSchemaProvider); + LogicalProject project = (LogicalProject) relRoot.rel; RelRecordType projetRecord = (RelRecordType) project.getRowType(); - validateOutputRecords(outputRecord, projetRecord); + + validateOutputRecords(outputRecord, outputSqlSchema, projetRecord); } - protected void validateOutputRecords(RelRecordType outputRecord, RelRecordType projectRecord) + protected void validateOutputRecords(RelRecordType outputRecord, SqlSchema outputSqlSchema, + RelRecordType projectRecord) throws SamzaSqlValidatorException { Map outputRecordMap = outputRecord.getFieldList().stream().collect( Collectors.toMap(RelDataTypeField::getName, RelDataTypeField::getType)); + Map outputFieldSchemaMap = outputSqlSchema.getFields().stream().collect( + Collectors.toMap(SqlSchema.SqlField::getFieldName, SqlSchema.SqlField::getFieldSchema)); Map projectRecordMap = projectRecord.getFieldList().stream().collect( Collectors.toMap(RelDataTypeField::getName, RelDataTypeField::getType)); - // There could be default values for the output schema and hence fields in project schema could be a subset of - // fields in output schema. - // TODO: SAMZA-2316: Validate that all non-default value fields in output schema are set in the projected fields. + // Ensure that all non-optional fields in output schema are set in the sql query and are of the + // same type. + for (Map.Entry entry : outputRecordMap.entrySet()) { + RelDataType projectFieldType = projectRecordMap.get(entry.getKey()); + SqlFieldSchema outputSqlFieldSchema = outputFieldSchemaMap.get(entry.getKey()); + + if (projectFieldType == null) { + // If an output schema field is not found in the sql query, ignore it if the field is optional. + // Otherwise, throw an error. + if (outputSqlFieldSchema.isOptional()) { + continue; + } + String errMsg = String.format("Field '%s' in output schema does not match any projected fields.", + entry.getKey()); + LOG.error(errMsg); + throw new SamzaSqlValidatorException(errMsg); + } else if (!compareFieldTypes(entry.getValue(), outputSqlFieldSchema, projectFieldType)) { + String errMsg = String.format("Field '%s' with type '%s' in output schema does not match the field type '%s' in" + + " projected fields.", entry.getKey(), entry.getValue(), projectFieldType); + LOG.error(errMsg); + throw new SamzaSqlValidatorException(errMsg); + } + } + + // Ensure that all fields from sql statement exist in the output schema and are of the same type. for (Map.Entry entry : projectRecordMap.entrySet()) { RelDataType outputFieldType = outputRecordMap.get(entry.getKey()); + SqlFieldSchema outputSqlFieldSchema = outputFieldSchemaMap.get(entry.getKey()); + if (outputFieldType == null) { + // If a field in sql query is not found in the output schema, ignore if it is a Samza Sql special op. + // Otherwise, throw an error. if (entry.getKey().equals(SamzaSqlRelMessage.OP_NAME)) { continue; } @@ -123,7 +160,7 @@ protected void validateOutputRecords(RelRecordType outputRecord, RelRecordType p entry.getKey()); LOG.error(errMsg); throw new SamzaSqlValidatorException(errMsg); - } else if (!compareFieldTypes(outputFieldType, entry.getValue())) { + } else if (!compareFieldTypes(outputFieldType, outputSqlFieldSchema, entry.getValue())) { String errMsg = String.format("Field '%s' with type '%s' in select query does not match the field type '%s' in" + " output schema.", entry.getKey(), entry.getValue(), outputFieldType); LOG.error(errMsg); @@ -132,7 +169,8 @@ protected void validateOutputRecords(RelRecordType outputRecord, RelRecordType p } } - protected boolean compareFieldTypes(RelDataType outputFieldType, RelDataType selectQueryFieldType) { + protected boolean compareFieldTypes(RelDataType outputFieldType, SqlFieldSchema sqlFieldSchema, + RelDataType selectQueryFieldType) { RelDataType projectFieldType; // JavaTypes are relevant for Udf argument and return types @@ -168,7 +206,8 @@ protected boolean compareFieldTypes(RelDataType outputFieldType, RelDataType sel return projectSqlType == SqlTypeName.FLOAT; case ROW: try { - validateOutputRecords((RelRecordType) outputFieldType, (RelRecordType) projectFieldType); + validateOutputRecords((RelRecordType) outputFieldType, sqlFieldSchema.getRowSchema(), + (RelRecordType) projectFieldType); } catch (SamzaSqlValidatorException e) { LOG.error("A field in select query does not match with the output schema.", e); return false; diff --git a/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.avsc b/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.avsc index 5e78bd9360..c307b10f0f 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.avsc +++ b/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.avsc @@ -26,14 +26,12 @@ { "name": "id", "doc": "Record id.", - "type": ["null", "int"], - "default":null + "type": "int" }, { "name": "bool_value", "doc": "Boolean Value.", - "type": ["null", "boolean"], - "default":null + "type": ["null", "boolean"] }, { "name": "double_value", @@ -72,8 +70,8 @@ "name": "MyFixed", "type":"fixed", "size":16 - } - ] + }], + "default":null }, { "name": "array_values", @@ -120,7 +118,8 @@ "doc" : "", "fields" : [ ] } - ] + ], + "default":null }, { "name": "array_records", diff --git a/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.java b/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.java index 7796004fd6..91a447f641 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/avro/schemas/ComplexRecord.java @@ -16,11 +16,17 @@ * specific language governing permissions and limitations * under the License. */ + +/** + * Autogenerated by Avro + * + * DO NOT EDIT DIRECTLY + */ package org.apache.samza.sql.avro.schemas; @SuppressWarnings("all") public class ComplexRecord extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord { - public static final org.apache.avro.Schema SCHEMA$ = org.apache.avro.Schema.parse("{\"type\":\"record\",\"name\":\"ComplexRecord\",\"namespace\":\"org.apache.samza.sql.avro.schemas\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"],\"doc\":\"Record id.\",\"default\":null},{\"name\":\"bool_value\",\"type\":[\"null\",\"boolean\"],\"doc\":\"Boolean Value.\",\"default\":null},{\"name\":\"double_value\",\"type\":[\"null\",\"double\"],\"doc\":\"double Value.\",\"default\":null},{\"name\":\"float_value\",\"type\":[\"null\",\"float\"],\"doc\":\"float Value.\",\"default\":null},{\"name\":\"string_value\",\"type\":[\"null\",\"string\"],\"doc\":\"string Value.\",\"default\":null},{\"name\":\"bytes_value\",\"type\":[\"null\",\"bytes\"],\"doc\":\"bytes Value.\",\"default\":null},{\"name\":\"long_value\",\"type\":[\"null\",\"long\"],\"doc\":\"long Value.\",\"default\":null},{\"name\":\"fixed_value\",\"type\":[\"null\",{\"type\":\"fixed\",\"name\":\"MyFixed\",\"size\":16}],\"doc\":\"fixed Value.\"},{\"name\":\"array_values\",\"type\":[\"null\",{\"type\":\"array\",\"items\":\"string\"}],\"doc\":\"array values in the record.\",\"default\":[]},{\"name\":\"map_values\",\"type\":[\"null\",{\"type\":\"map\",\"values\":\"string\"}],\"doc\":\"map values in the record.\",\"default\":[]},{\"name\":\"enum_value\",\"type\":[\"null\",{\"type\":\"enum\",\"name\":\"TestEnumType\",\"symbols\":[\"foo\",\"bar\"]}],\"doc\":\"enum value.\",\"default\":[]},{\"name\":\"empty_record\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"emptySubRecord\",\"fields\":[]}]},{\"name\":\"array_records\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"SubRecord\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"],\"doc\":\"sub record id\"},{\"name\":\"sub_values\",\"type\":{\"type\":\"array\",\"items\":\"string\"},\"doc\":\"Sub record \"}]}],\"doc\":\"array of records.\",\"default\":[]},{\"name\":\"union_value\",\"type\":[\"null\",\"SubRecord\",\"string\"],\"doc\":\"union Value.\",\"default\":null}]}"); + public static final org.apache.avro.Schema SCHEMA$ = org.apache.avro.Schema.parse("{\"type\":\"record\",\"name\":\"ComplexRecord\",\"namespace\":\"org.apache.samza.sql.avro.schemas\",\"fields\":[{\"name\":\"id\",\"type\":\"int\",\"doc\":\"Record id.\"},{\"name\":\"bool_value\",\"type\":[\"null\",\"boolean\"],\"doc\":\"Boolean Value.\"},{\"name\":\"double_value\",\"type\":[\"null\",\"double\"],\"doc\":\"double Value.\",\"default\":null},{\"name\":\"float_value\",\"type\":[\"null\",\"float\"],\"doc\":\"float Value.\",\"default\":null},{\"name\":\"string_value\",\"type\":[\"null\",\"string\"],\"doc\":\"string Value.\",\"default\":null},{\"name\":\"bytes_value\",\"type\":[\"null\",\"bytes\"],\"doc\":\"bytes Value.\",\"default\":null},{\"name\":\"long_value\",\"type\":[\"null\",\"long\"],\"doc\":\"long Value.\",\"default\":null},{\"name\":\"fixed_value\",\"type\":[\"null\",{\"type\":\"fixed\",\"name\":\"MyFixed\",\"size\":16}],\"doc\":\"fixed Value.\",\"default\":null},{\"name\":\"array_values\",\"type\":[\"null\",{\"type\":\"array\",\"items\":\"string\"}],\"doc\":\"array values in the record.\",\"default\":[]},{\"name\":\"map_values\",\"type\":[\"null\",{\"type\":\"map\",\"values\":\"string\"}],\"doc\":\"map values in the record.\",\"default\":[]},{\"name\":\"enum_value\",\"type\":[\"null\",{\"type\":\"enum\",\"name\":\"TestEnumType\",\"symbols\":[\"foo\",\"bar\"]}],\"doc\":\"enum value.\",\"default\":[]},{\"name\":\"empty_record\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"emptySubRecord\",\"fields\":[]}],\"default\":null},{\"name\":\"array_records\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"SubRecord\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"],\"doc\":\"sub record id\"},{\"name\":\"sub_values\",\"type\":{\"type\":\"array\",\"items\":\"string\"},\"doc\":\"Sub record \"}]}],\"doc\":\"array of records.\",\"default\":[]},{\"name\":\"union_value\",\"type\":[\"null\",\"SubRecord\",\"string\"],\"doc\":\"union Value.\",\"default\":null}]}"); /** Record id. */ public java.lang.Integer id; /** Boolean Value. */ diff --git a/samza-sql/src/test/java/org/apache/samza/sql/planner/TestSamzaSqlValidator.java b/samza-sql/src/test/java/org/apache/samza/sql/planner/TestSamzaSqlValidator.java index b2ce6f6f4a..4c9522eb22 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/planner/TestSamzaSqlValidator.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/planner/TestSamzaSqlValidator.java @@ -22,14 +22,11 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import org.apache.calcite.rel.RelRoot; import org.apache.samza.SamzaException; import org.apache.samza.config.Config; import org.apache.samza.config.MapConfig; -import org.apache.samza.sql.dsl.SamzaSqlDslConverter; import org.apache.samza.sql.runner.SamzaSqlApplicationConfig; import org.apache.samza.sql.runner.SamzaSqlApplicationRunner; -import org.apache.samza.sql.util.SamzaSqlQueryParser; import org.apache.samza.sql.util.SamzaSqlTestConfig; import org.junit.Assert; import org.junit.Before; @@ -54,7 +51,7 @@ public void setUp() { public void testBasicValidation() throws SamzaSqlValidatorException { Map config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, - "Insert into testavro.outputTopic(id) select id, name as string_value" + "Insert into testavro.outputTopic select id, true as bool_value, name as string_value" + " from testavro.level1.level2.SIMPLE1 as s where s.id = 1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); @@ -151,6 +148,46 @@ public void testUnsupportedOperator() throws SamzaSqlValidatorException { new SamzaSqlValidator(samzaConfig).validate(sqlStmts); } + @Test + public void testNonDefaultButNullableField() { + Map config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); + // bool_value is missing + config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, + "Insert into testavro.outputTopic(id) select Flatten(a) as id from (select MyTestArray(id) a from testavro.SIMPLE1)"); + Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); + + List sqlStmts = fetchSqlFromConfig(config); + try { + new SamzaSqlValidator(samzaConfig).validate(sqlStmts); + } catch (SamzaSqlValidatorException e) { + Assert.assertTrue(e.getMessage().contains("Field 'bool_value' in output schema does not match any projected fields.")); + return; + } + + Assert.fail("Validation test has failed."); + } + + @Test + public void testNonDefaultOutputField() { + Map config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(configs, 1); + // id is non-default field. + String sql = "Insert into testavro.outputTopic " + + " select NOT(id = 5) as bool_value, CASE WHEN id IN (5, 6, 7) THEN CAST('foo' AS VARCHAR) WHEN id < 5 THEN CAST('bars' AS VARCHAR) ELSE NULL END as string_value from testavro.SIMPLE1"; + config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, sql); + Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); + + List sqlStmts = fetchSqlFromConfig(config); + + try { + new SamzaSqlValidator(samzaConfig).validate(sqlStmts); + } catch (SamzaSqlValidatorException e) { + Assert.assertTrue(e.getMessage().contains("Field 'id' in output schema does not match")); + return; + } + + Assert.fail("Validation test has failed."); + } + @Test public void testFormatErrorString() { String sql = diff --git a/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java b/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java index d81cb3c04c..dec886e022 100644 --- a/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java +++ b/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java @@ -289,8 +289,8 @@ public void testEndToEndWithProjection() throws Exception { TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); - String sql1 = "Insert into testavro.outputTopic(id, long_value) " - + " select id, TIMESTAMPDIFF(HOUR, CURRENT_TIMESTAMP(), LOCALTIMESTAMP()) + MONTH(CURRENT_DATE()) as long_value from testavro.SIMPLE1"; + String sql1 = "Insert into testavro.outputTopic(id, bool_value, long_value) " + + " select id, NOT(id = 5) as bool_value, TIMESTAMPDIFF(HOUR, CURRENT_TIMESTAMP(), LOCALTIMESTAMP()) + MONTH(CURRENT_DATE()) as long_value from testavro.SIMPLE1"; List sqlStmts = Arrays.asList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -401,8 +401,8 @@ public void testEndToEndWithLike() throws Exception { TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); - String sql1 = "Insert into testavro.outputTopic(id, long_value) " - + " select id, name as string_value from testavro.SIMPLE1 where name like 'Name%'"; + String sql1 = "Insert into testavro.outputTopic(id, bool_value, string_value) " + + " select id, NOT(id = 5) as bool_value, name as string_value from testavro.SIMPLE1 where name like 'Name%'"; List sqlStmts = Arrays.asList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -427,8 +427,8 @@ public void testEndToEndFlatten() throws Exception { LOG.info(" Class Path : " + RelOptUtil.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath()); String sql1 = - "Insert into testavro.outputTopic(string_value, id, bytes_value, fixed_value, float_value) " - + " select Flatten(array_values) as string_value, id, bytes_value, fixed_value, float_value " + "Insert into testavro.outputTopic(string_value, id, bool_value, bytes_value, fixed_value, float_value) " + + " select Flatten(array_values) as string_value, id, NOT(id = 5) as bool_value, bytes_value, fixed_value, float_value " + " from testavro.COMPLEX1"; List sqlStmts = Collections.singletonList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -457,7 +457,7 @@ public void testEndToEndComplexRecord() throws SamzaSqlValidatorException { String sql1 = "Insert into testavro.outputTopic" - + " select map_values['key0'] as string_value, union_value, array_values, map_values, id, bytes_value," + + " select bool_value, map_values['key0'] as string_value, union_value, array_values, map_values, id, bytes_value," + " fixed_value, float_value from testavro.COMPLEX1"; List sqlStmts = Collections.singletonList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -502,7 +502,8 @@ public void testEndToEndFlattenWithUdf() throws Exception { TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); String sql1 = - "Insert into testavro.outputTopic(id) select Flatten(MyTestArray(id)) as id from testavro.SIMPLE1"; + "Insert into testavro.outputTopic(id, bool_value) select Flatten(MyTestArray(id)) as id, NOT(id = 5) as bool_value" + + " from testavro.SIMPLE1"; List sqlStmts = Collections.singletonList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -527,7 +528,8 @@ public void testEndToEndSubQuery() throws Exception { TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); String sql1 = - "Insert into testavro.outputTopic(id) select Flatten(a) as id from (select MyTestArray(id) a from testavro.SIMPLE1)"; + "Insert into testavro.outputTopic(id, bool_value) select Flatten(a) as id, true as bool_value" + + " from (select MyTestArray(id) a from testavro.SIMPLE1)"; List sqlStmts = Collections.singletonList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -551,8 +553,8 @@ public void testUdfUnTypedArgumentToTypedUdf() throws SamzaSqlValidatorException int numMessages = 20; TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); - String sql1 = "Insert into testavro.outputTopic(id, long_value) " - + "select id, MyTest(MyTestObj(id)) as long_value from testavro.SIMPLE1"; + String sql1 = "Insert into testavro.outputTopic(id, bool_value, long_value) " + + "select id, NOT(id = 5) as bool_value, MyTest(MyTestObj(id)) as long_value from testavro.SIMPLE1"; List sqlStmts = Collections.singletonList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -575,8 +577,8 @@ public void testEndToEndUdf() throws Exception { int numMessages = 20; TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); - String sql1 = "Insert into testavro.outputTopic(id, long_value) " - + "select id, MYTest(id) as long_value from testavro.SIMPLE1"; + String sql1 = "Insert into testavro.outputTopic(id, bool_value, long_value) " + + "select id, NOT(id = 5) as bool_value, MYTest(id) as long_value from testavro.SIMPLE1"; List sqlStmts = Collections.singletonList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -623,8 +625,8 @@ public void testEndToEndUdfPolymorphism() throws Exception { int numMessages = 20; TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); - String sql1 = "Insert into testavro.outputTopic(id, long_value) " - + "select MyTestPoly(id) as long_value, MyTestPoly(name) as id from testavro.SIMPLE1"; + String sql1 = "Insert into testavro.outputTopic(id, bool_value, long_value) " + + "select MyTestPoly(id) as long_value, NOT(id = 5) as bool_value, MyTestPoly(name) as id from testavro.SIMPLE1"; List sqlStmts = Collections.singletonList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); @@ -652,8 +654,8 @@ public void testRegexMatchUdfInWhereClause() throws Exception { TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages); String sql1 = - "Insert into testavro.outputTopic(id) " - + "select id " + "Insert into testavro.outputTopic(id, bool_value) " + + "select id, NOT(id = 5) as bool_value " + "from testavro.SIMPLE1 " + "where RegexMatch('.*4', name)"; List sqlStmts = Collections.singletonList(sql1);