(auron) branch master updated: [AURON #2273] Fallback Hudi incremental queries from native scan conversion (#2274)

[sl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

slfan1989 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git


The following commit(s) were added to refs/heads/master by this push:
     new e303eb384 [AURON #2273] Fallback Hudi incremental queries from native scan conversion (#2274)
e303eb384 is described below

commit e303eb384b74f7699630ebbc71cac4e35c270020
Author: Ming Wei <we...@gmail.com>
AuthorDate: Sat May 23 11:54:18 2026 +0800

    [AURON #2273] Fallback Hudi incremental queries from native scan conversion (#2274)
    
    # Which issue does this PR close?
    
    Closes #2273
    
    # Rationale for this change
    
    Hudi incremental queries depend on Hudi timeline and instant filtering
    semantics. Native Hudi scan currently converts supported Hudi file scans
    to native Parquet/ORC scans, but native scan does not implement
    incremental query semantics. These queries should fallback to Spark/Hudi
    until incremental scan support is explicitly added.
    
    # What changes are included in this PR?
    
    - Detect incremental Hudi query options in `HudiScanSupport`.
    - Fallback native scan conversion when:
      - `hoodie.datasource.query.type=incremental`
      - `hoodie.datasource.read.begin.instanttime`
      - `hoodie.datasource.read.end.instanttime`
    - Add unit coverage for incremental query options, including
    case-insensitive option keys.
    
    # Are there any user-facing changes?
    Yes. Hudi incremental queries will stay on Spark/Hudi scan instead of
    being converted to native scan.
    
    # How was this patch tested?
    - Added unit coverage in `HudiScanSupportSuite`.
    
    Signed-off-by: weimingdiit <we...@gmail.com>
---
 .../spark/sql/auron/hudi/HudiScanSupport.scala     | 12 +++++++++-
 .../sql/auron/hudi/HudiScanSupportSuite.scala      | 27 ++++++++++++++++++++++
 2 files changed, 38 insertions(+), 1 deletion(-)

diff --git a/thirdparty/auron-hudi/src/main/scala/org/apache/spark/sql/auron/hudi/HudiScanSupport.scala b/thirdparty/auron-hudi/src/main/scala/org/apache/spark/sql/auron/hudi/HudiScanSupport.scala
index 0ebcb1429..55392c52e 100644
--- a/thirdparty/auron-hudi/src/main/scala/org/apache/spark/sql/auron/hudi/HudiScanSupport.scala
+++ b/thirdparty/auron-hudi/src/main/scala/org/apache/spark/sql/auron/hudi/HudiScanSupport.scala
@@ -46,6 +46,8 @@ object HudiScanSupport extends Logging {
     "hoodie.table.type")
   private val hudiQueryTypeKeys =
     Seq("hoodie.datasource.query.type", "hoodie.datasource.view.type")
+  private val hudiIncrementalInstantKeys =
+    Seq("hoodie.datasource.read.begin.instanttime", "hoodie.datasource.read.end.instanttime")
   private val hudiBaseFileFormatKeys = Seq(
     "hoodie.table.base.file.format",
     "hoodie.datasource.write.base.file.format",
@@ -115,6 +117,9 @@ object HudiScanSupport extends Logging {
     if (hasTimeTravel(options)) {
       return false
     }
+    if (hasIncrementalQuery(options)) {
+      return false
+    }
 
     val tableType = tableTypeFromOptions(options)
       .orElse(tableTypeFromCatalog(catalogTable))
@@ -128,7 +133,7 @@ object HudiScanSupport extends Logging {
     queryType match {
       case Some(query) if readOptimizedQueryTypes.contains(query) => true
       case Some("snapshot") => !isMor
-      case Some("incremental" | "realtime") => false
+      case Some("realtime") => false
       case Some(_) => false
       case None =>
         // MOR snapshot reads may need log-file merging. Native scan is safe only
@@ -257,6 +262,11 @@ object HudiScanSupport extends Logging {
         "hoodie.datasource.read.as.of.timestamp")).isDefined
   }
 
+  private def hasIncrementalQuery(options: Map[String, String]): Boolean = {
+    queryTypeFromOptions(options).exists(_.equalsIgnoreCase("incremental")) ||
+    caseInsensitiveValue(options, hudiIncrementalInstantKeys).isDefined
+  }
+
   private def normalizePath(rawPath: String): String = {
     try {
       val uri = new URI(rawPath)
diff --git a/thirdparty/auron-hudi/src/test/scala/org/apache/spark/sql/auron/hudi/HudiScanSupportSuite.scala b/thirdparty/auron-hudi/src/test/scala/org/apache/spark/sql/auron/hudi/HudiScanSupportSuite.scala
index 1d51dd48f..dbb1f0c46 100644
--- a/thirdparty/auron-hudi/src/test/scala/org/apache/spark/sql/auron/hudi/HudiScanSupportSuite.scala
+++ b/thirdparty/auron-hudi/src/test/scala/org/apache/spark/sql/auron/hudi/HudiScanSupportSuite.scala
@@ -322,6 +322,33 @@ class HudiScanSupportSuite extends SparkFunSuite with SharedSparkSession {
           "hoodie.datasource.query.type" -> "snapshot")))
   }
 
+  test("hudi isSupported rejects incremental query options") {
+    val fileFormatName =
+      "org.apache.spark.sql.execution.datasources.parquet.HoodieParquetFileFormat"
+    val cowOptions = Map("hoodie.datasource.write.table.type" -> "COPY_ON_WRITE")
+
+    assert(
+      !HudiScanSupport.isSupported(
+        fileFormatName,
+        cowOptions + ("hoodie.datasource.query.type" -> "incremental")))
+    assert(
+      !HudiScanSupport.isSupported(
+        fileFormatName,
+        cowOptions + ("hoodie.datasource.view.type" -> "incremental")))
+    assert(
+      !HudiScanSupport.isSupported(
+        fileFormatName,
+        cowOptions + ("hoodie.datasource.read.begin.instanttime" -> "20240101010101")))
+    assert(
+      !HudiScanSupport.isSupported(
+        fileFormatName,
+        cowOptions + ("hoodie.datasource.read.end.instanttime" -> "20240102010101")))
+    assert(
+      !HudiScanSupport.isSupported(
+        fileFormatName,
+        cowOptions + ("Hoodie.DataSource.Read.Begin.InstantTime" -> "20240101010101")))
+  }
+
   test("hudi isSupported rejects non-Hudi formats") {
     assert(
       !HudiScanSupport.isSupported(