[HUDI-37][WIP] Persist the HoodieIndex type in the hoodie.properties file

hudi-cli/src/main/java/org/apache/hudi/cli/commands/TableCommand.java

@@ -108,7 +108,7 @@ public String createTable(
 
     final HoodieTableType tableType = HoodieTableType.valueOf(tableTypeStr);
     HoodieTableMetaClient.initTableType(HoodieCLI.conf, path, tableType, name, archiveFolder,
-        payloadClass, layoutVersion);
+        payloadClass, layoutVersion, null);
 
     // Now connect to ensure loading works
     return connect(path, layoutVersion, false, 0, 0, 0);

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

@@ -311,7 +311,7 @@ private void testUpsertsInternal(HoodieWriteConfig config,
         VERSION_0).build();
     HoodieTableMetaClient.initTableType(metaClient.getHadoopConf(), metaClient.getBasePath(), metaClient.getTableType(),
         metaClient.getTableConfig().getTableName(), metaClient.getArchivePath(),
-        metaClient.getTableConfig().getPayloadClass(), VERSION_0);
+        metaClient.getTableConfig().getPayloadClass(), VERSION_0, hoodieWriteConfig.getIndexType().toString());
     SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig, false);
 
     // Write 1 (only inserts)
@@ -491,7 +491,7 @@ private void testUpsertsUpdatePartitionPath(IndexType indexType, HoodieWriteConf
             .build()).withTimelineLayoutVersion(VERSION_0).build();
     HoodieTableMetaClient.initTableType(metaClient.getHadoopConf(), metaClient.getBasePath(),
         metaClient.getTableType(), metaClient.getTableConfig().getTableName(), metaClient.getArchivePath(),
-        metaClient.getTableConfig().getPayloadClass(), VERSION_0);
+        metaClient.getTableConfig().getPayloadClass(), VERSION_0, config.getIndexType().toString());
     SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig, false);
 
     // Write 1

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java

@@ -76,7 +76,7 @@ protected HoodieWriteConfig getHoodieWriteConfig(String basePath) {
   public void readLocalWriteHDFS() throws Exception {
     // Initialize table and filesystem
     HoodieTableMetaClient.initTableType(hadoopConf, dfsBasePath, HoodieTableType.valueOf(tableType),
-        tableName, HoodieAvroPayload.class.getName());
+        tableName, HoodieAvroPayload.class.getName(), null);
 
     // Create write client to write some records in
     HoodieWriteConfig cfg = getHoodieWriteConfig(dfsBasePath);
@@ -101,7 +101,7 @@ public void readLocalWriteHDFS() throws Exception {
 
       // Write to local
       HoodieTableMetaClient.initTableType(hadoopConf, tablePath, HoodieTableType.valueOf(tableType),
-          tableName, HoodieAvroPayload.class.getName());
+          tableName, HoodieAvroPayload.class.getName(), null);
 
       String writeCommitTime = localWriteClient.startCommit();
       LOG.info("Starting write commit " + writeCommitTime);

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestTableSchemaEvolution.java

@@ -129,7 +129,7 @@ public void testMORTable() throws Exception {
     // Create the table
     HoodieTableMetaClient.initTableType(metaClient.getHadoopConf(), metaClient.getBasePath(),
         HoodieTableType.MERGE_ON_READ, metaClient.getTableConfig().getTableName(),
-        metaClient.getArchivePath(), metaClient.getTableConfig().getPayloadClass(), VERSION_1);
+        metaClient.getArchivePath(), metaClient.getTableConfig().getPayloadClass(), VERSION_1, null);
 
     HoodieWriteConfig hoodieWriteConfig = getWriteConfig(TRIP_EXAMPLE_SCHEMA);
     SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig, false);
@@ -274,7 +274,7 @@ public void testCopyOnWriteTable() throws Exception {
     // Create the table
     HoodieTableMetaClient.initTableType(metaClient.getHadoopConf(), metaClient.getBasePath(),
         HoodieTableType.COPY_ON_WRITE, metaClient.getTableConfig().getTableName(),
-        metaClient.getArchivePath(), metaClient.getTableConfig().getPayloadClass(), VERSION_1);
+        metaClient.getArchivePath(), metaClient.getTableConfig().getPayloadClass(), VERSION_1, null);
 
     HoodieWriteConfig hoodieWriteConfig = getWriteConfig(TRIP_EXAMPLE_SCHEMA);
     SparkRDDWriteClient client = getHoodieWriteClient(hoodieWriteConfig, false);

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

@@ -329,38 +329,38 @@ public synchronized HoodieArchivedTimeline getArchivedTimeline() {
   public static HoodieTableMetaClient initTableTypeWithBootstrap(Configuration hadoopConf, String basePath, HoodieTableType tableType,
                                                                  String tableName, String archiveLogFolder, String payloadClassName,
                                                                  String baseFileFormat, String bootstrapIndexClass,
-                                                                 String bootstrapBasePath) throws IOException {
+                                                                 String bootstrapBasePath, String indexType) throws IOException {
     return initTableType(hadoopConf, basePath, tableType, tableName,
-        archiveLogFolder, payloadClassName, null, baseFileFormat, bootstrapIndexClass, bootstrapBasePath);
+        archiveLogFolder, payloadClassName, null, baseFileFormat, bootstrapIndexClass, bootstrapBasePath, indexType);
   }
 
   public static HoodieTableMetaClient initTableType(Configuration hadoopConf, String basePath, HoodieTableType tableType,
                                                     String tableName, String archiveLogFolder, String payloadClassName,
-                                                    String baseFileFormat) throws IOException {
+                                                    String baseFileFormat, String indexType) throws IOException {
     return initTableType(hadoopConf, basePath, tableType, tableName,
-        archiveLogFolder, payloadClassName, null, baseFileFormat, null, null);
+        archiveLogFolder, payloadClassName, null, baseFileFormat, null, null, indexType);
   }
 
   /**
    * Used primarily by tests, examples.
    */
   public static HoodieTableMetaClient initTableType(Configuration hadoopConf, String basePath, HoodieTableType tableType,
-                                                    String tableName, String payloadClassName) throws IOException {
+                                                    String tableName, String payloadClassName, String indexType) throws IOException {
     return initTableType(hadoopConf, basePath, tableType, tableName, null, payloadClassName,
-        null, null, null, null);
+        null, null, null, null, indexType);
   }
 
   public static HoodieTableMetaClient initTableType(Configuration hadoopConf, String basePath, HoodieTableType tableType,
                                                     String tableName, String archiveLogFolder, String payloadClassName,
-                                                    Integer timelineLayoutVersion) throws IOException {
+                                                    Integer timelineLayoutVersion, String indexType) throws IOException {
     return initTableType(hadoopConf, basePath, tableType, tableName, archiveLogFolder, payloadClassName,
-        timelineLayoutVersion, null, null, null);
+        timelineLayoutVersion, null, null, null, indexType);
   }
 
   private static HoodieTableMetaClient initTableType(Configuration hadoopConf, String basePath, HoodieTableType tableType,
                                                      String tableName, String archiveLogFolder, String payloadClassName,
                                                      Integer timelineLayoutVersion, String baseFileFormat,
-                                                     String bootstrapIndexClass, String bootstrapBasePath) throws IOException {
+                                                     String bootstrapIndexClass, String bootstrapBasePath, String indexType) throws IOException {
     Properties properties = new Properties();
     properties.setProperty(HoodieTableConfig.HOODIE_TABLE_NAME_PROP_NAME, tableName);
     properties.setProperty(HoodieTableConfig.HOODIE_TABLE_TYPE_PROP_NAME, tableType.name());
@@ -389,6 +389,9 @@ private static HoodieTableMetaClient initTableType(Configuration hadoopConf, Str
       properties.put(HoodieTableConfig.HOODIE_BOOTSTRAP_BASE_PATH, bootstrapBasePath);
     }
 
+    if (null != indexType) {
+      properties.put("hoodie.index.type", indexType);
+    }
     return HoodieTableMetaClient.initTableAndGetMetaClient(hadoopConf, basePath, properties);
   }
 

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

@@ -111,7 +111,8 @@ public void testLoadingInstantsFromFiles() throws IOException {
     // 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);
+        metaClient.getArchivePath(), metaClient.getTableConfig().getPayloadClass(), VERSION_0,
+        metaClient.getTableConfig().getProperties().getProperty("hoodie.index.type"));
     HoodieInstant instant6 = new HoodieInstant(State.REQUESTED, HoodieTimeline.COMPACTION_ACTION, "9");
     byte[] dummy = new byte[5];
     HoodieActiveTimeline oldTimeline = new HoodieActiveTimeline(new HoodieTableMetaClient(metaClient.getHadoopConf(),

hudi-examples/src/main/java/org/apache/hudi/examples/spark/HoodieWriteClientExample.java

@@ -84,17 +84,17 @@ public static void main(String[] args) throws Exception {
       // initialize the table, if not done already
       Path path = new Path(tablePath);
       FileSystem fs = FSUtils.getFs(tablePath, jsc.hadoopConfiguration());
-      if (!fs.exists(path)) {
-        HoodieTableMetaClient.initTableType(jsc.hadoopConfiguration(), tablePath, HoodieTableType.valueOf(tableType),
-                tableName, HoodieAvroPayload.class.getName());
-      }
 
       // Create the write client to write some records in
       HoodieWriteConfig cfg = HoodieWriteConfig.newBuilder().withPath(tablePath)
               .withSchema(HoodieExampleDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2)
               .withDeleteParallelism(2).forTable(tableName)
               .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BLOOM).build())
               .withCompactionConfig(HoodieCompactionConfig.newBuilder().archiveCommitsWith(20, 30).build()).build();
+      if (!fs.exists(path)) {
+        HoodieTableMetaClient.initTableType(jsc.hadoopConfiguration(), tablePath, HoodieTableType.valueOf(tableType),
+            tableName, HoodieAvroPayload.class.getName(), cfg.getIndexType().name());
+      }
       SparkRDDWriteClient<HoodieAvroPayload> client = new SparkRDDWriteClient<>(new HoodieSparkEngineContext(jsc), cfg);
 
       // inserts

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

@@ -94,7 +94,7 @@ public HoodieTestSuiteJob(HoodieTestSuiteConfig cfg, JavaSparkContext jsc) throw
 
     if (!fs.exists(new Path(cfg.targetBasePath))) {
       HoodieTableMetaClient.initTableType(jsc.hadoopConfiguration(), cfg.targetBasePath,
-          HoodieTableType.valueOf(cfg.tableType), cfg.targetTableName, "archived");
+          HoodieTableType.valueOf(cfg.tableType), cfg.targetTableName, "archived", null);
     }
   }
 

hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala

@@ -35,9 +35,10 @@ import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
 import org.apache.hudi.common.table.timeline.HoodieActiveTimeline
 import org.apache.hudi.common.util.ReflectionUtils
 import org.apache.hudi.config.HoodieBootstrapConfig.{BOOTSTRAP_BASE_PATH_PROP, BOOTSTRAP_INDEX_CLASS_PROP, DEFAULT_BOOTSTRAP_INDEX_CLASS}
-import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.config.{HoodieIndexConfig, HoodieWriteConfig}
 import org.apache.hudi.exception.HoodieException
 import org.apache.hudi.hive.{HiveSyncConfig, HiveSyncTool}
+import org.apache.hudi.index.HoodieIndex
 import org.apache.hudi.internal.HoodieDataSourceInternalWriter
 import org.apache.hudi.sync.common.AbstractSyncTool
 import org.apache.log4j.LogManager
@@ -110,9 +111,11 @@ private[hudi] object HoodieSparkSqlWriter {
       if (!tableExists) {
         val archiveLogFolder = parameters.getOrElse(
           HoodieTableConfig.HOODIE_ARCHIVELOG_FOLDER_PROP_NAME, "archived")
+        val indexType = parameters.getOrElse(
+          HoodieIndexConfig.INDEX_TYPE_PROP, HoodieIndexConfig.DEFAULT_INDEX_TYPE)
         val tableMetaClient = HoodieTableMetaClient.initTableType(sparkContext.hadoopConfiguration, path.get,
           tableType, tblName, archiveLogFolder, parameters(PAYLOAD_CLASS_OPT_KEY),
-          null.asInstanceOf[String])
+          null.asInstanceOf[String], indexType)
         tableConfig = tableMetaClient.getTableConfig
       }
 
@@ -216,7 +219,6 @@ private[hudi] object HoodieSparkSqlWriter {
                 parameters: Map[String, String],
                 df: DataFrame,
                 hoodieTableConfigOpt: Option[HoodieTableConfig] = Option.empty): Boolean = {
-
     val sparkContext = sqlContext.sparkContext
     val path = parameters.getOrElse("path", throw new HoodieException("'path' must be set."))
     val tableName = parameters.getOrElse(HoodieWriteConfig.TABLE_NAME,
@@ -251,9 +253,11 @@ private[hudi] object HoodieSparkSqlWriter {
     if (!tableExists) {
       val archiveLogFolder = parameters.getOrElse(
         HoodieTableConfig.HOODIE_ARCHIVELOG_FOLDER_PROP_NAME, "archived")
+      val indexType = parameters.getOrElse(
+        HoodieIndexConfig.INDEX_TYPE_PROP, HoodieIndexConfig.DEFAULT_INDEX_TYPE)
       HoodieTableMetaClient.initTableTypeWithBootstrap(sparkContext.hadoopConfiguration, path,
         HoodieTableType.valueOf(tableType), tableName, archiveLogFolder, parameters(PAYLOAD_CLASS_OPT_KEY),
-        null, bootstrapIndexClass, bootstrapBasePath)
+        null, bootstrapIndexClass, bootstrapBasePath, indexType)
     }
 
     val jsc = new JavaSparkContext(sqlContext.sparkContext)

hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java

@@ -127,7 +127,7 @@ public static void setUp() throws IOException, InterruptedException {
   public static void clear() throws IOException {
     fileSystem.delete(new Path(hiveSyncConfig.basePath), true);
     HoodieTableMetaClient.initTableType(configuration, hiveSyncConfig.basePath, HoodieTableType.COPY_ON_WRITE,
-        hiveSyncConfig.tableName, HoodieAvroPayload.class.getName());
+        hiveSyncConfig.tableName, HoodieAvroPayload.class.getName(), null);
 
     HoodieHiveClient client = new HoodieHiveClient(hiveSyncConfig, hiveServer.getHiveConf(), fileSystem);
     for (String tableName : createdTablesSet) {
@@ -162,7 +162,7 @@ public static void createCOWTable(String instantTime, int numberOfPartitions, bo
     Path path = new Path(hiveSyncConfig.basePath);
     FileIOUtils.deleteDirectory(new File(hiveSyncConfig.basePath));
     HoodieTableMetaClient.initTableType(configuration, hiveSyncConfig.basePath, HoodieTableType.COPY_ON_WRITE,
-        hiveSyncConfig.tableName, HoodieAvroPayload.class.getName());
+        hiveSyncConfig.tableName, HoodieAvroPayload.class.getName(), null);
     boolean result = fileSystem.mkdirs(path);
     checkResult(result);
     DateTime dateTime = DateTime.now();
@@ -178,7 +178,7 @@ public static void createMORTable(String commitTime, String deltaCommitTime, int
     Path path = new Path(hiveSyncConfig.basePath);
     FileIOUtils.deleteDirectory(new File(hiveSyncConfig.basePath));
     HoodieTableMetaClient.initTableType(configuration, hiveSyncConfig.basePath, HoodieTableType.MERGE_ON_READ,
-        hiveSyncConfig.tableName, HoodieAvroPayload.class.getName());
+        hiveSyncConfig.tableName, HoodieAvroPayload.class.getName(), null);
 
     boolean result = fileSystem.mkdirs(path);
     checkResult(result);

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/BootstrapExecutor.java

@@ -173,7 +173,7 @@ private void initializeTable() throws IOException {
 
     HoodieTableMetaClient.initTableTypeWithBootstrap(new Configuration(jssc.hadoopConfiguration()),
         cfg.targetBasePath, HoodieTableType.valueOf(cfg.tableType), cfg.targetTableName, "archived", cfg.payloadClassName,
-        cfg.baseFileFormat, cfg.bootstrapIndexClass, bootstrapBasePath);
+        cfg.baseFileFormat, cfg.bootstrapIndexClass, bootstrapBasePath, bootstrapConfig.getIndexType().name());
   }
 
   public HoodieWriteConfig getBootstrapConfig() {

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java

@@ -221,7 +221,8 @@ private void refreshTimeline() throws IOException {
     } else {
       this.commitTimelineOpt = Option.empty();
       HoodieTableMetaClient.initTableType(new Configuration(jssc.hadoopConfiguration()), cfg.targetBasePath,
-          HoodieTableType.valueOf(cfg.tableType), cfg.targetTableName, "archived", cfg.payloadClassName, cfg.baseFileFormat);
+          HoodieTableType.valueOf(cfg.tableType), cfg.targetTableName, "archived", cfg.payloadClassName, cfg.baseFileFormat,
+          getHoodieClientConfig(this.schemaProvider).getIndexType().name());
     }
   }
 
@@ -289,7 +290,8 @@ public Pair<SchemaProvider, Pair<String, JavaRDD<HoodieRecord>>> readFromSource(
       }
     } else {
       HoodieTableMetaClient.initTableType(new Configuration(jssc.hadoopConfiguration()), cfg.targetBasePath,
-          HoodieTableType.valueOf(cfg.tableType), cfg.targetTableName, "archived", cfg.payloadClassName, cfg.baseFileFormat);
+          HoodieTableType.valueOf(cfg.tableType), cfg.targetTableName, "archived", cfg.payloadClassName, cfg.baseFileFormat,
+          getHoodieClientConfig(this.schemaProvider).getIndexType().name());
     }
 
     if (!resumeCheckpointStr.isPresent() && cfg.checkpoint != null) {

hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieSnapshotExporter.java

@@ -81,7 +81,7 @@ public void init() throws Exception {
     dfs().mkdirs(new Path(sourcePath));
     HoodieTableMetaClient
         .initTableType(jsc().hadoopConfiguration(), sourcePath, HoodieTableType.COPY_ON_WRITE, TABLE_NAME,
-            HoodieAvroPayload.class.getName());
+            HoodieAvroPayload.class.getName(), null);
 
     // Prepare data as source Hudi dataset
     HoodieWriteConfig cfg = getHoodieWriteConfig(sourcePath);

hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/UtilitiesTestBase.java

@@ -180,7 +180,7 @@ private static void clearHiveDb() throws IOException {
     HiveSyncConfig hiveSyncConfig = getHiveSyncConfig("/dummy", "dummy");
     hiveConf.addResource(hiveServer.getHiveConf());
     HoodieTableMetaClient.initTableType(dfs.getConf(), hiveSyncConfig.basePath, HoodieTableType.COPY_ON_WRITE,
-        hiveSyncConfig.tableName, null);
+        hiveSyncConfig.tableName, null, null);
     HoodieHiveClient client = new HoodieHiveClient(hiveSyncConfig, hiveConf, dfs);
     client.updateHiveSQL("drop database if exists " + hiveSyncConfig.databaseName);
     client.updateHiveSQL("create database " + hiveSyncConfig.databaseName);