@Nullable
  public static Object getValueFromLiteral(ValueLiteralExpression expr) {
    LogicalType logicalType = expr.getOutputDataType().getLogicalType();
    switch (logicalType.getTypeRoot()) {
      case TIMESTAMP_WITHOUT_TIME_ZONE:
        return expr.getValueAs(LocalDateTime.class)
            .map(ldt -> ldt.toInstant(ZoneOffset.UTC).toEpochMilli())
            .orElse(null);
      case TIME_WITHOUT_TIME_ZONE:
        return expr.getValueAs(LocalTime.class)
            .map(lt -> lt.get(ChronoField.MILLI_OF_DAY))
            .orElse(null);
      case DATE:
        return expr.getValueAs(LocalDate.class)
            .map(date -> (int) date.toEpochDay())
            .orElse(null);
      // NOTE: All integral types of size less than Int are encoded as Ints in MT
      case BOOLEAN:
        return expr.getValueAs(Boolean.class).orElse(null);
      case TINYINT:
        return expr.getValueAs(Byte.class).orElse(null);
      case SMALLINT:
        return expr.getValueAs(Short.class).orElse(null);
      case INTEGER:
        return expr.getValueAs(Integer.class).orElse(null);
      case BIGINT:
        return expr.getValueAs(Long.class).orElse(null);
      case FLOAT:
        return expr.getValueAs(Float.class).orElse(null);
      case DOUBLE:
        return expr.getValueAs(Double.class).orElse(null);
      case BINARY:
      case VARBINARY:
        return expr.getValueAs(byte[].class).orElse(null);
      case CHAR:
      case VARCHAR:
        return expr.getValueAs(String.class).orElse(null);
      case DECIMAL:
        return expr.getValueAs(BigDecimal.class).orElse(null);
      default:
        throw new UnsupportedOperationException("Unsupported type: " + logicalType);
    }
  }
@Test
  void getValueFromLiteralForNonNull() {
    List<RowType.RowField> fields = ((RowType) ROW_DATA_TYPE_FIELD_NON_NULL.getLogicalType()).getFields();
    List<DataType> dataTypes = ROW_DATA_TYPE_FIELD_NON_NULL.getChildren();
    // tests for non-null literals
    List<Object> dataList = new ArrayList<>(fields.size());
    dataList.add(new Byte("1")); // f_tinyint
    dataList.add(new Short("2")); // f_smallint
    dataList.add(new Integer("3")); // f_int
    dataList.add(new Long("4")); // f_long
    dataList.add(new Float(5.0)); // f_float
    dataList.add(new Double(6.0)); // f_double
    dataList.add(new Boolean(true)); // f_boolean
    dataList.add(new BigDecimal(3.0)); // f_decimal
    dataList.add(getUTF8Bytes("hudi")); // f_bytes
    dataList.add("hudi ok"); // f_string
    dataList.add(LocalTime.of(1, 11, 11)); // f_time
    dataList.add(LocalDate.of(2023, 1, 2)); // f_date
    dataList.add(LocalDateTime.of(2023, 1, 2, 3, 4)); // f_timestamp
    CallExpression callExpression;
    for (int i = 0; i < fields.size(); i++) {
      // 1. Build all types
      callExpression = new CallExpression(
          BuiltInFunctionDefinitions.IS_NOT_NULL,
          Arrays.asList(
              new FieldReferenceExpression(
                  fields.get(i).getName(),
                  dataTypes.get(i),
                  i,
                  i),
              new ValueLiteralExpression(dataList.get(i), dataTypes.get(i))),
          DataTypes.BOOLEAN());
      List<Expression> childExprs = callExpression.getChildren();
      // 2. Parse each type
      if (dataList.get(i) instanceof LocalTime) {
        assertEquals(((LocalTime) dataList.get(i)).get(ChronoField.MILLI_OF_DAY), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
      } else if (dataList.get(i) instanceof LocalDate) {
        assertEquals((int) ((LocalDate) dataList.get(i)).toEpochDay(), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
      } else if (dataList.get(i) instanceof LocalDateTime) {
        assertEquals(((LocalDateTime) dataList.get(i)).toInstant(ZoneOffset.UTC).toEpochMilli(), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
      } else {
        assertEquals(dataList.get(i), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
      }
    }
  }