feat(metadata): Improve Logical Type Handling on Col Stats

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java

@@ -37,6 +37,7 @@
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.index.HoodieIndexUtils;
 import org.apache.hudi.io.HoodieRangeInfoHandle;
+import org.apache.hudi.stats.ValueMetadata;
 import org.apache.hudi.table.HoodieTable;
 
 import org.slf4j.Logger;
@@ -52,7 +53,6 @@
 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.mapping;
 import static java.util.stream.Collectors.toList;
-import static org.apache.hudi.avro.HoodieAvroWrapperUtils.unwrapAvroValueWrapper;
 import static org.apache.hudi.common.util.CollectionUtils.isNullOrEmpty;
 import static org.apache.hudi.index.HoodieIndexUtils.getLatestBaseFilesForAllPartitions;
 import static org.apache.hudi.metadata.MetadataPartitionType.COLUMN_STATS;
@@ -231,12 +231,13 @@ protected List<Pair<String, BloomIndexFileInfo>> loadColumnRangesFromMetaIndex(
     List<Pair<String, BloomIndexFileInfo>> result = new ArrayList<>(fileToColumnStatsMap.size());
 
     for (Map.Entry<Pair<String, String>, HoodieMetadataColumnStats> entry : fileToColumnStatsMap.entrySet()) {
+      ValueMetadata valueMetadata = ValueMetadata.getValueMetadata(entry.getValue().getValueType());
       result.add(Pair.of(entry.getKey().getLeft(),
           new BloomIndexFileInfo(
               partitionAndFileNameToFileId.get(entry.getKey()),
               // NOTE: Here we assume that the type of the primary key field is string
-              unwrapAvroValueWrapper(entry.getValue().getMinValue()).toString(),
-              unwrapAvroValueWrapper(entry.getValue().getMaxValue()).toString()
+              valueMetadata.unwrapValue(entry.getValue().getMinValue()).toString(),
+              valueMetadata.unwrapValue(entry.getValue().getMaxValue()).toString()
           )));
     }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java

@@ -28,7 +28,6 @@
 import org.apache.hudi.common.model.DeleteRecord;
 import org.apache.hudi.common.model.FileSlice;
 import org.apache.hudi.common.model.HoodieBaseFile;
-import org.apache.hudi.common.model.HoodieColumnRangeMetadata;
 import org.apache.hudi.common.model.HoodieDeltaWriteStat;
 import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.model.HoodiePartitionMetadata;
@@ -59,7 +58,9 @@
 import org.apache.hudi.exception.HoodieAppendException;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieUpsertException;
+import org.apache.hudi.metadata.HoodieIndexVersion;
 import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.stats.HoodieColumnRangeMetadata;
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.util.CommonClientUtils;
@@ -83,6 +84,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
 import static org.apache.hudi.metadata.HoodieTableMetadataUtil.collectColumnRangeMetadata;
 
 /**
@@ -432,15 +434,17 @@ protected void processAppendResult(AppendResult result, Option<HoodieLogBlock> d
     updateWriteStatus(result, stat);
 
     if (config.isMetadataColumnStatsIndexEnabled()) {
+      HoodieIndexVersion indexVersion = HoodieTableMetadataUtil.existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS, hoodieTable.getMetaClient());
       Set<String> columnsToIndexSet = new HashSet<>(HoodieTableMetadataUtil
           .getColumnsToIndex(hoodieTable.getMetaClient().getTableConfig(),
               config.getMetadataConfig(), Lazy.eagerly(Option.of(writeSchemaWithMetaFields)),
-              Option.of(this.recordMerger.getRecordType())).keySet());
+              Option.of(this.recordMerger.getRecordType()), indexVersion).keySet());
       final List<Pair<String, Schema.Field>> fieldsToIndex = columnsToIndexSet.stream()
           .map(fieldName -> HoodieAvroUtils.getSchemaForField(writeSchemaWithMetaFields, fieldName)).collect(Collectors.toList());
       try {
         Map<String, HoodieColumnRangeMetadata<Comparable>> columnRangeMetadataMap =
-            collectColumnRangeMetadata(recordList.iterator(), fieldsToIndex, stat.getPath(), writeSchemaWithMetaFields, storage.getConf());
+            collectColumnRangeMetadata(recordList.iterator(), fieldsToIndex, stat.getPath(), writeSchemaWithMetaFields, storage.getConf(),
+                indexVersion);
         stat.putRecordsStats(columnRangeMetadataMap);
       } catch (HoodieException e) {
         throw new HoodieAppendException("Failed to extract append result", e);

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java

@@ -35,7 +35,6 @@
 import org.apache.hudi.table.HoodieTable;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.parquet.avro.AvroSchemaConverter;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
 import org.apache.parquet.schema.MessageType;
 import org.slf4j.Logger;
@@ -45,6 +44,8 @@
 import java.util.Collections;
 import java.util.List;
 
+import static org.apache.parquet.avro.HoodieAvroParquetSchemaConverter.getAvroSchemaConverter;
+
 /**
  * Compared to other Write Handles, HoodieBinaryCopyHandle merge multiple inputFiles into a single outputFile without performing
  * extra operations like data serialization/deserialization or compression/decompression.
@@ -75,12 +76,12 @@ private MessageType getWriteSchema(HoodieWriteConfig config, List<StoragePath> i
         return fileSchema;
       } catch (Exception e) {
         LOG.error("Failed to read schema from input file", e);
-        throw new HoodieIOException("Failed to read schema from input file when schema evolution is disabled: " + inputFiles.get(0), 
+        throw new HoodieIOException("Failed to read schema from input file when schema evolution is disabled: " + inputFiles.get(0),
             e instanceof IOException ? (IOException) e : new IOException(e));
       }
     } else {
       // Default behavior: use the table's write schema for evolution
-      return new AvroSchemaConverter(conf).convert(writeSchemaWithMetaFields);
+      return getAvroSchemaConverter(conf).convert(writeSchemaWithMetaFields);
     }
   }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java

@@ -123,6 +123,7 @@
 import static org.apache.hudi.metadata.HoodieMetadataWriteUtils.createMetadataWriteConfig;
 import static org.apache.hudi.metadata.HoodieTableMetadata.METADATA_TABLE_NAME_SUFFIX;
 import static org.apache.hudi.metadata.HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
 import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX;
 import static org.apache.hudi.metadata.HoodieTableMetadataUtil.createRecordIndexDefinition;
 import static org.apache.hudi.metadata.HoodieTableMetadataUtil.existingIndexVersionOrDefault;
@@ -578,10 +579,11 @@ private Pair<List<String>, Pair<Integer, HoodieData<HoodieRecord>>> initializeCo
     if (partitionIdToAllFilesMap.isEmpty()) {
       return Pair.of(Collections.emptyList(), Pair.of(fileGroupCount, engineContext.emptyHoodieData()));
     }
+    HoodieIndexVersion columnStatsIndexVersion = existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS, dataMetaClient);
     // Find the columns to index
     final List<String> columnsToIndex = new ArrayList<>(HoodieTableMetadataUtil.getColumnsToIndex(dataMetaClient.getTableConfig(),
         dataWriteConfig.getMetadataConfig(), tableSchema, true,
-        Option.of(dataWriteConfig.getRecordMerger().getRecordType())).keySet());
+        Option.of(dataWriteConfig.getRecordMerger().getRecordType()), columnStatsIndexVersion).keySet());
 
     if (columnsToIndex.isEmpty()) {
       // this can only happen if meta fields are disabled and cols to index is not explicitly overridden.

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/NineToEightDowngradeHandler.java

@@ -83,8 +83,8 @@ public UpgradeDowngrade.TableConfigChangeSet downgrade(HoodieWriteConfig config,
         && isComplexKeyGeneratorWithSingleRecordKeyField(metaClient.getTableConfig())) {
       throw new HoodieUpgradeDowngradeException(getComplexKeygenErrorMessage("downgrade"));
     }
-    // Handle secondary index.
-    UpgradeDowngradeUtils.dropNonV1SecondaryIndexPartitions(
+    // Handle index Changes
+    UpgradeDowngradeUtils.dropNonV1IndexPartitions(
         config, context, table, upgradeDowngradeHelper, "downgrading from table version nine to eight");
     // Update table properties.
     Set<ConfigProperty> propertiesToRemove = new HashSet<>();

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngradeUtils.java

@@ -56,7 +56,6 @@
 import org.apache.hudi.metadata.HoodieTableMetadata;
 import org.apache.hudi.metadata.HoodieTableMetadataUtil;
 import org.apache.hudi.metadata.HoodieIndexVersion;
-import org.apache.hudi.metadata.MetadataPartitionType;
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
@@ -285,23 +284,15 @@ static boolean isMetadataTableBehindDataTable(HoodieWriteConfig config,
    * @param table         Hoodie table
    * @param operationType Type of operation (upgrade/downgrade)
    */
-  public static void dropNonV1SecondaryIndexPartitions(HoodieWriteConfig config, HoodieEngineContext context,
-                                                       HoodieTable table, SupportsUpgradeDowngrade upgradeDowngradeHelper, String operationType) {
+  public static void dropNonV1IndexPartitions(HoodieWriteConfig config, HoodieEngineContext context,
+                                              HoodieTable table, SupportsUpgradeDowngrade upgradeDowngradeHelper, String operationType) {
     HoodieTableMetaClient metaClient = table.getMetaClient();
     try (BaseHoodieWriteClient writeClient = upgradeDowngradeHelper.getWriteClient(config, context)) {
       List<String> mdtPartitions = metaClient.getTableConfig().getMetadataPartitions()
           .stream()
-          .filter(partition -> {
-            // Only drop secondary indexes that are not V1
-            return metaClient.getIndexForMetadataPartition(partition)
-                .map(indexDef -> {
-                  if (MetadataPartitionType.fromPartitionPath(indexDef.getIndexName()).equals(MetadataPartitionType.SECONDARY_INDEX)) {
-                    return HoodieIndexVersion.V1.lowerThan(indexDef.getVersion());
-                  }
-                  return false;
-                })
-                .orElse(false);
-          })
+          .filter(partition -> metaClient.getIndexForMetadataPartition(partition)
+              .map(indexDef -> HoodieIndexVersion.V1.lowerThan(indexDef.getVersion()))
+              .orElse(false))
           .collect(Collectors.toList());
       LOG.info("Dropping from MDT partitions for {}: {}", operationType, mdtPartitions);
       if (!mdtPartitions.isEmpty()) {

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestNineToEightDowngradeHandler.java

@@ -266,7 +266,7 @@ void testDowngradeDropsOnlyV2OrAboveIndexes() {
       )).thenAnswer(invocation -> null); // Do nothing
 
       // Mock the dropNonV1SecondaryIndexPartitions to simulate dropping V2 indexes
-      mockedUtils.when(() -> UpgradeDowngradeUtils.dropNonV1SecondaryIndexPartitions(
+      mockedUtils.when(() -> UpgradeDowngradeUtils.dropNonV1IndexPartitions(
           eq(config),
           eq(context),
           eq(table),

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/log/block/HoodieFlinkParquetDataBlock.java

@@ -19,7 +19,6 @@
 package org.apache.hudi.io.log.block;
 
 import org.apache.hudi.avro.AvroSchemaCache;
-import org.apache.hudi.common.model.HoodieColumnRangeMetadata;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.table.log.block.HoodieLogBlock;
 import org.apache.hudi.common.table.log.block.HoodieParquetDataBlock;
@@ -29,6 +28,8 @@
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.io.storage.ColumnRangeMetadataProvider;
 import org.apache.hudi.io.storage.HoodieIOFactory;
+import org.apache.hudi.metadata.HoodieIndexVersion;
+import org.apache.hudi.stats.HoodieColumnRangeMetadata;
 import org.apache.hudi.storage.HoodieStorage;
 
 import org.apache.avro.Schema;
@@ -97,10 +98,10 @@ public ByteArrayOutputStream getContentBytes(HoodieStorage storage) throws IOExc
   }
 
   @Override
-  public Map<String, HoodieColumnRangeMetadata<Comparable>> getColumnRangeMeta(String filePath) {
+  public Map<String, HoodieColumnRangeMetadata<Comparable>> getColumnRangeMeta(String filePath, HoodieIndexVersion indexVersion) {
     ValidationUtils.checkArgument(parquetMetadata != null, "parquetMetadata should not be null.");
     ParquetUtils parquetUtils = new ParquetUtils();
-    List<HoodieColumnRangeMetadata<Comparable>> columnMetaList = parquetUtils.readColumnStatsFromMetadata(parquetMetadata, filePath, Option.empty());
+    List<HoodieColumnRangeMetadata<Comparable>> columnMetaList = parquetUtils.readColumnStatsFromMetadata(parquetMetadata, filePath, Option.empty(), indexVersion);
     return columnMetaList.stream().collect(Collectors.toMap(HoodieColumnRangeMetadata::getColumnName, colMeta -> colMeta));
   }
 }

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/v2/RowDataLogWriteHandle.java

@@ -19,7 +19,6 @@
 package org.apache.hudi.io.v2;
 
 import org.apache.hudi.common.engine.TaskContextSupplier;
-import org.apache.hudi.common.model.HoodieColumnRangeMetadata;
 import org.apache.hudi.common.model.HoodieDeltaWriteStat;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.table.log.AppendResult;
@@ -36,7 +35,9 @@
 import org.apache.hudi.io.log.block.HoodieFlinkAvroDataBlock;
 import org.apache.hudi.io.log.block.HoodieFlinkParquetDataBlock;
 import org.apache.hudi.io.storage.ColumnRangeMetadataProvider;
+import org.apache.hudi.metadata.HoodieIndexVersion;
 import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.stats.HoodieColumnRangeMetadata;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.action.commit.BucketType;
 import org.apache.hudi.util.Lazy;
@@ -55,6 +56,7 @@
 import static org.apache.hudi.common.config.HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME;
 import static org.apache.hudi.common.config.HoodieStorageConfig.PARQUET_COMPRESSION_RATIO_FRACTION;
 import static org.apache.hudi.common.config.HoodieStorageConfig.PARQUET_DICTIONARY_ENABLED;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
 
 /**
  * A write handle that supports creating a log file and writing records based on record Iterator.
@@ -107,24 +109,22 @@ protected void processAppendResult(AppendResult result, Option<HoodieLogBlock> d
 
     // for parquet data block, we can get column stats from parquet footer directly.
     if (config.isMetadataColumnStatsIndexEnabled()) {
+      HoodieIndexVersion indexVersion = HoodieTableMetadataUtil.existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS, hoodieTable.getMetaClient());
       Set<String> columnsToIndexSet = new HashSet<>(HoodieTableMetadataUtil
           .getColumnsToIndex(hoodieTable.getMetaClient().getTableConfig(),
               config.getMetadataConfig(), Lazy.eagerly(Option.of(writeSchemaWithMetaFields)),
-              Option.of(HoodieRecord.HoodieRecordType.FLINK)).keySet());
+              Option.of(HoodieRecord.HoodieRecordType.FLINK), indexVersion).keySet());
 
       Map<String, HoodieColumnRangeMetadata<Comparable>> columnRangeMetadata;
       if (dataBlock.isEmpty()) {
         // only delete block exists
         columnRangeMetadata = new HashMap<>();
-        for (String col: columnsToIndexSet) {
-          columnRangeMetadata.put(col, HoodieColumnRangeMetadata.create(
-              stat.getPath(), col, null, null, 0L, 0L, 0L, 0L));
-        }
+        columnsToIndexSet.forEach(col -> columnRangeMetadata.put(col, HoodieColumnRangeMetadata.createEmpty(stat.getPath(), col, indexVersion)));
       } else {
         ValidationUtils.checkArgument(dataBlock.get() instanceof ColumnRangeMetadataProvider,
             "Log block for Flink ingestion should always be an instance of ColumnRangeMetadataProvider for collecting column stats efficiently.");
         columnRangeMetadata =
-            ((ColumnRangeMetadataProvider) dataBlock.get()).getColumnRangeMeta(stat.getPath()).entrySet().stream()
+            ((ColumnRangeMetadataProvider) dataBlock.get()).getColumnRangeMeta(stat.getPath(), indexVersion).entrySet().stream()
                 .filter(e -> columnsToIndexSet.contains(e.getKey()))
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
       }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkInternalSchemaConverter.java

@@ -57,6 +57,7 @@
 import org.apache.spark.sql.types.StructField;
 import org.apache.spark.sql.types.StructType;
 import org.apache.spark.sql.types.StructType$;
+import org.apache.spark.sql.types.TimestampNTZType$;
 import org.apache.spark.sql.types.TimestampType;
 import org.apache.spark.sql.types.TimestampType$;
 import org.apache.spark.sql.types.UserDefinedType;
@@ -267,10 +268,14 @@ private static DataType constructSparkSchemaFromType(Type type) {
       case DATE:
         return DateType$.MODULE$;
       case TIME:
+      case TIME_MILLIS:
         throw new UnsupportedOperationException(String.format("cannot convert %s type to Spark", type));
       case TIMESTAMP:
-        // todo support TimeStampNTZ
+      case TIMESTAMP_MILLIS:
         return TimestampType$.MODULE$;
+      case LOCAL_TIMESTAMP_MILLIS:
+      case LOCAL_TIMESTAMP_MICROS:
+        return TimestampNTZType$.MODULE$;
       case STRING:
         return StringType$.MODULE$;
       case UUID:
@@ -280,7 +285,9 @@ private static DataType constructSparkSchemaFromType(Type type) {
       case BINARY:
         return BinaryType$.MODULE$;
       case DECIMAL:
-        Types.DecimalType decimal = (Types.DecimalType) type;
+      case DECIMAL_BYTES:
+      case DECIMAL_FIXED:
+        Types.DecimalBase decimal = (Types.DecimalBase) type;
         return DecimalType$.MODULE$.apply(decimal.precision(), decimal.scale());
       default:
         throw new UnsupportedOperationException(String.format("cannot convert unknown type: %s to Spark", type));