Tmp commit

Author: huberylee

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/build/BuildTaskExecutor.java

@@ -32,7 +32,7 @@
 import org.apache.hudi.common.util.HoodieTimer;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.exception.HoodieBuildException;
-import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.io.storage.HoodieFileReaderFactory;
 import org.apache.hudi.secondary.index.HoodieSecondaryIndex;
 import org.apache.hudi.secondary.index.SecondaryIndexBuilder;
@@ -233,8 +233,8 @@ private void addBatch(GenericRecord[] records, int size) {
     indexBuilders.forEach(indexBuilder -> {
       try {
         indexBuilder.addBatch(records, size);
-      } catch (IOException e) {
-        throw new HoodieIOException("Add records to index builder failed, baseFile: "
+      } catch (Exception e) {
+        throw new HoodieException("Add records to index builder failed, baseFile: "
             + baseFilePath + ", builderName: " + indexBuilder.getName(), e);
       }
     });

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/build/SparkExecuteBuildCommitActionExecutor.java

@@ -46,12 +46,15 @@
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.catalyst.QualifiedTableName;
 import org.apache.spark.sql.catalyst.catalog.SessionCatalog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
 
 public class SparkExecuteBuildCommitActionExecutor<T extends HoodieRecordPayload, I, K, O>
     extends BaseActionExecutor<T, I, K, O, HoodieBuildCommitMetadata> {
+  private static final Logger LOG = LoggerFactory.getLogger(SparkExecuteBuildCommitActionExecutor.class);
 
   private HoodieBuildPlan buildPlan;
 
@@ -85,17 +88,25 @@ public HoodieBuildCommitMetadata execute() {
     SerializableConfiguration conf = new SerializableConfiguration(context.getHadoopConf().get());
     JavaSparkContext sparkContext = HoodieSparkEngineContext.getSparkContext(context);
     HoodieJavaRDD<BuildStatus> buildStatusRDD =
-        HoodieJavaRDD.of(sparkContext.parallelize(buildTasks, buildTasks.size())
-            .mapPartitions(buildTasksPartition -> {
-              List<BuildStatus> buildStatuses = new ArrayList<>();
-              buildTasksPartition.forEachRemaining(buildTask -> {
-                BuildStatus buildStatus = new BuildTaskExecutor(buildTask, table.getConfig().getBasePath(),
-                    indexFolderPath, serializableSchema, conf).execute();
-                buildStatuses.add(buildStatus);
-              });
+        null;
+    try {
+      buildStatusRDD = HoodieJavaRDD.of(sparkContext.parallelize(buildTasks, buildTasks.size())
+          .mapPartitions(buildTasksPartition -> {
+            List<BuildStatus> buildStatuses = new ArrayList<>();
+            buildTasksPartition.forEachRemaining(buildTask -> {
+              BuildStatus buildStatus = new BuildTaskExecutor(buildTask, table.getConfig().getBasePath(),
+                  indexFolderPath, serializableSchema, conf).execute();
+              buildStatuses.add(buildStatus);
+            });
 
-              return buildStatuses.iterator();
-            }));
+            return buildStatuses.iterator();
+          }));
+    } catch (Exception e) {
+      String tableName = table.getMetaClient().getTableConfig().getTableName();
+      String basePath = table.getConfig().getBasePath();
+      LOG.error("Build execute failed, table: {}, basePath: {}, schema: {}", tableName, basePath, schema);
+      throw new RuntimeException(e);
+    }
 
     // Invalidate cached table for queries do not access related table through {@code DefaultSource}
     SessionCatalog sessionCatalog = SparkSession.active().sessionState().catalog();

hudi-common/src/main/java/org/apache/hudi/secondary/index/lucene/LuceneIndexBuilder.java

@@ -96,7 +96,7 @@ public LuceneIndexBuilder(HoodieBuildTaskConfig indexConfig) {
       fieldTypes[i] = AvroInternalSchemaConverter.buildTypeFromAvroSchema(field.schema()).typeId();
       fieldNames.add(field.name());
     });
-    LOG.info("Init lucene index builder ok, name: {}, indexFields: {}", name, fieldNames);
+    LOG.warn("Init lucene index builder ok, name: {}, indexFields: {}", name, fieldNames);
   }
 
   @Override

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestBuildProcedure.scala

@@ -23,10 +23,11 @@ import org.apache.hadoop.fs.Path
 import org.apache.hudi.common.table.timeline.{HoodieActiveTimeline, HoodieInstant}
 import org.apache.hudi.common.util.{Option => HOption}
 import org.apache.hudi.{HoodieCLIUtils, HoodieDataSourceHelpers}
+import org.apache.spark.internal.Logging
 
 import scala.collection.JavaConverters.asScalaIteratorConverter
 
-class TestBuildProcedure extends HoodieSparkProcedureTestBase {
+class TestBuildProcedure extends HoodieSparkProcedureTestBase with Logging{
   test("Test Call run_build Procedure by Table for None-Partitioned Table") {
     withTempDir { tmp =>
       Seq("cow", "mor").foreach { tableType =>
@@ -51,6 +52,8 @@ class TestBuildProcedure extends HoodieSparkProcedureTestBase {
         spark.sql(s"insert into $tableName values(2, 'a2', 10, 1001)")
         spark.sql(s"insert into $tableName values(3, 'a3', 10, 1002)")
         val client = HoodieCLIUtils.createHoodieClientFromPath(spark, basePath, Map.empty)
+        val schema = client.getConfig.getSchema
+        logWarning(s"Exec build, table: ${tableName}, basePath: ${basePath}, schema: ${schema}")
         // Generate the first build plan
         val firstScheduledBuild = HoodieActiveTimeline.createNewInstantTime
         client.scheduleBuildAtInstant(firstScheduledBuild, HOption.empty())
@@ -164,6 +167,8 @@ class TestBuildProcedure extends HoodieSparkProcedureTestBase {
         spark.sql(s"insert into $tableName values(2, 'a2', 10, 1001)")
         spark.sql(s"insert into $tableName values(3, 'a3', 10, 1002)")
         val client = HoodieCLIUtils.createHoodieClientFromPath(spark, basePath, Map.empty)
+        val schema = client.getConfig.getSchema
+        logWarning(s"Exec build, table: ${tableName}, basePath: ${basePath}, schema: ${schema}")
         // Generate the first build plan
         val firstScheduledBuild = HoodieActiveTimeline.createNewInstantTime
         client.scheduleBuildAtInstant(firstScheduledBuild, HOption.empty())
@@ -280,6 +285,8 @@ class TestBuildProcedure extends HoodieSparkProcedureTestBase {
         spark.sql(s"insert into $tableName values(2, 'a2', 10, 1001)")
         spark.sql(s"insert into $tableName values(3, 'a3', 10, 1002)")
         val client = HoodieCLIUtils.createHoodieClientFromPath(spark, basePath, Map.empty)
+        val schema = client.getConfig.getSchema
+        logWarning(s"Exec build, table: ${tableName}, basePath: ${basePath}, schema: ${schema}")
         // Generate the first build plan
         val firstScheduledBuild = HoodieActiveTimeline.createNewInstantTime
         client.scheduleBuildAtInstant(firstScheduledBuild, HOption.empty())