[HUDI-2276] Enable metadata table by default for readers and writers

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiveLog.java

@@ -22,6 +22,7 @@
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
 import org.apache.hudi.avro.model.HoodieCleanerPlan;
 import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.client.SparkRDDWriteClient;
 import org.apache.hudi.client.utils.MetadataConversionUtils;
 import org.apache.hudi.common.HoodieCleanStat;
 import org.apache.hudi.common.fs.HoodieWrapperFileSystem;
@@ -96,6 +97,7 @@ public void testArchiveEmptyTable() throws IOException {
     HoodieWriteConfig cfg =
         HoodieWriteConfig.newBuilder().withPath(basePath).withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA)
             .withParallelism(2, 2).forTable("test-trip-table").build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
     HoodieTable table = HoodieSparkTable.create(cfg, context, metaClient);
     HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
@@ -109,7 +111,7 @@ public void testArchiveTableWithArchival() throws IOException {
         .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2)
         .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 4).build())
         .forTable("test-trip-table").build();
-    HoodieTestUtils.init(hadoopConf, basePath);
+    initMetadataTable(cfg);
     // Requested Compaction
     HoodieTestDataGenerator.createCompactionAuxiliaryMetadata(basePath,
         new HoodieInstant(State.REQUESTED, HoodieTimeline.COMPACTION_ACTION, "100"), wrapperFs.getConf());
@@ -235,6 +237,7 @@ public void testArchiveTableWithNoArchival() throws IOException {
         .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2).forTable("test-trip-table")
         .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 5).build())
         .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
     HoodieTable table = HoodieSparkTable.create(cfg, context, metaClient);
     HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
@@ -301,6 +304,7 @@ public void testArchiveCommitSafety() throws IOException {
         .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2).forTable("test-trip-table")
         .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 5).build())
         .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
     HoodieTable table = HoodieSparkTable.create(cfg, context, metaClient);
     HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
@@ -328,14 +332,15 @@ public void testArchiveCommitSavepointNoHole() throws IOException {
         .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2).forTable("test-trip-table")
         .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 5).build())
         .build();
-
+    initMetadataTable(cfg);
     HoodieTestDataGenerator.createCommitFile(basePath, "100", wrapperFs.getConf());
     HoodieTestDataGenerator.createCommitFile(basePath, "101", wrapperFs.getConf());
     HoodieTestDataGenerator.createSavepointFile(basePath, "101", wrapperFs.getConf());
     HoodieTestDataGenerator.createCommitFile(basePath, "102", wrapperFs.getConf());
     HoodieTestDataGenerator.createCommitFile(basePath, "103", wrapperFs.getConf());
     HoodieTestDataGenerator.createCommitFile(basePath, "104", wrapperFs.getConf());
     HoodieTestDataGenerator.createCommitFile(basePath, "105", wrapperFs.getConf());
+
     HoodieTable table = HoodieSparkTable.create(cfg, context);
     HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
 
@@ -359,6 +364,7 @@ public void testArchiveRollbacks() throws IOException {
             .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2).forTable("test-trip-table")
             .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 3).build())
             .build();
+    initMetadataTable(cfg);
 
     createCommitAndRollbackFile("100", "101", false);
     createCommitAndRollbackFile("102", "103", false);
@@ -388,6 +394,8 @@ public void testArchiveCommitCompactionNoHole() throws IOException {
         .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2).forTable("test-trip-table")
         .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 5).build())
         .build();
+    initMetadataTable(cfg);
+
     HoodieTestDataGenerator.createCommitFile(basePath, "100", wrapperFs.getConf());
     HoodieTestDataGenerator.createCompactionRequestedFile(basePath, "101", wrapperFs.getConf());
     HoodieTestDataGenerator.createCompactionAuxiliaryMetadata(basePath,
@@ -436,6 +444,7 @@ public void testArchiveCommitTimeline() throws IOException {
                     .withParallelism(2, 2).forTable("test-trip-table")
                     .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 3).build())
                     .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
 
     HoodieTestDataGenerator.createCommitFile(basePath, "1", wrapperFs.getConf());
@@ -488,6 +497,7 @@ public void testArchiveCompletedClean() throws IOException {
                     .withParallelism(2, 2).forTable("test-trip-table")
                     .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 3).build())
                     .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
 
     createCleanMetadata("10", false);
@@ -513,6 +523,7 @@ public void testArchiveCompletedRollback() throws IOException {
                     .withParallelism(2, 2).forTable("test-trip-table")
                     .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 3).build())
                     .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
 
     createCommitAndRollbackFile("6", "10", false);
@@ -543,6 +554,7 @@ public void testArchiveCompletedShouldRetainMinInstantsIfInstantsGreaterThanMaxt
                     .withParallelism(2, 2).forTable("test-trip-table")
                     .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(minInstants, maxInstants).build())
                     .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
     for (int i = 0; i < maxInstants + 2; i++) {
       createCleanMetadata(i + "", false);
@@ -564,6 +576,7 @@ public void testArchiveCompletedShouldNotArchiveIfInstantsLessThanMaxtoKeep() th
                     .withParallelism(2, 2).forTable("test-trip-table")
                     .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(minInstants, maxInstants).build())
                     .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
     for (int i = 0; i < maxInstants; i++) {
       createCleanMetadata(i + "", false);
@@ -585,6 +598,7 @@ public void testArchiveCompletedRollbackAndClean() throws IOException {
                     .withParallelism(2, 2).forTable("test-trip-table")
                     .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(minInstantsToKeep, maxInstantsToKeep).build())
                     .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
 
     int startInstant = 1;
@@ -618,6 +632,7 @@ public void testArchiveInflightClean() throws IOException {
                     .withParallelism(2, 2).forTable("test-trip-table")
                     .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 3).build())
                     .build();
+    initMetadataTable(cfg);
     metaClient = HoodieTableMetaClient.reload(metaClient);
 
     createCleanMetadata("10", false);
@@ -677,4 +692,11 @@ private HoodieInstant createRollbackMetadata(String rollbackTime, String commitT
     }
     return new HoodieInstant(inflight, "rollback", rollbackTime);
   }
+
+  private void initMetadataTable(HoodieWriteConfig writeConfig) {
+    // Init the metadata table, so that commits/instants created externally can get synced later. Without
+    // instants getting synced, archival of those instants will not trigger.
+    SparkRDDWriteClient client = new SparkRDDWriteClient(context, writeConfig);
+    client.syncTableMetadata();
+  }
 }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowCreateHandle.java

@@ -19,6 +19,7 @@
 package org.apache.hudi.io.storage.row;
 
 import org.apache.hudi.client.HoodieInternalWriteStatus;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieWriteStat;
 import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
@@ -168,11 +169,15 @@ public void testGlobalFailure() throws Exception {
   @Test
   public void testInstantiationFailure() throws IOException {
     // init config and table
-    HoodieWriteConfig cfg = SparkDatasetTestUtils.getConfigBuilder(basePath).withPath("/dummypath/abc/").build();
+    HoodieWriteConfig cfg = SparkDatasetTestUtils.getConfigBuilder("/dummypath/abc/")
+        .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+            .enable(false)
+            .build())
+        .build();
     HoodieTable table = HoodieSparkTable.create(cfg, context, metaClient);
 
     try {
-      new HoodieRowCreateHandle(table, cfg, " def", UUID.randomUUID().toString(), "001", RANDOM.nextInt(100000), RANDOM.nextLong(), RANDOM.nextLong(), SparkDatasetTestUtils.STRUCT_TYPE);
+      new HoodieRowCreateHandle(table, cfg, "def", UUID.randomUUID().toString(), "001", RANDOM.nextInt(100000), RANDOM.nextLong(), RANDOM.nextLong(), SparkDatasetTestUtils.STRUCT_TYPE);
       fail("Should have thrown exception");
     } catch (HoodieInsertException ioe) {
       // expected

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

@@ -1079,8 +1079,15 @@ public void testLogFileCountsAfterCompaction(boolean populateMetaFields) throws
       HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
       HoodieSparkWriteableTestTable.of(table, HoodieTestDataGenerator.AVRO_SCHEMA_WITH_METADATA_FIELDS)
           .withLogAppends(updatedRecords);
-      // In writeRecordsToLogFiles, no commit files are getting added, so resetting file-system view state
-      ((SyncableFileSystemView) (table.getSliceView())).reset();
+
+      // Mark 2nd delta-instant as completed
+      metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(State.INFLIGHT,
+          HoodieTimeline.DELTA_COMMIT_ACTION, newCommitTime));
+      metaClient.getActiveTimeline().saveAsComplete(
+          new HoodieInstant(State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, newCommitTime), Option.empty());
+
+      metaClient = HoodieTableMetaClient.reload(metaClient);
+      table = HoodieSparkTable.create(config, context, metaClient);
 
       // Verify that all data file has one log file
       for (String partitionPath : dataGen.getPartitionPaths()) {
@@ -1091,12 +1098,6 @@ public void testLogFileCountsAfterCompaction(boolean populateMetaFields) throws
         }
       }
 
-      // Mark 2nd delta-instant as completed
-      metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(State.INFLIGHT,
-          HoodieTimeline.DELTA_COMMIT_ACTION, newCommitTime));
-      metaClient.getActiveTimeline().saveAsComplete(
-          new HoodieInstant(State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, newCommitTime), Option.empty());
-
       // Do a compaction
       String compactionInstantTime = writeClient.scheduleCompaction(Option.empty()).get().toString();
       JavaRDD<WriteStatus> result = (JavaRDD<WriteStatus>) writeClient.compact(compactionInstantTime);

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

@@ -40,7 +40,7 @@ public final class HoodieMetadataConfig extends HoodieConfig {
   // Enable the internal Metadata Table which saves file listings
   public static final ConfigProperty<Boolean> METADATA_ENABLE_PROP = ConfigProperty
       .key(METADATA_PREFIX + ".enable")
-      .defaultValue(false)
+      .defaultValue(true)
       .sinceVersion("0.7.0")
       .withDocumentation("Enable the internal metadata table which serves table metadata like level file listings");
 
@@ -51,8 +51,6 @@ public final class HoodieMetadataConfig extends HoodieConfig {
       .sinceVersion("0.7.0")
       .withDocumentation("Validate contents of metadata table on each access; e.g against the actual listings from lake storage");
 
-  public static final boolean DEFAULT_METADATA_ENABLE_FOR_READERS = false;
-
   // Enable metrics for internal Metadata Table
   public static final ConfigProperty<Boolean> METADATA_METRICS_ENABLE_PROP = ConfigProperty
       .key(METADATA_PREFIX + ".metrics.enable")

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieROTablePathFilter.java

@@ -18,6 +18,7 @@
 
 package org.apache.hudi.hadoop;
 
+import org.apache.hudi.common.config.HoodieMetadataConfig;
 import org.apache.hudi.common.config.SerializableConfiguration;
 import org.apache.hudi.common.engine.HoodieLocalEngineContext;
 import org.apache.hudi.common.fs.FSUtils;
@@ -29,7 +30,6 @@
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.TableNotFoundException;
 import org.apache.hudi.hadoop.utils.HoodieHiveUtils;
-import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils;
 
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
@@ -176,7 +176,7 @@ public boolean accept(Path path) {
           }
 
           fsView = FileSystemViewManager.createInMemoryFileSystemView(engineContext,
-              metaClient, HoodieInputFormatUtils.buildMetadataConfig(getConf()));
+              metaClient, HoodieMetadataConfig.newBuilder().build());
           String partition = FSUtils.getRelativePartitionPath(new Path(metaClient.getBasePath()), folder);
           List<HoodieBaseFile> latestFiles = fsView.getLatestBaseFiles(partition).collect(Collectors.toList());
           // populate the cache

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java

@@ -70,10 +70,6 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-import static org.apache.hudi.common.config.HoodieMetadataConfig.DEFAULT_METADATA_ENABLE_FOR_READERS;
-import static org.apache.hudi.common.config.HoodieMetadataConfig.METADATA_ENABLE_PROP;
-import static org.apache.hudi.common.config.HoodieMetadataConfig.METADATA_VALIDATE_PROP;
-
 public class HoodieInputFormatUtils {
 
   // These positions have to be deterministic across all tables
@@ -416,13 +412,6 @@ public static Map<HoodieTableMetaClient, List<Path>> groupSnapshotPathsByMetaCli
     return grouped;
   }
 
-  public static HoodieMetadataConfig buildMetadataConfig(Configuration conf) {
-    return HoodieMetadataConfig.newBuilder()
-        .enable(conf.getBoolean(METADATA_ENABLE_PROP.key(), DEFAULT_METADATA_ENABLE_FOR_READERS))
-        .validate(conf.getBoolean(METADATA_VALIDATE_PROP.key(), METADATA_VALIDATE_PROP.defaultValue()))
-        .build();
-  }
-
   public static List<FileStatus> filterFileStatusForSnapshotMode(JobConf job, Map<String, HoodieTableMetaClient> tableMetaClientMap,
                                                                  List<Path> snapshotPaths) throws IOException {
     HoodieLocalEngineContext engineContext = new HoodieLocalEngineContext(job);
@@ -443,7 +432,8 @@ public static List<FileStatus> filterFileStatusForSnapshotMode(JobConf job, Map<
         HoodieTimeline timeline = HoodieHiveUtils.getTableTimeline(metaClient.getTableConfig().getTableName(), job, metaClient);
 
         HoodieTableFileSystemView fsView = fsViewCache.computeIfAbsent(metaClient, tableMetaClient ->
-            FileSystemViewManager.createInMemoryFileSystemViewWithTimeline(engineContext, tableMetaClient, buildMetadataConfig(job), timeline));
+            FileSystemViewManager.createInMemoryFileSystemViewWithTimeline(engineContext, tableMetaClient,
+                HoodieMetadataConfig.newBuilder().build(), timeline));
         List<HoodieBaseFile> filteredBaseFiles = new ArrayList<>();
         for (Path p : entry.getValue()) {
           String relativePartitionPath = FSUtils.getRelativePartitionPath(new Path(metaClient.getBasePath()), p);

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java

@@ -18,6 +18,7 @@
 
 package org.apache.hudi.hadoop.utils;
 
+import org.apache.hudi.common.config.HoodieMetadataConfig;
 import org.apache.hudi.common.engine.HoodieLocalEngineContext;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.FileSlice;
@@ -105,7 +106,7 @@ public static InputSplit[] getRealtimeSplits(Configuration conf, Stream<FileSpli
         if (!fsCache.containsKey(metaClient)) {
           HoodieLocalEngineContext engineContext = new HoodieLocalEngineContext(conf);
           HoodieTableFileSystemView fsView = FileSystemViewManager.createInMemoryFileSystemView(engineContext,
-              metaClient, HoodieInputFormatUtils.buildMetadataConfig(conf));
+              metaClient, HoodieMetadataConfig.newBuilder().build());
           fsCache.put(metaClient, fsView);
         }
         HoodieTableFileSystemView fsView = fsCache.get(metaClient);

hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/reader/DFSHoodieDatasetInputReader.java

@@ -88,7 +88,7 @@ protected List<String> getPartitions(Option<Integer> partitionsLimit) throws IOE
     // calls in metrics as they are not part of normal HUDI operation.
     HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
     List<String> partitionPaths = FSUtils.getAllPartitionPaths(engineContext, metaClient.getBasePath(),
-        HoodieMetadataConfig.DEFAULT_METADATA_ENABLE_FOR_READERS, HoodieMetadataConfig.METADATA_VALIDATE_PROP.defaultValue(), false);
+        HoodieMetadataConfig.METADATA_ENABLE_PROP.defaultValue(), HoodieMetadataConfig.METADATA_VALIDATE_PROP.defaultValue(), false);
     // Sort partition so we can pick last N partitions by default
     Collections.sort(partitionPaths);
     if (!partitionPaths.isEmpty()) {

hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieFileIndex.scala

@@ -124,7 +124,7 @@ case class HoodieFileIndex(
     // would be able to run SET hoodie.metadata.enable=true in the spark sql session to enable metadata listing.
     properties.put(HoodieMetadataConfig.METADATA_ENABLE_PROP,
       sqlConf.getConfString(HoodieMetadataConfig.METADATA_ENABLE_PROP.key(),
-        HoodieMetadataConfig.DEFAULT_METADATA_ENABLE_FOR_READERS.toString))
+        HoodieMetadataConfig.METADATA_ENABLE_PROP.defaultValue().toString))
     properties.put(HoodieMetadataConfig.METADATA_VALIDATE_PROP,
       sqlConf.getConfString(HoodieMetadataConfig.METADATA_VALIDATE_PROP.key(),
         HoodieMetadataConfig.METADATA_VALIDATE_PROP.defaultValue().toString))