Fix hudi-spark2 module tests

Author: codope

hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/internal/HoodieBulkInsertInternalWriterTestBase.java

@@ -29,7 +29,6 @@
 import org.apache.hudi.keygen.SimpleKeyGenerator;
 import org.apache.hudi.testutils.HoodieClientTestHarness;
 import org.apache.hudi.testutils.SparkDatasetTestUtils;
-
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.junit.jupiter.api.AfterEach;
@@ -62,6 +61,7 @@ public void setUp() throws Exception {
     initFileSystem();
     initTestDataGenerator();
     initMetaClient();
+    initTimelineService();
   }
 
   @AfterEach
@@ -87,11 +87,11 @@ protected HoodieWriteConfig getWriteConfig(boolean populateMetaFields, String hi
 
   protected void assertWriteStatuses(List<HoodieInternalWriteStatus> writeStatuses, int batches, int size,
                                      Option<List<String>> fileAbsPaths, Option<List<String>> fileNames) {
-    assertWriteStatuses(writeStatuses, batches, size, false, fileAbsPaths, fileNames);
+    assertWriteStatuses(writeStatuses, batches, size, false, fileAbsPaths, fileNames, false);
   }
 
   protected void assertWriteStatuses(List<HoodieInternalWriteStatus> writeStatuses, int batches, int size, boolean areRecordsSorted,
-                                     Option<List<String>> fileAbsPaths, Option<List<String>> fileNames) {
+                                     Option<List<String>> fileAbsPaths, Option<List<String>> fileNames, boolean isHiveStylePartitioning) {
     if (areRecordsSorted) {
       assertEquals(batches, writeStatuses.size());
     } else {
@@ -114,7 +114,8 @@ protected void assertWriteStatuses(List<HoodieInternalWriteStatus> writeStatuses
     int counter = 0;
     for (HoodieInternalWriteStatus writeStatus : writeStatuses) {
       // verify write status
-      assertEquals(HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[counter % 3], writeStatus.getPartitionPath());
+      String actualPartitionPathFormat = isHiveStylePartitioning ? SparkDatasetTestUtils.PARTITION_PATH_FIELD_NAME + "=%s" : "%s";
+      assertEquals(String.format(actualPartitionPathFormat, HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[counter % 3]), writeStatus.getPartitionPath());
       if (areRecordsSorted) {
         assertEquals(writeStatus.getTotalRecords(), size);
       } else {
@@ -142,7 +143,7 @@ protected void assertWriteStatuses(List<HoodieInternalWriteStatus> writeStatuses
         assertEquals(sizeMap.get(HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[counter % 3]), writeStat.getNumWrites());
       }
       assertEquals(fileId, writeStat.getFileId());
-      assertEquals(HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[counter++ % 3], writeStat.getPartitionPath());
+      assertEquals(String.format(actualPartitionPathFormat, HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[counter++ % 3]), writeStat.getPartitionPath());
       assertEquals(0, writeStat.getNumDeletes());
       assertEquals(0, writeStat.getNumUpdateWrites());
       assertEquals(0, writeStat.getTotalWriteErrors());

hudi-spark-datasource/hudi-spark2/src/test/java/org/apache/hudi/internal/TestHoodieBulkInsertDataInternalWriter.java

@@ -18,19 +18,14 @@
 
 package org.apache.hudi.internal;
 
-import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.table.HoodieSparkTable;
 import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.testutils.SparkDatasetTestUtils;
-
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.InternalRow;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -50,7 +45,6 @@
 /**
  * Unit tests {@link HoodieBulkInsertDataInternalWriter}.
  */
-@Disabled("temp")
 public class TestHoodieBulkInsertDataInternalWriter extends
     HoodieBulkInsertInternalWriterTestBase {
 
@@ -106,7 +100,7 @@ public void testDataInternalWriter(boolean sorted, boolean populateMetaFields) t
       Option<List<String>> fileNames = Option.of(new ArrayList<>());
 
       // verify write statuses
-      assertWriteStatuses(commitMetadata.getWriteStatuses(), batches, size, sorted, fileAbsPaths, fileNames);
+      assertWriteStatuses(commitMetadata.getWriteStatuses(), batches, size, sorted, fileAbsPaths, fileNames,false);
 
       // verify rows
       Dataset<Row> result = sqlContext.read().parquet(fileAbsPaths.get().toArray(new String[0]));
@@ -148,14 +142,11 @@ public void testDataInternalWriterHiveStylePartitioning() throws Exception {
       Option<List<String>> fileNames = Option.of(new ArrayList<>());
 
       // verify write statuses
-      assertWriteStatuses(commitMetadata.getWriteStatuses(), batches, size, sorted, fileAbsPaths, fileNames);
+      assertWriteStatuses(commitMetadata.getWriteStatuses(), batches, size, sorted, fileAbsPaths, fileNames, true);
 
       // verify rows
       Dataset<Row> result = sqlContext.read().parquet(fileAbsPaths.get().toArray(new String[0]));
       assertOutput(totalInputRows, result, instantTime, fileNames, populateMetaFields);
-
-      result.collectAsList().forEach(entry -> Assertions.assertTrue(entry.getAs(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString()
-          .contains(SparkDatasetTestUtils.PARTITION_PATH_FIELD_NAME + "=")));
     }
   }
 
@@ -204,7 +195,7 @@ public void testGlobalFailure() throws Exception {
     Option<List<String>> fileAbsPaths = Option.of(new ArrayList<>());
     Option<List<String>> fileNames = Option.of(new ArrayList<>());
     // verify write statuses
-    assertWriteStatuses(commitMetadata.getWriteStatuses(), 1, size / 2, false, fileAbsPaths, fileNames);
+    assertWriteStatuses(commitMetadata.getWriteStatuses(), 1, size / 2, false, fileAbsPaths, fileNames, false);
 
     // verify rows
     Dataset<Row> result = sqlContext.read().parquet(fileAbsPaths.get().toArray(new String[0]));

hudi-spark-datasource/hudi-spark2/src/test/java/org/apache/hudi/internal/TestHoodieDataSourceInternalWriter.java

@@ -24,7 +24,6 @@
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.testutils.HoodieClientTestUtils;
-
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.InternalRow;
@@ -54,7 +53,6 @@
 /**
  * Unit tests {@link HoodieDataSourceInternalWriter}.
  */
-@Disabled("temp")
 public class TestHoodieDataSourceInternalWriter extends
     HoodieBulkInsertInternalWriterTestBase {
 

hudi-spark-datasource/hudi-spark3/src/test/java/org/apache/hudi/spark3/internal/TestHoodieBulkInsertDataInternalWriter.java

@@ -102,7 +102,7 @@ public void testDataInternalWriter(boolean sorted, boolean populateMetaFields) t
       Option<List<String>> fileNames = Option.of(new ArrayList<>());
 
       // verify write statuses
-      assertWriteStatuses(commitMetadata.getWriteStatuses(), batches, size, sorted, fileAbsPaths, fileNames);
+      assertWriteStatuses(commitMetadata.getWriteStatuses(), batches, size, sorted, fileAbsPaths, fileNames, false);
 
       // verify rows
       Dataset<Row> result = sqlContext.read().parquet(fileAbsPaths.get().toArray(new String[0]));