[HUDI-559] : Make the timeline layout version default to be null version

hudi-client/src/main/java/org/apache/hudi/client/utils/ClientUtils.java

@@ -37,6 +37,7 @@ public class ClientUtils {
   public static HoodieTableMetaClient createMetaClient(JavaSparkContext jsc, HoodieWriteConfig config,
       boolean loadActiveTimelineOnLoad) {
     return new HoodieTableMetaClient(jsc.hadoopConfiguration(), config.getBasePath(), loadActiveTimelineOnLoad,
-        config.getConsistencyGuardConfig(), Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion())));
+        config.getConsistencyGuardConfig(),
+        Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion())));
   }
 }

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

@@ -757,6 +757,7 @@ public class HoodieWriteConfig extends DefaultHoodieConfig {
       // Ensure Layout Version is good
       new TimelineLayoutVersion(Integer.parseInt(layoutVersion));
 
+
       // Build WriteConfig at the end
       HoodieWriteConfig config = new HoodieWriteConfig(props);
       Preconditions.checkArgument(config.getBasePath() != null);

hudi-client/src/test/java/org/apache/hudi/TestHoodieClientBase.java

@@ -26,6 +26,7 @@ import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodiePartitionMetadata;
 import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.TimelineLayoutVersion;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.HoodieTimeline;
 import org.apache.hudi.common.table.SyncableFileSystemView;
@@ -145,6 +146,7 @@ public class TestHoodieClientBase extends HoodieClientTestHarness {
   HoodieWriteConfig.Builder getConfigBuilder(String schemaStr, IndexType indexType) {
     return HoodieWriteConfig.newBuilder().withPath(basePath).withSchema(schemaStr)
         .withParallelism(2, 2).withBulkInsertParallelism(2).withFinalizeWriteParallelism(2)
+        .withTimelineLayoutVersion(TimelineLayoutVersion.CURR_VERSION)
         .withWriteStatusClass(MetadataMergeWriteStatus.class)
         .withConsistencyGuardConfig(ConsistencyGuardConfig.newBuilder().withConsistencyCheckEnabled(true).build())
         .withCompactionConfig(HoodieCompactionConfig.newBuilder().compactionSmallFileSize(1024 * 1024).build())

hudi-client/src/test/java/org/apache/hudi/TestHoodieClientOnCopyOnWriteStorage.java

@@ -70,6 +70,7 @@ import java.util.stream.Collectors;
 
 import static org.apache.hudi.common.HoodieTestDataGenerator.NULL_SCHEMA;
 import static org.apache.hudi.common.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static org.apache.hudi.common.model.TimelineLayoutVersion.VERSION_0;
 import static org.apache.hudi.common.util.ParquetUtils.readRowKeysFromParquet;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -271,7 +272,10 @@ public class TestHoodieClientOnCopyOnWriteStorage extends TestHoodieClientBase {
       throws Exception {
     // Force using older timeline layout
     HoodieWriteConfig hoodieWriteConfig = getConfigBuilder().withProps(config.getProps()).withTimelineLayoutVersion(
-        TimelineLayoutVersion.VERSION_0).build();
+        VERSION_0).build();
+    HoodieTableMetaClient.initTableType(metaClient.getHadoopConf(), metaClient.getBasePath(), metaClient.getTableType(),
+        metaClient.getTableConfig().getTableName(), metaClient.getArchivePath(),
+        metaClient.getTableConfig().getPayloadClass(), VERSION_0);
     HoodieWriteClient client = getHoodieWriteClient(hoodieWriteConfig, false);
 
     // Write 1 (only inserts)

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java

@@ -22,6 +22,7 @@ import org.apache.hudi.common.model.HoodieFileFormat;
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
 import org.apache.hudi.common.model.TimelineLayoutVersion;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.exception.HoodieIOException;
 
 import org.apache.hadoop.fs.FSDataInputStream;
@@ -144,10 +145,10 @@ public class HoodieTableConfig implements Serializable {
     return DEFAULT_TABLE_TYPE;
   }
 
-  public TimelineLayoutVersion getTimelineLayoutVersion() {
-    return new TimelineLayoutVersion(Integer.valueOf(props.getProperty(HOODIE_TIMELINE_LAYOUT_VERSION,
-        String.valueOf(DEFAULT_TIMELINE_LAYOUT_VERSION))));
-
+  public Option<TimelineLayoutVersion> getTimelineLayoutVersion() {
+    return props.containsKey(HOODIE_TIMELINE_LAYOUT_VERSION)
+        ? Option.of(new TimelineLayoutVersion(Integer.valueOf(props.getProperty(HOODIE_TIMELINE_LAYOUT_VERSION))))
+        : Option.empty();
   }
 
   /**

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java

@@ -117,7 +117,14 @@ public class HoodieTableMetaClient implements Serializable {
     TableNotFoundException.checkTableValidity(fs, basePathDir, metaPathDir);
     this.tableConfig = new HoodieTableConfig(fs, metaPath, payloadClassName);
     this.tableType = tableConfig.getTableType();
-    this.timelineLayoutVersion = layoutVersion.orElse(tableConfig.getTimelineLayoutVersion());
+    Option<TimelineLayoutVersion> tableConfigVersion = tableConfig.getTimelineLayoutVersion();
+    if (layoutVersion.isPresent() && tableConfigVersion.isPresent()) {
+      // Ensure layout version passed in config is not lower than the one seen in hoodie.properties
+      Preconditions.checkArgument(layoutVersion.get().compareTo(tableConfigVersion.get()) >= 0,
+          "Layout Version defined in hoodie properties has higher version (" + tableConfigVersion.get()
+              + ") than the one passed in config (" + layoutVersion.get() + ")");
+    }
+    this.timelineLayoutVersion = layoutVersion.orElseGet(() -> tableConfig.getTimelineLayoutVersion().get());
     this.loadActiveTimelineOnLoad = loadActiveTimelineOnLoad;
     LOG.info("Finished Loading Table of type " + tableType + "(version=" + timelineLayoutVersion + ") from " + basePath);
     if (loadActiveTimelineOnLoad) {

hudi-common/src/test/java/org/apache/hudi/common/table/string/TestHoodieActiveTimeline.java

@@ -37,6 +37,7 @@ import org.junit.rules.ExpectedException;
 import java.io.IOException;
 import java.util.stream.Stream;
 
+import static org.apache.hudi.common.model.TimelineLayoutVersion.VERSION_0;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -102,11 +103,14 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
         timeline.getCommitTimeline().filterPendingExcludingCompaction().getInstants());
 
     // Backwards compatibility testing for reading compaction plans
+    metaClient = HoodieTableMetaClient.initTableType(metaClient.getHadoopConf(),
+        metaClient.getBasePath(), metaClient.getTableType(), metaClient.getTableConfig().getTableName(),
+        metaClient.getArchivePath(), metaClient.getTableConfig().getPayloadClass(), VERSION_0);
     HoodieInstant instant6 = new HoodieInstant(State.REQUESTED, HoodieTimeline.COMPACTION_ACTION, "9");
     byte[] dummy = new byte[5];
     HoodieActiveTimeline oldTimeline = new HoodieActiveTimeline(new HoodieTableMetaClient(metaClient.getHadoopConf(),
         metaClient.getBasePath(), true, metaClient.getConsistencyGuardConfig(),
-        Option.of(new TimelineLayoutVersion(TimelineLayoutVersion.VERSION_0))));
+        Option.of(new TimelineLayoutVersion(VERSION_0))));
     // Old Timeline writes both to aux and timeline folder
     oldTimeline.saveToCompactionRequested(instant6, Option.of(dummy));
     // Now use latest timeline version