Fix hudi connector gets stuck #19506

[willzgw]

Description

• Fix HUDI connector gets stuck when an exception occurs
  • Solve HUDI connector stuck by using ExceptionCallback to ensure the asyncQueue is closed when an exception occurs.
  • Optimize the handling of multiple Futures with Futures.whenAllComplete instead of for_loop.
• Add a new HUDI session property: ignore_absent_partitions
  • Boolean property.
  • When set to false, the Trino HUDI connector will report an error if the partition does not exist; when set to true, non-existing partitions will be ignored.

Additional context and related issues

• Trino Hudi connector gets stuck while attempting to read empty partitions of a partitioned Hudi table  #19506

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix: Trino Hudi connector gets stuck while attempting to read empty partitions of a partitioned Hudi table. ({issue}`19506 `)
* New HUDI session property: ignore_absent_partitions

[ebyhr]

Can you add a test?

[electrum]

Is it possible to add a test for this?

[willzgw]

Hi @electrum
I'm not quite sure what to do when we get an empty partition, ignore it or throw an exception?

[uroell]

Hi @electrum I'm not quite sure what to do when we get an empty partition, ignore it or throw an exception?

Hi @willzgw, thank you very much for your contribution! This helps a lot. Just a suggestion for your question: In the hive connector there is this property called "hive.ignore-absent-partitions" with default "false" to switch the behavior. Maybe you could implement the same for the hudi connector?

[codesorcery]

#20151 fixes the underlying problem of Trino failing on empty Hudi partitions. So when this PR gets merged, there should be no reason to handle it here.
So it should be enough to reduce the scope of this PR to something like

--- a/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
+++ b/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
@@ -14,6 +14,7 @@
 package io.trino.plugin.hudi.split;
 
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import io.trino.plugin.hive.util.AsyncQueue;
 import io.trino.plugin.hudi.HudiTableHandle;
 import io.trino.plugin.hudi.partition.HudiPartitionInfoLoader;
@@ -28,8 +29,8 @@ import java.util.List;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
-import java.util.concurrent.Future;
 
+import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static io.trino.plugin.hudi.HudiErrorCode.HUDI_CANNOT_OPEN_SPLIT;
 import static io.trino.plugin.hudi.HudiSessionProperties.getSplitGeneratorParallelism;
 import static java.util.Objects.requireNonNull;
@@ -66,7 +67,7 @@ public class HudiBackgroundSplitLoader
     {
         Deque<String> partitionQueue = new ConcurrentLinkedDeque<>(partitions);
         List<HudiPartitionInfoLoader> splitGeneratorList = new ArrayList<>();
-        List<Future> splitGeneratorFutures = new ArrayList<>();
+        List<ListenableFuture<Void>> splitGeneratorFutures = new ArrayList<>();
 
         // Start a number of partition split generators to generate the splits in parallel
         for (int i = 0; i < splitGeneratorNumThreads; i++) {
@@ -79,16 +80,18 @@ public class HudiBackgroundSplitLoader
             // Let the split generator stop once the partition queue is empty
             generator.stopRunning();
         }
-
-        // Wait for all split generators to finish
-        for (Future future : splitGeneratorFutures) {
-            try {
-                future.get();
-            }
-            catch (InterruptedException | ExecutionException e) {
-                throw new TrinoException(HUDI_CANNOT_OPEN_SPLIT, "Error generating Hudi split", e);
-            }
+        try {
+            // Wait for all split generators to finish
+            Futures.whenAllComplete(splitGeneratorFutures) // also succeeds when some tasks fail
+                    .run(asyncQueue::finish, directExecutor())
+                    .get(); // will throw an ExecutionException when one of the tasks failed
+        }
+        catch (ExecutionException e) {
+            throw new TrinoException(HUDI_CANNOT_OPEN_SPLIT, "Error generating Hudi split", e);
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new TrinoException(HUDI_CANNOT_OPEN_SPLIT, "Error generating Hudi split", e);
         }
-        asyncQueue.finish();
     }
 }

[willzgw]

Hi @ebyhr @electrum
I tried to add a test to TestHudiSmokeTest but Trino will throw exception when creating table due to partition not exists.
It seems not convenient to add test cases.
Do you have any suggestions?

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[ebyhr]

I tried to add a test to TestHudiSmokeTest but Trino will throw exception when creating table due to partition not exists.

@willzgw Sorry for my late reply. We could create the table on Spark and add the contents under resources directory. You can refer to resources/hudi-testing-data/hudi_cow_pt_tbl.

[willzgw]

I tried to add a test to TestHudiSmokeTest but Trino will throw exception when creating table due to partition not exists.

@willzgw Sorry for my late reply. We could create the table on Spark and add the contents under resources directory. You can refer to resources/hudi-testing-data/hudi_cow_pt_tbl.

Hi @ebyhr
Thanks for your advice.
I've actually tried this approach before. We want to test the case where the Partition does not exist, which requires the metadata to contain files that do not actually exist (in resource). However, it will report an error by prompting that the corresponding file is missing from the Resource.

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[mosabua]

Should we merge this PR given that you approved @electrum ?

[codope]

Changes look good to me. One thing is we also need to update the huddi connector page for the new config introuced by this PR. I am ok if it's done in this PR or as a followup. Also, please rebase.

[codesorcery]

Do I see it right, that one then has to explicitly set hudi.ignore-absent-partitions=false to get the current behavior? IMHO, that should then definitely be highlighted in the release notes, since it's a breaking change.

[chenjian2664]

Please rearrange the commits and squash to two, one is fixing the stuck issue, another is adding the session property.

[mosabua]

@electrum can we merge this?

[chenjian2664]

LGTM % commit message

Revisit the first commit changes and think the commit message could be improved.
The changes is actual Use Futures.whenAllComplete to handle splitGeneratorFutures (in HudiBackgroundSplitLoader)
Put the commit body something like:

This commit improved async handling using `Futures.whenAllComplete` 
to wait/handle all the splitGeneratorFutures instead of using 
for-loop to handle the future one by one, the change guarantee the 
`asyncQueue.finish()` always be called to avoid the split leak

[willzgw]

@ebyhr @mosabua
Hi master, Could you please help review this PR.
I just rebased the branch and updated the description.
This modification has been validated at Shopee, and we've been running it in our production environment for over a year.

[yihua]

LGTM
@ebyhr @mosabua @electrum could you help merge the PR if no more revision is required?

[mosabua]

Thank you @electrum @yihua @codope @chenjian2664 and @willzgw