feat: MDT Test framework without writing data files

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -2144,7 +2144,10 @@ public boolean isMetadataColumnStatsIndexEnabled() {
    * @return {@code true} if the partition stats index is enabled, {@code false} otherwise.
    */
   public boolean isPartitionStatsIndexEnabled() {
-    return isMetadataColumnStatsIndexEnabled();
+    if (isMetadataColumnStatsIndexEnabled()) {
+      return isMetadataTableEnabled() && getMetadataConfig().isPartitionStatsIndexEnabled();
+    }
+    return false;
   }
 
   /**

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

@@ -440,7 +440,8 @@ private boolean initializeFromFilesystem(String dataTableInstantTime, List<Metad
     while (iterator.hasNext()) {
       MetadataPartitionType partitionType = iterator.next();
       if (partitionType == PARTITION_STATS && !dataMetaClient.getTableConfig().isTablePartitioned()) {
-        // Partition stats index cannot be enabled for a non-partitioned table
+        LOG.debug("Partition stats index cannot be enabled for a non-partitioned table. Removing from initialization list. Please disable {}",
+            HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key());
         iterator.remove();
         this.enabledPartitionTypes.remove(partitionType);
       }
@@ -487,12 +488,6 @@ private boolean initializeFromFilesystem(String dataTableInstantTime, List<Metad
             initializeFilegroupsAndCommit(partitionType, relativePartitionPath, fileGroupCountAndRecordsPair, instantTimeForPartition);
             break;
           case PARTITION_STATS:
-            // For PARTITION_STATS, COLUMN_STATS should also be enabled
-            if (!dataWriteConfig.isMetadataColumnStatsIndexEnabled()) {
-              LOG.debug("Skipping partition stats initialization as column stats index is not enabled. Please enable {}",
-                  HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key());
-              continue;
-            }
             fileGroupCountAndRecordsPair = initializePartitionStatsIndex(lazyLatestMergedPartitionFileSliceList, tableSchema);
             initializeFilegroupsAndCommit(partitionType, PARTITION_STATS.getPartitionPath(), fileGroupCountAndRecordsPair, instantTimeForPartition);
             break;
@@ -697,19 +692,19 @@ private Lazy<List<Pair<String, FileSlice>>> getLazyLatestMergedPartitionFileSlic
     });
   }
 
-  void initializeFilegroupsAndCommit(MetadataPartitionType partitionType,
-                                     String relativePartitionPath,
-                                     Pair<Integer, HoodieData<HoodieRecord>> fileGroupCountAndRecordsPair,
-                                     String instantTimeForPartition) throws IOException {
+  public void initializeFilegroupsAndCommit(MetadataPartitionType partitionType,
+                                            String relativePartitionPath,
+                                            Pair<Integer, HoodieData<HoodieRecord>> fileGroupCountAndRecordsPair,
+                                            String instantTimeForPartition) throws IOException {
     initializeFilegroupsAndCommit(partitionType, relativePartitionPath, fileGroupCountAndRecordsPair,
         instantTimeForPartition, Collections.emptyList());
   }
 
-  void initializeFilegroupsAndCommit(MetadataPartitionType partitionType,
-                                     String relativePartitionPath,
-                                     Pair<Integer, HoodieData<HoodieRecord>> fileGroupCountAndRecordsPair,
-                                     String instantTimeForPartition,
-                                     List<String> columnsToIndex) throws IOException {
+  public void initializeFilegroupsAndCommit(MetadataPartitionType partitionType,
+                                            String relativePartitionPath,
+                                            Pair<Integer, HoodieData<HoodieRecord>> fileGroupCountAndRecordsPair,
+                                            String instantTimeForPartition,
+                                            List<String> columnsToIndex) throws IOException {
     String partitionTypeName = partitionType.name();
     LOG.info("Initializing {} index with {} mappings", partitionTypeName, fileGroupCountAndRecordsPair.getKey());
     HoodieTimer partitionInitTimer = HoodieTimer.start();
@@ -903,6 +898,9 @@ private Pair<Integer, HoodieData<HoodieRecord>> initializeFilesPartition(Map<Str
     final int totalDataFilesCount = partitionIdToAllFilesMap.values().stream().mapToInt(Map::size).sum();
     LOG.info("Committing total {} partitions and {} files to metadata", partitions.size(), totalDataFilesCount);
 
+    if (partitions.isEmpty()) {
+      return Pair.of(fileGroupCount, engineContext.emptyHoodieData());
+    }
     // Record which saves the list of all partitions
     HoodieRecord record = HoodieMetadataPayload.createPartitionListRecord(partitions);
     HoodieData<HoodieRecord> allPartitionsRecord = engineContext.parallelize(Collections.singletonList(record), 1);
@@ -938,7 +936,7 @@ String getTimelineHistoryPath() {
     return TIMELINE_HISTORY_PATH.defaultValue();
   }
 
-  private HoodieTableMetaClient initializeMetaClient() throws IOException {
+  protected HoodieTableMetaClient initializeMetaClient() throws IOException {
     HoodieTableMetaClient.newTableBuilder()
         .setTableType(HoodieTableType.MERGE_ON_READ)
         .setTableName(dataWriteConfig.getTableName() + METADATA_TABLE_NAME_SUFFIX)
@@ -1047,7 +1045,7 @@ private List<DirectoryInfo> listAllPartitionsFromMDT(String initializationTime,
    * File groups will be named as :
    * record-index-bucket-0000, .... -> ..., record-index-bucket-0009
    */
-  private void initializeFileGroups(HoodieTableMetaClient dataMetaClient, MetadataPartitionType metadataPartition, String instantTime,
+  public void initializeFileGroups(HoodieTableMetaClient dataMetaClient, MetadataPartitionType metadataPartition, String instantTime,
                                     int fileGroupCount, String relativePartitionPath, Option<String> dataPartitionName) throws IOException {
 
     // Archival of data table has a dependency on compaction(base files) in metadata table.

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

@@ -44,6 +44,7 @@
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.model.HoodieWriteStat;
 import org.apache.hudi.common.model.WriteConcurrencyMode;
+// import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.schema.HoodieSchemaUtils;
@@ -390,9 +391,8 @@ public static Map<String, HoodieData<HoodieRecord>> convertMetadataToRecords(Hoo
           dataMetaClient, metadataConfig, recordTypeOpt);
       partitionToRecordsMap.put(MetadataPartitionType.COLUMN_STATS.getPartitionPath(), metadataColumnStatsRDD);
     }
+
     if (enabledPartitionTypes.contains(MetadataPartitionType.PARTITION_STATS.getPartitionPath())) {
-      checkState(MetadataPartitionType.COLUMN_STATS.isMetadataPartitionAvailable(dataMetaClient),
-          "Column stats partition must be enabled to generate partition stats. Please enable: " + HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key());
       // Generate Hoodie Pair data of partition name and list of column range metadata for all the files in that partition
       boolean isDeletePartition = commitMetadata.getOperationType().equals(WriteOperationType.DELETE_PARTITION);
       final HoodieData<HoodieRecord> partitionStatsRDD = convertMetadataToPartitionStatRecords(

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

@@ -29,5 +29,9 @@ public interface MetadataTableFileGroupIndexParser extends Serializable {
 
   int getFileGroupIndex(String fileID);
 
+  default int getFileGroupIndex(String partitionPath, int fileGroupIndexInPartition) {
+    return fileGroupIndexInPartition;
+  }
+
   int getNumberOfFileGroups();
 }

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

@@ -74,7 +74,6 @@
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieDuplicateDataFileDetectedException;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.exception.HoodieInsertException;
@@ -110,7 +109,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
@@ -797,64 +795,8 @@ void reconcileAgainstMarkers(HoodieEngineContext context,
                                boolean consistencyCheckEnabled,
                                boolean shouldFailOnDuplicateDataFileDetection,
                                WriteMarkers markers) throws HoodieIOException {
-    try {
-      // Reconcile marker and data files with WriteStats so that partially written data-files due to failed
-      // (but succeeded on retry) tasks are removed.
-      String basePath = getMetaClient().getBasePath().toString();
-
-      if (!markers.doesMarkerDirExist()) {
-        // can happen if it was an empty write say.
-        return;
-      }
-
-      // Ignores log file appended for update, since they are already fail-safe.
-      // but new created log files should be included.
-      Set<String> invalidDataPaths = getInvalidDataPaths(markers);
-      Set<String> validDataPaths = stats.stream()
-          .map(HoodieWriteStat::getPath)
-          .collect(Collectors.toSet());
-      Set<String> validCdcDataPaths = stats.stream()
-          .map(HoodieWriteStat::getCdcStats)
-          .filter(Objects::nonNull)
-          .flatMap(cdcStat -> cdcStat.keySet().stream())
-          .collect(Collectors.toSet());
-
-      // Contains list of partially created files. These needs to be cleaned up.
-      invalidDataPaths.removeAll(validDataPaths);
-      invalidDataPaths.removeAll(validCdcDataPaths);
-
-      if (!invalidDataPaths.isEmpty()) {
-        if (shouldFailOnDuplicateDataFileDetection) {
-          throw new HoodieDuplicateDataFileDetectedException("Duplicate data files detected " + invalidDataPaths);
-        }
-
-        log.info("Removing duplicate files created due to task retries before committing. Paths=" + invalidDataPaths);
-        Map<String, List<Pair<String, String>>> invalidPathsByPartition = invalidDataPaths.stream()
-            .map(dp ->
-                Pair.of(new StoragePath(basePath, dp).getParent().toString(),
-                    new StoragePath(basePath, dp).toString()))
-            .collect(Collectors.groupingBy(Pair::getKey));
-
-        // Ensure all files in delete list is actually present. This is mandatory for an eventually consistent FS.
-        // Otherwise, we may miss deleting such files. If files are not found even after retries, fail the commit
-        if (consistencyCheckEnabled) {
-          // This will either ensure all files to be deleted are present.
-          waitForAllFiles(context, invalidPathsByPartition, FileVisibility.APPEAR);
-        }
-
-        // Now delete partially written files
-        context.setJobStatus(this.getClass().getSimpleName(), "Delete all partially written files: " + config.getTableName());
-        deleteInvalidFilesByPartitions(context, invalidPathsByPartition);
-
-        // Now ensure the deleted files disappear
-        if (consistencyCheckEnabled) {
-          // This will either ensure all files to be deleted are absent.
-          waitForAllFiles(context, invalidPathsByPartition, FileVisibility.DISAPPEAR);
-        }
-      }
-    } catch (IOException ioe) {
-      throw new HoodieIOException(ioe.getMessage(), ioe);
-    }
+    log.warn("Skipping reconcile markers for instant: {}", instantTs);
+    return;
   }
 
   /**

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/MetadataWriterTestUtils.java

@@ -0,0 +1,53 @@
+/*
+ * 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.metadata;
+
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Test utility class to access protected methods from HoodieBackedTableMetadataWriter.
+ * This class is in the same package to access protected methods without duplication.
+ */
+public class MetadataWriterTestUtils {
+
+  /**
+   * Tag records with location using the metadata writer's tagRecordsWithLocation method.
+   * This is a wrapper around the protected method to make it accessible from tests.
+   *
+   * @param metadataWriter The metadata writer instance
+   * @param partitionRecordsMap Map of partition path to records
+   * @param isInitializing Whether this is during initialization
+   * @return Pair of tagged records and file group IDs
+   */
+  @SuppressWarnings("rawtypes")
+  public static <I, O> Pair<HoodieData<HoodieRecord>, List<HoodieFileGroupId>> tagRecordsWithLocation(
+      HoodieBackedTableMetadataWriter<I, O> metadataWriter,
+      Map<String, HoodieData<HoodieRecord>> partitionRecordsMap,
+      boolean isInitializing) {
+    // Access the protected method - this works because we're in the same package
+    return metadataWriter.tagRecordsWithLocation(partitionRecordsMap, isInitializing);
+  }
+}
+

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieBackedTableMetadataWriter.java

@@ -118,12 +118,12 @@ public static HoodieTableMetadataWriter create(StorageConfiguration<?> conf, Hoo
     this(hadoopConf, writeConfig, failedWritesCleaningPolicy, engineContext, inflightInstantTimestamp, false);
   }
 
-  SparkHoodieBackedTableMetadataWriter(StorageConfiguration<?> hadoopConf,
-                                       HoodieWriteConfig writeConfig,
-                                       HoodieFailedWritesCleaningPolicy failedWritesCleaningPolicy,
-                                       HoodieEngineContext engineContext,
-                                       Option<String> inflightInstantTimestamp,
-                                       boolean streamingWrites) {
+  protected SparkHoodieBackedTableMetadataWriter(StorageConfiguration<?> hadoopConf,
+                                                 HoodieWriteConfig writeConfig,
+                                                 HoodieFailedWritesCleaningPolicy failedWritesCleaningPolicy,
+                                                 HoodieEngineContext engineContext,
+                                                 Option<String> inflightInstantTimestamp,
+                                                 boolean streamingWrites) {
     super(hadoopConf, writeConfig, failedWritesCleaningPolicy, engineContext, inflightInstantTimestamp, streamingWrites);
   }
 

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestCleanPlanExecutor.java

@@ -341,7 +341,8 @@ public void testKeepLatestFileVersions() throws Exception {
   public void testKeepLatestFileVersionsWithBootstrapFileClean() throws Exception {
     HoodieWriteConfig config =
         HoodieWriteConfig.newBuilder().withPath(basePath)
-            .withMetadataConfig(HoodieMetadataConfig.newBuilder().withMetadataIndexColumnStats(false).build())
+            .withMetadataConfig(HoodieMetadataConfig.newBuilder().withMetadataIndexColumnStats(false)
+                .withMetadataIndexPartitionStats(false).build())
             .withCleanConfig(HoodieCleanConfig.newBuilder()
                 .withCleanBootstrapBaseFileEnabled(true)
                 .withCleanerParallelism(1)

hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java

@@ -445,6 +445,21 @@ public final class HoodieMetadataConfig extends HoodieConfig {
       .sinceVersion("1.0.1")
       .withDocumentation("Options for the expression index, e.g. \"expr='from_unixtime', format='yyyy-MM-dd'\"");
 
+  public static final ConfigProperty<Boolean> ENABLE_METADATA_INDEX_PARTITION_STATS = ConfigProperty
+      .key(METADATA_PREFIX + ".index.partition.stats.enable")
+      // The defaultValue(false) here is the initial default, but it's overridden later based on
+      // column stats setting.
+      .defaultValue(false)
+      .sinceVersion("1.0.0")
+      .withDocumentation("Enable aggregating stats for each column at the storage partition level. "
+          + "Enabling this can improve query performance by leveraging partition and column stats "
+          + "for (partition) filtering. "
+          + "Important: The default value for this configuration is dynamically set based on the "
+          + "effective value of " + ENABLE_METADATA_INDEX_COLUMN_STATS.key() + ". If column stats "
+          + "index is enabled (default for Spark engine), partition stats indexing will also be "
+          + "enabled by default. Conversely, if column stats indexing is disabled (default for "
+          + "Flink and Java engines), partition stats indexing will also be disabled by default.");
+
   public static final ConfigProperty<Integer> METADATA_INDEX_PARTITION_STATS_FILE_GROUP_COUNT = ConfigProperty
       .key(METADATA_PREFIX + ".index.partition.stats.file.group.count")
       .defaultValue(1)
@@ -821,7 +836,7 @@ private Map<String, String> getExpressionIndexOptions(String configValue) {
   }
 
   public boolean isPartitionStatsIndexEnabled() {
-    return getBooleanOrDefault(ENABLE_METADATA_INDEX_COLUMN_STATS);
+    return getBooleanOrDefault(ENABLE_METADATA_INDEX_PARTITION_STATS);
   }
 
   public int getPartitionStatsIndexFileGroupCount() {
@@ -1124,6 +1139,11 @@ public Builder withExpressionIndexOptions(Map<String, String> options) {
       return this;
     }
 
+    public Builder withMetadataIndexPartitionStats(boolean enable) {
+      metadataConfig.setValue(ENABLE_METADATA_INDEX_PARTITION_STATS, String.valueOf(enable));
+      return this;
+    }
+
     public Builder withMetadataIndexPartitionStatsFileGroupCount(int fileGroupCount) {
       metadataConfig.setValue(METADATA_INDEX_PARTITION_STATS_FILE_GROUP_COUNT, String.valueOf(fileGroupCount));
       return this;
@@ -1192,6 +1212,7 @@ public Builder withRepartitionDefaultPartitions(int defaultPartitions) {
     public HoodieMetadataConfig build() {
       metadataConfig.setDefaultValue(ENABLE, getDefaultMetadataEnable(engineType));
       metadataConfig.setDefaultValue(ENABLE_METADATA_INDEX_COLUMN_STATS, getDefaultColStatsEnable(engineType));
+      metadataConfig.setDefaultValue(ENABLE_METADATA_INDEX_PARTITION_STATS, metadataConfig.isColumnStatsIndexEnabled());
       metadataConfig.setDefaultValue(SECONDARY_INDEX_ENABLE_PROP, getDefaultSecondaryIndexEnable(engineType));
       metadataConfig.setDefaultValue(STREAMING_WRITE_ENABLED, getDefaultForStreamingWriteEnabled(engineType));
       // fix me: disable when schema on read is enabled.
@@ -1248,13 +1269,6 @@ private boolean getDefaultSecondaryIndexEnable(EngineType engineType) {
     }
   }
 
-  /**
-   * The config is now deprecated. Partition stats are configured using the column stats config itself.
-   */
-  @Deprecated
-  public static final String ENABLE_METADATA_INDEX_PARTITION_STATS =
-      METADATA_PREFIX + ".index.partition.stats.enable";
-
   /**
    * @deprecated Use {@link #ENABLE} and its methods.
    */