SAMZA-2321: Samza-sql - Introduce REAL Samza sql schema type (apache#1154)

* Fix discrepancy between sql REAL and FLOAT

Author: atoomula

samza-api/src/main/java/org/apache/samza/sql/schema/SamzaSqlFieldType.java

@@ -28,8 +28,9 @@ public enum SamzaSqlFieldType {
   INT32, // four-byte signed integer.
   INT64, // eight-byte signed integer.
   DECIMAL, // Decimal integer
-  FLOAT,
-  DOUBLE,
+  REAL, // 4 bytes
+  FLOAT, // 8 bytes
+  DOUBLE, // 8 bytes
   STRING, // String.
   DATETIME, // Date and time.
   BOOLEAN, // Boolean.

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

@@ -270,9 +270,6 @@ public static Object convertToJavaObject(Object avroObj, Schema schema) {
         return new ByteString(fixed.bytes());
       case BYTES:
         return new ByteString(((ByteBuffer) avroObj).array());
-      case FLOAT:
-        // Convert Float to Double similar to how JavaTypeFactoryImpl represents Float type
-        return Double.parseDouble(Float.toString((Float) avroObj));
 
       default:
         return avroObj;
@@ -296,8 +293,6 @@ private static boolean isSchemaCompatibleWithRelObj(Object relObj, Schema unionS
         return relObj instanceof ByteString;
       case BYTES:
         return relObj instanceof ByteString;
-      case FLOAT:
-        return relObj instanceof Float || relObj instanceof Double;
       default:
         return true;
     }
@@ -319,8 +314,6 @@ private static boolean isSchemaCompatibleWithAvroObj(Object avroObj, Schema unio
         return avroObj instanceof GenericData.Fixed;
       case BYTES:
         return avroObj instanceof ByteBuffer;
-      case FLOAT:
-        return avroObj instanceof Float;
       default:
         return true;
     }

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

@@ -81,7 +81,8 @@ private SqlFieldSchema convertField(Schema fieldSchema, boolean isNullable, bool
       case DOUBLE:
         return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.DOUBLE, isNullable, isOptional);
       case FLOAT:
-        return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.FLOAT, isNullable, isOptional);
+        // Avro FLOAT is 4 bytes which maps to Sql REAL. Sql FLOAT is 8-bytes
+        return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.REAL, isNullable, isOptional);
       case ENUM:
         return SqlFieldSchema.createPrimitiveSchema(SamzaSqlFieldType.STRING, isNullable, isOptional);
       case UNION:

samza-sql/src/main/java/org/apache/samza/sql/planner/RelSchemaConverter.java

@@ -79,6 +79,8 @@ private RelDataType getRelDataType(SqlFieldSchema fieldSchema) {
         return createTypeWithNullability(createSqlType(SqlTypeName.BOOLEAN), fieldSchema.isNullable());
       case DOUBLE:
         return createTypeWithNullability(createSqlType(SqlTypeName.DOUBLE), fieldSchema.isNullable());
+      case REAL:
+        return createTypeWithNullability(createSqlType(SqlTypeName.REAL), fieldSchema.isNullable());
       case FLOAT:
         return createTypeWithNullability(createSqlType(SqlTypeName.FLOAT), fieldSchema.isNullable());
       case STRING:

samza-sql/src/test/java/org/apache/samza/sql/avro/TestAvroRelConversion.java

@@ -56,7 +56,6 @@
 import org.apache.samza.sql.avro.schemas.Profile;
 import org.apache.samza.sql.avro.schemas.SimpleRecord;
 import org.apache.samza.sql.avro.schemas.StreetNumRecord;
-import org.apache.samza.sql.avro.schemas.SubRecord;
 import org.apache.samza.sql.data.SamzaSqlRelMessage;
 import org.apache.samza.sql.planner.RelSchemaConverter;
 import org.apache.samza.sql.schema.SqlSchema;
@@ -353,7 +352,7 @@ private void validateAvroSerializedData(byte[] serializedData, Object unionValue
     Assert.assertEquals(message.getSamzaSqlRelRecord().getField("bool_value").get(), boolValue);
     Assert.assertEquals(message.getSamzaSqlRelRecord().getField("double_value").get(), doubleValue);
     Assert.assertEquals(message.getSamzaSqlRelRecord().getField("string_value").get(), new Utf8(testStrValue));
-    Assert.assertEquals(message.getSamzaSqlRelRecord().getField("float_value").get(), doubleValue);
+    Assert.assertEquals(message.getSamzaSqlRelRecord().getField("float_value").get(), floatValue);
     Assert.assertEquals(message.getSamzaSqlRelRecord().getField("long_value").get(), longValue);
     if (unionValue instanceof String) {
       Assert.assertEquals(message.getSamzaSqlRelRecord().getField("union_value").get(), new Utf8((String) unionValue));
@@ -386,10 +385,6 @@ private void validateAvroSerializedData(byte[] serializedData, Object unionValue
         Assert.assertTrue(record.get(field.name()).equals(complexRecordValue.get(field.name())));
       } else {
         Object expected = complexRecordValue.get(field.name());
-        if (expected instanceof Float) {
-          // AvroRelConverter converts float to double to be in sync with what Calcite does in JavaTypeFactoryImpl
-          expected = Double.parseDouble(Float.toString((Float) expected));
-        }
         Assert.assertEquals(expected, record.get(field.name()));
       }
     }

samza-sql/src/test/java/org/apache/samza/sql/system/TestAvroSystemFactory.java

@@ -325,6 +325,8 @@ private Object createComplexRecord(int index) {
       record.put("id", index);
       record.put("string_value", "Name" + index);
       record.put("bytes_value", ByteBuffer.wrap(("sample bytes").getBytes()));
+      record.put("float_value", index + 0.123456f);
+      record.put("double_value", index + 0.0123456789);
       MyFixed myFixedVar = new MyFixed();
       myFixedVar.bytes(DEFAULT_TRACKING_ID_BYTES);
       record.put("fixed_value", myFixedVar);

samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java

@@ -304,7 +304,7 @@ public void testEndToEndWithProjection() throws Exception {
         .sorted()
         .collect(Collectors.toList());
     Assert.assertEquals(numMessages, outMessages.size());
-    Assert.assertTrue(IntStream.range(0, numMessages).boxed().collect(Collectors.toList()).equals(outMessages));
+    Assert.assertEquals(IntStream.range(0, numMessages).boxed().collect(Collectors.toList()), outMessages);
   }
 
   @Test
@@ -472,6 +472,31 @@ public void testEndToEndComplexRecord() throws SamzaSqlValidatorException {
     Assert.assertEquals(numMessages, outMessages.size());
   }
 
+  @Test
+  public void testEndToEndWithFloatToStringConversion() throws Exception {
+    int numMessages = 20;
+
+    TestAvroSystemFactory.messages.clear();
+    Map<String, String> staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(numMessages);
+    String sql1 = "Insert into testavro.outputTopic"
+        + " select 'urn:li:member:' || cast(cast(float_value as int) as varchar) as string_value, id, float_value, "
+        + " double_value, true as bool_value from testavro.COMPLEX1";
+    List<String> sqlStmts = Arrays.asList(sql1);
+    staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts));
+
+    Config config = new MapConfig(staticConfigs);
+    new SamzaSqlValidator(config).validate(sqlStmts);
+
+    runApplication(config);
+
+    List<Integer> outMessages = TestAvroSystemFactory.messages.stream()
+        .map(x -> Integer.valueOf(((GenericRecord) x.getMessage()).get("string_value").toString().split(":")[3]))
+        .sorted()
+        .collect(Collectors.toList());
+    Assert.assertEquals(numMessages, outMessages.size());
+    Assert.assertEquals(IntStream.range(0, numMessages).boxed().collect(Collectors.toList()), outMessages);
+  }
+
   @Ignore
   @Test
   public void testEndToEndNestedRecord() throws SamzaSqlValidatorException {