[MINOR] Add tests for autoUpgrade config during table creation

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

@@ -60,6 +60,7 @@
 import org.apache.hudi.common.util.collection.Triple;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
 import org.apache.hudi.exception.TableNotFoundException;
 import org.apache.hudi.keygen.constant.KeyGeneratorType;
 import org.apache.hudi.metadata.HoodieTableMetadata;
@@ -1473,10 +1474,7 @@ public Properties build() {
       tableConfig.setValue(HoodieTableConfig.NAME, tableName);
       tableConfig.setValue(HoodieTableConfig.TYPE, tableType.name());
 
-      if (null == tableVersion) {
-        tableVersion = HoodieTableVersion.current();
-      }
-
+      tableVersion = getTableVersionProperly();
       tableConfig.setTableVersion(tableVersion);
       tableConfig.setInitialVersion(tableVersion);
 
@@ -1591,6 +1589,20 @@ public Properties build() {
       return tableConfig.getProps();
     }
 
+    HoodieTableVersion getTableVersionProperly() {
+      if (tableVersion == null) {
+        return HoodieTableVersion.current();
+      } else if (tableVersion.greaterThanOrEquals(HoodieTableVersion.SIX)) {
+        return tableVersion;
+      } else {
+        String errorMessage = String.format(
+            "Creating a table in version \"%d\" is not allowed in current Hudi binary. "
+                + "Please set \"hoodie.write.table.version\" config with a value >= 6",
+            tableVersion.versionCode());
+        throw new HoodieNotSupportedException(errorMessage);
+      }
+    }
+
     public HoodieTableMetaClient initTable(StorageConfiguration<?> configuration, String basePath) throws IOException {
       return initTable(configuration, new StoragePath(basePath));
     }

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java

@@ -29,6 +29,7 @@
 import org.apache.hudi.common.testutils.HoodieTestUtils;
 import org.apache.hudi.common.util.FileIOUtils;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieNotSupportedException;
 import org.apache.hudi.metadata.HoodieIndexVersion;
 import org.apache.hudi.metadata.MetadataPartitionType;
 import org.apache.hudi.storage.StoragePath;
@@ -283,4 +284,71 @@ void testDeleteDefinition() throws IOException {
         new String(FileIOUtils.readDataFromPath(metaClient.getStorage(), new StoragePath(metaClient.getIndexDefinitionPath())).get()));
     assertTrue(indexMetadata.getIndexDefinitions().isEmpty());
   }
+
+  @Test
+  void testGetTableVersionProperlyWithNullTableVersion() {
+    HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
+        .setTableName("test_table")
+        .setTableType(HoodieTableType.COPY_ON_WRITE);
+
+    // tableVersion is null by default
+    HoodieTableVersion result = tableBuilder.getTableVersionProperly();
+    assertEquals(HoodieTableVersion.current(), result);
+  }
+
+  @Test
+  void testGetTableVersionProperlyWithCurrentVersion() {
+    HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
+        .setTableName("test_table")
+        .setTableType(HoodieTableType.COPY_ON_WRITE)
+        .setTableVersion(HoodieTableVersion.current());
+    HoodieTableVersion result = tableBuilder.getTableVersionProperly();
+    assertEquals(HoodieTableVersion.current(), result);
+  }
+
+  @Test
+  void testGetTableVersionProperlyWithValidVersions() {
+    HoodieTableVersion[] lowerVersions = {
+        HoodieTableVersion.SIX,
+        HoodieTableVersion.SEVEN,
+        HoodieTableVersion.EIGHT,
+        HoodieTableVersion.NINE
+    };
+
+    for (HoodieTableVersion version : lowerVersions) {
+      HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
+          .setTableName("test_table")
+          .setTableType(HoodieTableType.COPY_ON_WRITE)
+          .setTableVersion(version);
+
+      HoodieTableVersion result = tableBuilder.getTableVersionProperly();
+      assertEquals(version, result);
+    }
+  }
+
+  @Test
+  void testGetTableVersionProperlyWithInvalidValues() {
+    // Test with different lower versions and autoUpgrade=true
+    HoodieTableVersion[] lowerVersions = {
+        HoodieTableVersion.ZERO,
+        HoodieTableVersion.ONE,
+        HoodieTableVersion.TWO,
+        HoodieTableVersion.THREE,
+        HoodieTableVersion.FOUR,
+        HoodieTableVersion.FIVE
+    };
+
+    for (HoodieTableVersion version : lowerVersions) {
+      HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
+          .setTableName("test_table")
+          .setTableType(HoodieTableType.COPY_ON_WRITE)
+          .setTableVersion(version);
+
+      HoodieNotSupportedException exception = assertThrows(
+          HoodieNotSupportedException.class,
+          () -> tableBuilder.getTableVersionProperly());
+      assertTrue(exception.getMessage().contains(
+          String.format("Creating a table in version \"%d\"", version.versionCode())));
+    }
+  }
 }

hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java

@@ -27,35 +27,43 @@
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.table.timeline.InstantFileNameGenerator;
 import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
 import org.apache.hudi.common.testutils.HoodieTestUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieNotSupportedException;
 import org.apache.hudi.exception.HoodieUpgradeDowngradeException;
 import org.apache.hudi.metadata.HoodieTableMetadata;
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
 
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -347,6 +355,60 @@ public void testMetadataTableUpgradeDowngradeFailure(HoodieTableVersion fromVers
         "Exception message should contain metadata table failure message");
   }
 
+  @ParameterizedTest
+  @CsvSource({"1","2","3","4","5","6","7","8","9","null"})
+  void testTableVersionDuringTableCreation(String targetTableVersion) {
+    SparkSession spark = spark();
+    Map<String, String> writeOptions = new HashMap<>();
+    if (!targetTableVersion.equals("null")) {
+      writeOptions.put(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), targetTableVersion);
+    }
+    writeOptions.put(HoodieWriteConfig.TBL_NAME.key(), "testTableCreation");
+    HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
+    List<String> records = recordsToStrings(dataGen.generateInserts("001", 5));
+    Dataset<Row> inputDF = spark.read().json(jsc().parallelize(records, 2));
+
+    // Create table, and validate.
+    Set<String> failSet = new HashSet<>(Arrays.asList("1", "2", "3", "4", "5"));
+    if (failSet.contains(targetTableVersion)) {
+      // This fails at HoodieTableMetaClient.
+      assertThrows(HoodieNotSupportedException.class, () -> {
+        inputDF.write().format("org.apache.hudi")
+            .partitionBy("partition")
+            .options(writeOptions)
+            .mode(SaveMode.Append)
+            .save(basePath());
+      });
+    } else if (targetTableVersion.equals("7")) {
+      // Fails at write table version check.
+      assertThrows(IllegalArgumentException.class, () -> {
+        inputDF.write().format("org.apache.hudi")
+            .partitionBy("partition")
+            .options(writeOptions)
+            .mode(SaveMode.Append)
+            .save(basePath());
+      });
+    } else {
+      inputDF.write().format("org.apache.hudi")
+          .partitionBy("partition")
+          .options(writeOptions)
+          .mode(SaveMode.Append)
+          .save(basePath());
+      metaClient = HoodieTableMetaClient.builder().setConf(storageConf()).setBasePath(basePath()).build();
+      // If no write version is specified, use current.
+      if (!targetTableVersion.equals("null")) {
+        assertEquals(
+            Integer.valueOf(targetTableVersion),
+            metaClient.getTableConfig().getTableVersion().versionCode());
+      } else {
+        // Otherwise, the table version is the target table version.
+        assertEquals(
+            HoodieTableVersion.current(),
+            metaClient.getTableConfig().getTableVersion());
+      }
+    }
+  }
+
   /**
    * Load a fixture table from resources and copy it to a temporary location for testing.
    */