Fix Iceberg tests

Author: Abacn

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();