Fix EarliestLatestAnySqlAggregator to handle COMPLEX type

This changelist allows EarliestLatestAnySqlAggregator to accept COMPLEX
type as an operand. For its return type, we set it to VARCHAR, since
COMPLEX column is only generated by stringFirst/stringLast during ingestion
rollup.

sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/EarliestLatestAnySqlAggregator.java

@@ -28,11 +28,15 @@
 import org.apache.calcite.sql.SqlAggFunction;
 import org.apache.calcite.sql.SqlFunctionCategory;
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperatorBinding;
 import org.apache.calcite.sql.type.InferTypes;
 import org.apache.calcite.sql.type.OperandTypes;
-import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlReturnTypeInference;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
 import org.apache.calcite.util.Optionality;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.query.aggregation.any.DoubleAnyAggregatorFactory;
 import org.apache.druid.query.aggregation.any.FloatAnyAggregatorFactory;
@@ -85,6 +89,7 @@ AggregatorFactory createAggregatorFactory(String name, String fieldName, ValueTy
           case DOUBLE:
             return new DoubleFirstAggregatorFactory(name, fieldName);
           case STRING:
+          case COMPLEX:
             return new StringFirstAggregatorFactory(name, fieldName, maxStringBytes);
           default:
             throw new ISE("Cannot build EARLIEST aggregatorFactory for type[%s]", type);
@@ -104,6 +109,7 @@ AggregatorFactory createAggregatorFactory(String name, String fieldName, ValueTy
           case DOUBLE:
             return new DoubleLastAggregatorFactory(name, fieldName);
           case STRING:
+          case COMPLEX:
             return new StringLastAggregatorFactory(name, fieldName, maxStringBytes);
           default:
             throw new ISE("Cannot build LATEST aggregatorFactory for type[%s]", type);
@@ -219,6 +225,30 @@ public Aggregation toDruidAggregation(
     );
   }
 
+  static class EarliestLatestReturnTypeInference implements SqlReturnTypeInference
+  {
+    private final int ordinal;
+    private static final Logger log = new Logger(EarliestLatestReturnTypeInference.class);
+
+    public EarliestLatestReturnTypeInference(int ordinal)
+    {
+      this.ordinal = ordinal;
+    }
+
+    @Override
+    public RelDataType inferReturnType(SqlOperatorBinding sqlOperatorBinding)
+    {
+      RelDataType type = sqlOperatorBinding.getOperandType(0);
+      // For non-number and non-string type, which is COMPLEX type, we set the return type to VARCHAR.
+      if (!SqlTypeUtil.isNumeric(type) &&
+          !SqlTypeUtil.isString(type)) {
+        return sqlOperatorBinding.getTypeFactory().createSqlType(SqlTypeName.VARCHAR);
+      } else {
+        return type;
+      }
+    }
+  }
+
   private static class EarliestLatestSqlAggFunction extends SqlAggFunction
   {
     EarliestLatestSqlAggFunction(AggregatorType aggregatorType)
@@ -227,14 +257,14 @@ private static class EarliestLatestSqlAggFunction extends SqlAggFunction
           aggregatorType.name(),
           null,
           SqlKind.OTHER_FUNCTION,
-          ReturnTypes.ARG0,
+          new EarliestLatestReturnTypeInference(0),
           InferTypes.RETURN_TYPE,
           OperandTypes.or(
               OperandTypes.NUMERIC,
               OperandTypes.BOOLEAN,
               OperandTypes.sequence(
                   "'" + aggregatorType.name() + "(expr, maxBytesPerString)'\n",
-                  OperandTypes.STRING,
+                  OperandTypes.ANY,
                   OperandTypes.and(OperandTypes.NUMERIC, OperandTypes.LITERAL)
               )
           ),