class BeamSqlCastExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, BeamSqlExpressionEnvironment env) {
    SqlTypeName castOutputType = getOutputType();
    switch (castOutputType) {
      case BOOLEAN:
        return BeamSqlPrimitive.of(
            SqlTypeName.BOOLEAN,
            SqlFunctions.toBoolean(opValueEvaluated(index, inputRow, window, env)));
      case INTEGER:
        return BeamSqlPrimitive.of(
            SqlTypeName.INTEGER,
            SqlFunctions.toInt(opValueEvaluated(index, inputRow, window, env)));
      case DOUBLE:
        return BeamSqlPrimitive.of(
            SqlTypeName.DOUBLE,
            SqlFunctions.toDouble(opValueEvaluated(index, inputRow, window, env)));
      case SMALLINT:
        return BeamSqlPrimitive.of(
            SqlTypeName.SMALLINT,
            SqlFunctions.toShort(opValueEvaluated(index, inputRow, window, env)));
      case TINYINT:
        return BeamSqlPrimitive.of(
            SqlTypeName.TINYINT,
            SqlFunctions.toByte(opValueEvaluated(index, inputRow, window, env)));
      case BIGINT:
        return BeamSqlPrimitive.of(
            SqlTypeName.BIGINT,
            SqlFunctions.toLong(opValueEvaluated(index, inputRow, window, env)));
      case DECIMAL:
        return BeamSqlPrimitive.of(
            SqlTypeName.DECIMAL,
            SqlFunctions.toBigDecimal(opValueEvaluated(index, inputRow, window, env)));
      case FLOAT:
        return BeamSqlPrimitive.of(
            SqlTypeName.FLOAT,
            SqlFunctions.toFloat(opValueEvaluated(index, inputRow, window, env)));
      case CHAR:
      case VARCHAR:
        // TODO: We should do standards-compliant string conversions here.
        return BeamSqlPrimitive.of(
            SqlTypeName.VARCHAR, opValueEvaluated(index, inputRow, window, env).toString());
      case DATE:
        return BeamSqlPrimitive.of(
            SqlTypeName.DATE, toDate(opValueEvaluated(index, inputRow, window, env)));
      case TIMESTAMP:
        return BeamSqlPrimitive.of(
            SqlTypeName.TIMESTAMP, toTimeStamp(opValueEvaluated(index, inputRow, window, env)));
      default:
        throw new UnsupportedOperationException(
            String.format("Cast to type %s not supported", castOutputType));
    }
  }

  public  BeamSqlCastExpression(List<BeamSqlExpression> operands, SqlTypeName castType);

}