fix iceberg unit test

Fix sql/jdbc integration test

Fix Iceberg tests

Author: Abacn

.github/trigger_files/beam_PostCommit_SQL.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run ",
-  "modification": 3
+  "modification": 4
 }

sdks/java/extensions/sql/jdbc/src/test/java/org/apache/beam/sdk/extensions/sql/jdbc/BeamSqlLineIT.java

@@ -129,9 +129,9 @@ public void testSelectFromPubsub() throws Exception {
 
     assertThat(
         Arrays.asList(
-            Arrays.asList("2018-07-01 21:25:20", "enroute", "40.702", "-74.001"),
-            Arrays.asList("2018-07-01 21:26:06", "enroute", "40.703", "-74.002"),
-            Arrays.asList("2018-07-02 13:26:06", "enroute", "30.0", "-72.32324")),
+            Arrays.asList("2018-07-01 21:25:20.000000", "enroute", "40.702", "-74.001"),
+            Arrays.asList("2018-07-01 21:26:06.000000", "enroute", "40.703", "-74.002"),
+            Arrays.asList("2018-07-02 13:26:06.000000", "enroute", "30.0", "-72.32324")),
         everyItem(IsIn.isOneOf(expectedResult.get(30, TimeUnit.SECONDS).toArray())));
   }
 
@@ -170,8 +170,8 @@ public void testFilterForSouthManhattan() throws Exception {
 
     assertThat(
         Arrays.asList(
-            Arrays.asList("2018-07-01 21:25:20", "enroute", "40.701", "-74.001"),
-            Arrays.asList("2018-07-01 21:26:06", "enroute", "40.702", "-74.002")),
+            Arrays.asList("2018-07-01 21:25:20.000000", "enroute", "40.701", "-74.001"),
+            Arrays.asList("2018-07-01 21:26:06.000000", "enroute", "40.702", "-74.002")),
         everyItem(IsIn.isOneOf(expectedResult.get(30, TimeUnit.SECONDS).toArray())));
   }
 

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java

@@ -108,6 +108,7 @@ public static String beamRow2CsvLine(Row row, CSVFormat csvFormat) {
    * @return The casted object in Schema.Field.Type.
    */
   public static Object autoCastField(Schema.Field field, @Nullable Object rawObj) {
+    // handle null
     if (rawObj == null) {
       if (!field.getType().getNullable()) {
         throw new IllegalArgumentException(String.format("Field %s not nullable", field.getName()));
@@ -116,12 +117,14 @@ public static Object autoCastField(Schema.Field field, @Nullable Object rawObj)
     }
 
     FieldType type = field.getType();
+    // handle NlsString
     if (CalciteUtils.isStringType(type)) {
       if (rawObj instanceof NlsString) {
         return ((NlsString) rawObj).getValue();
       } else {
         return rawObj;
       }
+      // handle date/time
     } else if (CalciteUtils.DATE.typesEqual(type) || CalciteUtils.NULLABLE_DATE.typesEqual(type)) {
       if (rawObj instanceof GregorianCalendar) { // used by the SQL CLI
         GregorianCalendar calendar = (GregorianCalendar) rawObj;
@@ -143,6 +146,7 @@ public static Object autoCastField(Schema.Field field, @Nullable Object rawObj)
     } else if (CalciteUtils.isDateTimeType(type)) {
       // Internal representation of Date in Calcite is convertible to Joda's Datetime.
       return new DateTime(rawObj);
+      // handle decimal
     } else if (type.getTypeName().isNumericType()
         && ((rawObj instanceof String)
             || (rawObj instanceof BigDecimal && type.getTypeName() != TypeName.DECIMAL))) {
@@ -168,8 +172,14 @@ public static Object autoCastField(Schema.Field field, @Nullable Object rawObj)
               String.format("Column type %s is not supported yet!", type));
       }
     } else if (type.getTypeName().isPrimitiveType()) {
+      // handle bytes represented by ByteString
       if (TypeName.BYTES.equals(type.getTypeName()) && rawObj instanceof ByteString) {
         return ((ByteString) rawObj).getBytes();
+        // handle Float <-> Double mixed use
+      } else if (TypeName.FLOAT.equals(type.getTypeName()) && rawObj instanceof Double) {
+        return ((Double) rawObj).floatValue();
+      } else if (TypeName.DOUBLE.equals(type.getTypeName()) && rawObj instanceof Float) {
+        return ((Float) rawObj).doubleValue();
       }
     }
     return rawObj;

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigquery/BeamSqlUnparseContext.java

@@ -128,9 +128,11 @@ public SqlNode toSql(RexProgram program, RexNode rex) {
     } else if (SqlKind.SEARCH.equals(rex.getKind())) {
       // Workaround CALCITE-4716
       RexCall search = (RexCall) rex;
-      RexLocalRef ref = (RexLocalRef) search.operands.get(1);
-      RexLiteral literal = (RexLiteral) program.getExprList().get(ref.getIndex());
-      rex = search.clone(search.getType(), ImmutableList.of(search.operands.get(0), literal));
+      if (search.operands.get(1) instanceof RexLocalRef) {
+        RexLocalRef ref = (RexLocalRef) search.operands.get(1);
+        RexLiteral literal = (RexLiteral) program.getExprList().get(ref.getIndex());
+        rex = search.clone(search.getType(), ImmutableList.of(search.operands.get(0), literal));
+      }
     }
 
     return super.toSql(program, rex);

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FilterUtils.java

@@ -153,7 +153,9 @@ static Expression convert(@Nullable String filter, Schema schema) {
   }
 
   private static Expression convert(SqlNode expression, Schema schema) throws SqlParseException {
-    checkArgument(expression instanceof SqlBasicCall);
+    checkArgument(
+        expression instanceof SqlBasicCall,
+        String.format("Expected SqlBasicCall, get %s", expression.getClass().getName()));
     SqlBasicCall call = (SqlBasicCall) expression;
 
     SqlOperator op = call.getOperator();
@@ -340,7 +342,7 @@ private static Object convertLiteral(SqlLiteral literal, String field, TypeID ty
         return literal.getValueAs(String.class);
       case DATE:
         LocalDate date;
-        if (SqlTypeName.STRING_TYPES.contains(typeName)) {
+        if (SqlTypeName.STRING_TYPES.contains(typeName) || SqlTypeName.UNKNOWN.equals(typeName)) {
           date = LocalDate.parse(literal.getValueAs(String.class));
         } else if (SqlTypeName.DATE.equals(typeName)) {
           DateString dateValue = literal.getValueAs(DateString.class);
@@ -351,7 +353,7 @@ private static Object convertLiteral(SqlLiteral literal, String field, TypeID ty
         return DateTimeUtil.daysFromDate(date);
       case TIME:
         LocalTime time;
-        if (SqlTypeName.STRING_TYPES.contains(typeName)) {
+        if (SqlTypeName.STRING_TYPES.contains(typeName) || SqlTypeName.UNKNOWN.equals(typeName)) {
           time = LocalTime.parse(literal.getValueAs(String.class));
         } else if (SqlTypeName.TIME.equals(typeName)) {
           TimeString timeString = literal.getValueAs(TimeString.class);
@@ -362,7 +364,7 @@ private static Object convertLiteral(SqlLiteral literal, String field, TypeID ty
         return DateTimeUtil.microsFromTime(time);
       case TIMESTAMP:
         LocalDateTime datetime;
-        if (SqlTypeName.STRING_TYPES.contains(typeName)) {
+        if (SqlTypeName.STRING_TYPES.contains(typeName) || SqlTypeName.UNKNOWN.equals(typeName)) {
           String value = literal.getValueAs(String.class);
           datetime = getLocalDateTime(value);
         } else if (SqlTypeName.DATE.equals(typeName)) {