yihua commented on code in PR #13711: URL: https://github.com/apache/hudi/pull/13711#discussion_r2359708388
########## hudi-common/src/main/java/org/apache/hudi/stats/ValueType.java: ########## @@ -0,0 +1,668 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.stats; + +import org.apache.hudi.avro.AvroSchemaUtils; +import org.apache.hudi.avro.HoodieAvroWrapperUtils; +import org.apache.hudi.common.util.DateTimeUtils; + +import org.apache.avro.LogicalTypes; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.util.Utf8; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.LogicalTypeTokenParser; +import org.apache.parquet.schema.PrimitiveType; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.math.MathContext; +import java.math.RoundingMode; +import java.nio.ByteBuffer; +import java.sql.Timestamp; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneOffset; +import java.util.Objects; +import java.util.UUID; +import java.util.function.BiFunction; +import java.util.function.Function; + +import static org.apache.hudi.avro.HoodieAvroUtils.convertBytesToBigDecimal; +import static org.apache.hudi.common.util.DateTimeUtils.instantToMicros; +import static org.apache.hudi.common.util.DateTimeUtils.instantToNanos; +import static org.apache.hudi.common.util.DateTimeUtils.microsToInstant; +import static org.apache.hudi.common.util.DateTimeUtils.nanosToInstant; +import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes; + +public enum ValueType { + V1(HoodieAvroWrapperUtils.PrimitiveWrapperType.V1.getClazz(), HoodieAvroWrapperUtils.PrimitiveWrapperType.V1, + ValueType::passThrough, ValueType::passThrough, ValueType::passThrough), + NULL(HoodieAvroWrapperUtils.PrimitiveWrapperType.NULL.getClazz(), HoodieAvroWrapperUtils.PrimitiveWrapperType.NULL, + ValueType::passThrough, ValueType::passThrough, ValueType::passThrough), + BOOLEAN(HoodieAvroWrapperUtils.PrimitiveWrapperType.BOOLEAN, ValueType::castToBoolean), + INT(HoodieAvroWrapperUtils.PrimitiveWrapperType.INT, ValueType::castToInteger), + LONG(HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, ValueType::castToLong), + FLOAT(HoodieAvroWrapperUtils.PrimitiveWrapperType.FLOAT, ValueType::castToFloat), + DOUBLE(HoodieAvroWrapperUtils.PrimitiveWrapperType.DOUBLE, ValueType::castToDouble), + STRING(HoodieAvroWrapperUtils.PrimitiveWrapperType.STRING, ValueType::castToString), + BYTES(HoodieAvroWrapperUtils.PrimitiveWrapperType.BYTES, ValueType::castToBytes), + FIXED(HoodieAvroWrapperUtils.PrimitiveWrapperType.BYTES, ValueType::castToFixed), + DECIMAL(BigDecimal.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.BYTES, + ValueType::castToDecimal, ValueType::toDecimal, ValueType::fromDecimal), + UUID(UUID.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.STRING, + ValueType::castToUUID, ValueType::toUUID, ValueType::fromUUID), + DATE(LocalDate.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.INT, + ValueType::castToDate, ValueType::toDate, ValueType::fromDate), + TIME_MILLIS(LocalTime.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.INT, + ValueType::castToTimeMillis, ValueType::toTimeMillis, ValueType::fromTimeMillis), + TIME_MICROS(LocalTime.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, + ValueType::castToTimeMicros, ValueType::toTimeMicros, ValueType::fromTimeMicros), + TIMESTAMP_MILLIS(Instant.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, + ValueType::castToTimestampMillis, ValueType::toTimestampMillis, ValueType::fromTimestampMillis), + TIMESTAMP_MICROS(Instant.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, + ValueType::castToTimestampMicros, ValueType::toTimestampMicros, ValueType::fromTimestampMicros), + TIMESTAMP_NANOS(Instant.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, + ValueType::castToTimestampNanos, ValueType::toTimestampNanos, ValueType::fromTimestampNanos), + LOCAL_TIMESTAMP_MILLIS(LocalDateTime.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, + ValueType::castToLocalTimestampMillis, ValueType::toLocalTimestampMillis, ValueType::fromLocalTimestampMillis), + LOCAL_TIMESTAMP_MICROS(LocalDateTime.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, + ValueType::castToLocalTimestampMicros, ValueType::toLocalTimestampMicros, ValueType::fromLocalTimestampMicros), + LOCAL_TIMESTAMP_NANOS(LocalDateTime.class, HoodieAvroWrapperUtils.PrimitiveWrapperType.LONG, + ValueType::castToLocalTimestampNanos, ValueType::toLocalTimestampNanos, ValueType::fromLocalTimestampNanos); + + private final Class<?> internalType; + private final HoodieAvroWrapperUtils.PrimitiveWrapperType primitiveWrapperType; + private final BiFunction<Object, ValueMetadata, Comparable<?>> standardize; + private final BiFunction<Comparable<?>, ValueMetadata, Comparable<?>> toComplex; + private final BiFunction<Comparable<?>, ValueMetadata, Comparable<?>> toPrimitive; + + ValueType(HoodieAvroWrapperUtils.PrimitiveWrapperType primitiveWrapperType, Function<Object, Object> single) { + this(primitiveWrapperType.getClazz(), + primitiveWrapperType, + (val, meta) -> (Comparable<?>) single.apply(val), + ValueType::passThrough, + ValueType::passThrough); + } + + ValueType(Class<?> internalType, + HoodieAvroWrapperUtils.PrimitiveWrapperType primitiveWrapperType, + BiFunction<Object, ValueMetadata, Comparable<?>> standardize, + BiFunction<Comparable<?>, ValueMetadata, Comparable<?>> toComplex, + BiFunction<Comparable<?>, ValueMetadata, Comparable<?>> toPrimitive) { + this.internalType = internalType; + this.primitiveWrapperType = primitiveWrapperType; + this.standardize = standardize; + this.toComplex = toComplex; + this.toPrimitive = toPrimitive; + } + + Comparable<?> standardizeJavaTypeAndPromote(Object val, ValueMetadata meta) { + if (val == null) { + return null; + } + return standardize.apply(val, meta); + } + + private Comparable<?> convertIntoPrimitive(Comparable<?> val, ValueMetadata meta) { + if (val == null) { + return null; + } + return toPrimitive.apply(val, meta); + } + + private Comparable<?> convertIntoComplex(Comparable<?> val, ValueMetadata meta) { + if (val == null) { + return null; + } + return toComplex.apply(val, meta); + } + + void validate(Object val) { + if (val == null) { + return; + } + + if (!internalType.isInstance(val)) { + throw new IllegalArgumentException(String.format( + "should be %s, but got %s", + internalType.getSimpleName(), + val.getClass().getSimpleName() + )); + } + } + + public Object wrapValue(Comparable<?> val, ValueMetadata meta) { + if (meta.getValueType() == V1) { + return primitiveWrapperType.wrap(val); + } + + if (val == null) { + return null; + } + if (!this.internalType.isInstance(val)) { + throw new IllegalArgumentException(String.format( + "should be %s, but got %s", + this.internalType.getSimpleName(), + val.getClass().getSimpleName() + )); + } + return primitiveWrapperType.wrap(convertIntoPrimitive(val, meta)); + } + + public Comparable<?> unwrapValue(Object val, ValueMetadata meta) { + if (meta.getValueType() == V1) { + return primitiveWrapperType.unwrap(val); + } + + if (val == null) { + return null; + } + + if (!primitiveWrapperType.getWrapperClass().isInstance(val)) { + if (!(val instanceof GenericRecord)) { + throw new IllegalArgumentException(String.format( + "should be %s, but got %s", + primitiveWrapperType.getWrapperClass().getSimpleName(), + val.getClass().getSimpleName() + )); + } else if (((GenericRecord) val).getSchema().getField("value") != null) { + return standardizeJavaTypeAndPromote(HoodieAvroWrapperUtils.unwrapGenericRecord(val), meta); + } else { + throw new IllegalArgumentException(String.format( + "should be %s, but got %s", + primitiveWrapperType.getWrapperClass().getSimpleName(), + val.getClass().getSimpleName() + )); + } + } + return convertIntoComplex(primitiveWrapperType.unwrap(val), meta); + } + + private static ValueType[] myEnumValues; + + public static ValueType fromInt(int i) { + if (ValueType.myEnumValues == null) { + ValueType.myEnumValues = ValueType.values(); + } + return ValueType.myEnumValues[i]; + } + + public static ValueType fromPrimitiveType(PrimitiveType primitiveType) { + if (primitiveType.getLogicalTypeAnnotation() != null) { + return LogicalTypeTokenParser.fromLogicalTypeAnnotation(primitiveType); + } + switch (primitiveType.getPrimitiveTypeName()) { + case INT64: + return ValueType.LONG; + case INT32: + return ValueType.INT; + case BOOLEAN: + return ValueType.BOOLEAN; + case BINARY: + return ValueType.BYTES; + case FLOAT: + return ValueType.FLOAT; + case DOUBLE: + return ValueType.DOUBLE; + case INT96: + // TODO: probably wrong + return ValueType.DECIMAL; + case FIXED_LEN_BYTE_ARRAY: + return ValueType.FIXED; + default: + throw new IllegalArgumentException("Unsupported primitive type: " + primitiveType.getPrimitiveTypeName()); + } + } + + public static ValueType fromSchema(Schema schema) { + switch (schema.getType()) { + case NULL: + if (schema.getLogicalType() == null) { + return ValueType.NULL; + } + throw new IllegalArgumentException("Unsupported logical type for Null: " + schema.getLogicalType()); + case BOOLEAN: + if (schema.getLogicalType() == null) { + return ValueType.BOOLEAN; + } + throw new IllegalArgumentException("Unsupported logical type for Boolean: " + schema.getLogicalType()); + case INT: + if (schema.getLogicalType() == null) { + return ValueType.INT; + } else if (schema.getLogicalType() instanceof LogicalTypes.Date) { + return ValueType.DATE; + } else if (schema.getLogicalType() instanceof LogicalTypes.TimeMillis) { + return ValueType.TIME_MILLIS; + } + throw new IllegalArgumentException("Unsupported logical type for Int: " + schema.getLogicalType()); + case LONG: + if (schema.getLogicalType() == null) { + return ValueType.LONG; + } else if (schema.getLogicalType() instanceof LogicalTypes.TimeMicros) { + return ValueType.TIME_MICROS; + } else if (schema.getLogicalType() instanceof LogicalTypes.TimestampMillis) { + return ValueType.TIMESTAMP_MILLIS; + } else if (schema.getLogicalType() instanceof LogicalTypes.TimestampMicros) { + return ValueType.TIMESTAMP_MICROS; + } else if (schema.getLogicalType() instanceof LogicalTypes.LocalTimestampMillis) { + return ValueType.LOCAL_TIMESTAMP_MILLIS; + } else if (schema.getLogicalType() instanceof LogicalTypes.LocalTimestampMicros) { + return ValueType.LOCAL_TIMESTAMP_MICROS; + } + throw new IllegalArgumentException("Unsupported logical type for Long: " + schema.getLogicalType()); + case FLOAT: + if (schema.getLogicalType() == null) { + return ValueType.FLOAT; + } + throw new IllegalArgumentException("Unsupported logical type for Float: " + schema.getLogicalType()); + case DOUBLE: + if (schema.getLogicalType() == null) { + return ValueType.DOUBLE; + } + throw new IllegalArgumentException("Unsupported logical type for Double: " + schema.getLogicalType()); + case BYTES: + if (schema.getLogicalType() == null) { + return ValueType.BYTES; + } else if (schema.getLogicalType() instanceof LogicalTypes.Decimal) { + return ValueType.DECIMAL; + } + throw new IllegalArgumentException("Unsupported logical type for Bytes: " + schema.getLogicalType()); + case STRING: + if (schema.getLogicalType() == null) { + return ValueType.STRING; + } else if (Objects.equals(schema.getLogicalType().getName(), LogicalTypes.uuid().getName())) { + return ValueType.UUID; + } + throw new IllegalArgumentException("Unsupported logical type for String: " + schema.getLogicalType()); + case FIXED: + if (schema.getLogicalType() == null) { + return ValueType.FIXED; + } else if (schema.getLogicalType() instanceof LogicalTypes.Decimal) { + return ValueType.DECIMAL; + } + throw new IllegalArgumentException("Unsupported logical type for Fixed: " + schema.getLogicalType()); + case UNION: + return fromSchema(AvroSchemaUtils.resolveNullableSchema(schema)); + default: + throw new IllegalArgumentException("Unsupported type: " + schema.getType()); + } + } + + // Casting to standardize types and also type promotion + private static Comparable<?> passThrough(Object val, ValueMetadata meta) { + return (Comparable<?>) val; + } + + private static Boolean castToBoolean(Object val) { + if (val instanceof Boolean) { + return (Boolean) val; + } else { + throw new UnsupportedOperationException("Unable to convert boolean: " + val.getClass()); + } + } + + private static Integer castToInteger(Object val) { + if (val == null) { + return null; + } + if (val instanceof Integer) { + return (Integer) val; + } else if (val instanceof Long) { + return ((Long) val).intValue(); + } else if (val instanceof Float) { + return ((Float) val).intValue(); + } else if (val instanceof Double) { + return ((Double) val).intValue(); + } else if (val instanceof Boolean) { + return ((Boolean) val) ? 1 : 0; + } else { + // best effort casting + return Integer.parseInt(val.toString()); + } + } + + private static Long castToLong(Object val) { + if (val == null) { + return null; + } + if (val instanceof Integer) { + return ((Integer) val).longValue(); + } else if (val instanceof Long) { + return ((Long) val); + } else if (val instanceof Float) { + return ((Float) val).longValue(); + } else if (val instanceof Double) { + return ((Double) val).longValue(); + } else if (val instanceof Boolean) { + return ((Boolean) val) ? 1L : 0L; + } else { + // best effort casting + return Long.parseLong(val.toString()); + } + } + + private static Float castToFloat(Object val) { + if (val == null) { + return null; + } + if (val instanceof Integer) { + return ((Integer) val).floatValue(); + } else if (val instanceof Long) { + return ((Long) val).floatValue(); + } else if (val instanceof Float) { + return (Float) val; + } else if (val instanceof Double) { + return ((Double) val).floatValue(); Review Comment: Got it. It should be removed if such evolution is not allowed. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
