test(spark): add streaming source and writer support coverage

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala

@@ -20,6 +20,7 @@ package org.apache.hudi.functional
 import org.apache.hudi.DataSourceReadOptions
 import org.apache.hudi.DataSourceReadOptions.{START_OFFSET, STREAMING_READ_TABLE_VERSION}
 import org.apache.hudi.DataSourceWriteOptions.{ORDERING_FIELDS, RECORDKEY_FIELD}
+import org.apache.hudi.common.config.HoodieReaderConfig
 import org.apache.hudi.common.model.HoodieTableType
 import org.apache.hudi.common.model.HoodieTableType.{COPY_ON_WRITE, MERGE_ON_READ}
 import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient, HoodieTableVersion}
@@ -294,6 +295,66 @@ class TestStreamingSource extends StreamTest {
     })
   }
 
+  /**
+   * Exercises the legacy incremental streaming path in [[HoodieStreamSourceV1]], which is taken
+   * when the streaming read table version is below EIGHT and the file group reader is disabled.
+   * This drives the [[IncrementalRelationV1]] (COW) / [[MergeOnReadIncrementalRelationV1]] (MOR)
+   * branches of `getBatch` rather than the newer HadoopFsRelation factory path.
+   */
+  private def testLegacyIncrementalStreamSource(tableType: HoodieTableType): Unit = {
+    withTempDir { inputDir =>
+      val tablePath = s"${inputDir.getCanonicalPath}/test_${tableType.name}_legacy_stream"
+      HoodieTableMetaClient.newTableBuilder()
+        .setTableType(tableType)
+        .setTableName(getTableName(tablePath))
+        .setTableVersion(HoodieTableVersion.SIX)
+        .setRecordKeyFields("id")
+        .setOrderingFields("ts")
+        .initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()), tablePath)
+
+      addData(tablePath, Seq(("1", "a1", "10", "000")), tableVersion = HoodieTableVersion.SIX)
+      val df = spark.readStream
+        .format("org.apache.hudi")
+        .option(WRITE_TABLE_VERSION.key, HoodieTableVersion.SIX.versionCode().toString)
+        .option(STREAMING_READ_TABLE_VERSION.key, HoodieTableVersion.SIX.versionCode().toString)
+        // force the legacy (non file-group-reader) incremental relation path
+        .option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key, "false")
+        .load(tablePath)
+        .select("id", "name", "price", "ts")
+
+      testStream(df)(
+        AssertOnQuery { q => q.processAllAvailable(); true },
+        CheckAnswerRows(Seq(Row("1", "a1", "10", "000")), lastOnly = true, isSorted = false),
+        StopStream,
+
+        addDataToQuery(tablePath,
+          Seq(("2", "a2", "12", "000"),
+            ("3", "a3", "12", "000")),
+          tableVersion = HoodieTableVersion.SIX),
+        StartStream(),
+        AssertOnQuery { q => q.processAllAvailable(); true },
+        CheckAnswerRows(
+          Seq(Row("2", "a2", "12", "000"),
+            Row("3", "a3", "12", "000")),
+          lastOnly = true, isSorted = false),
+        StopStream,
+
+        addDataToQuery(tablePath, Seq(("4", "a4", "13", "000")), tableVersion = HoodieTableVersion.SIX),
+        StartStream(),
+        AssertOnQuery { q => q.processAllAvailable(); true },
+        CheckAnswerRows(Seq(Row("4", "a4", "13", "000")), lastOnly = true, isSorted = false)
+      )
+    }
+  }
+
+  test("test cow stream source with legacy file group reader disabled") {
+    testLegacyIncrementalStreamSource(COPY_ON_WRITE)
+  }
+
+  test("test mor stream source with legacy file group reader disabled") {
+    testLegacyIncrementalStreamSource(MERGE_ON_READ)
+  }
+
   private def testCheckpointTranslation(tableName: String,
                                         tableType: HoodieTableType,
                                         writeTableVersion: HoodieTableVersion,
@@ -413,9 +474,10 @@ class TestStreamingSource extends StreamTest {
   }
 
   private def addDataToQuery(inputPath: String,
-                             rows: Seq[(String, String, String, String)]): AssertOnQuery = {
+                             rows: Seq[(String, String, String, String)],
+                             tableVersion: HoodieTableVersion = HoodieTableVersion.current): AssertOnQuery = {
     AssertOnQuery { _=>
-      addData(inputPath, rows)
+      addData(inputPath, rows, tableVersion = tableVersion)
       true
     }
   }

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala

@@ -508,6 +508,52 @@ class TestStructuredStreaming extends HoodieSparkClientTestBase {
     assertEquals(25, metaClient.getActiveTimeline.countInstants())
   }
 
+  /**
+   * Injects a failing micro-batch into [[HoodieStreamingSink]] by pointing the record key at a
+   * non-existent column so that every Hudi write throws. With STREAMING_IGNORE_FAILED_BATCH
+   * enabled the sink must swallow the failure (unpersisting the write RDDs), let the streaming
+   * query complete without crashing, and produce no commit.
+   */
+  @Test
+  def testStructuredStreamingIgnoreFailedBatch(): Unit = {
+    val (sourcePath, destPath) = initStreamingSourceAndDestPath("source", "dest")
+    val records1 = recordsToStrings(dataGen.generateInsertsForPartition(
+      "000", 100, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH)).asScala.toList
+    val inputDF1 = spark.read.json(spark.sparkContext.parallelize(records1, 2))
+    val schema = inputDF1.schema
+    inputDF1.coalesce(1).write.mode(SaveMode.Append).json(sourcePath)
+
+    val failingOpts = commonOpts ++ Map(
+      DataSourceWriteOptions.RECORDKEY_FIELD.key -> "non_existent_field",
+      DataSourceWriteOptions.STREAMING_IGNORE_FAILED_BATCH.key -> "true",
+      DataSourceWriteOptions.STREAMING_RETRY_CNT.key -> "1"
+    )
+
+    val query = spark.readStream
+      .schema(schema)
+      .json(sourcePath)
+      .writeStream
+      .format("org.apache.hudi")
+      .options(failingOpts)
+      .outputMode(OutputMode.Append)
+      .option("checkpointLocation", s"$basePath/checkpoint_ignore")
+      .start(destPath)
+
+    // Must not throw even though the micro-batch fails; the failure is ignored.
+    query.processAllAvailable()
+    query.stop()
+
+    // No completed commit must exist since every batch failed and was ignored.
+    val completedCommits =
+      try {
+        HoodieTestUtils.createMetaClient(storage, destPath)
+          .getActiveTimeline.getCommitsTimeline.filterCompletedInstants().countInstants()
+      } catch {
+        case _: TableNotFoundException => 0
+      }
+    assertEquals(0, completedCommits)
+  }
+
   @ParameterizedTest
   @CsvSource(Array(
     "COPY_ON_WRITE,EVENT_TIME_ORDERING",

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala

@@ -208,4 +208,66 @@ class TestDataSkippingQuery extends HoodieSparkSqlTestBase {
       }
     }
   }
+
+  test("Test column stats data skipping across multiple files with range, IN and equality predicates") {
+    Seq("cow", "mor").foreach { tableType =>
+      withTempDir { tmp =>
+        val tableName = generateTableName
+        withSQLConf(
+          "hoodie.metadata.enable" -> "true",
+          "hoodie.metadata.index.column.stats.enable" -> "true",
+          "hoodie.enable.data.skipping" -> "true",
+          "hoodie.metadata.index.column.stats.column.list" -> "id,price",
+          // keep every commit in its own base file so column-stats pruning has multiple files to skip
+          "hoodie.parquet.small.file.limit" -> "0"
+        ) {
+          spark.sql(
+            s"""
+               |create table $tableName (
+               |  id int,
+               |  name string,
+               |  price double,
+               |  ts long
+               |) using hudi
+               | tblproperties (primaryKey = 'id', orderingFields = 'ts', type = '$tableType')
+               | location '${tmp.getCanonicalPath}'
+                  """.stripMargin)
+          // Three separate commits, each landing in its own base file with disjoint id / price ranges.
+          spark.sql(s"insert into $tableName values (1, 'a1', 10, 1000), (2, 'a2', 20, 1000)")
+          spark.sql(s"insert into $tableName values (11, 'b1', 110, 2000), (12, 'b2', 120, 2000)")
+          spark.sql(s"insert into $tableName values (21, 'c1', 210, 3000), (22, 'c2', 220, 3000)")
+
+          // Equality on the indexed key column -> only the first file qualifies.
+          checkAnswer(s"select id, name, price from $tableName where id = 1")(
+            Seq(1, "a1", 10.0)
+          )
+          // Range predicate that only the last file can satisfy.
+          checkAnswer(s"select id, name, price from $tableName where id > 20")(
+            Seq(21, "c1", 210.0),
+            Seq(22, "c2", 220.0)
+          )
+          // IN predicate touching the first and last files, skipping the middle one.
+          checkAnswer(s"select id, name, price from $tableName where id in (2, 22)")(
+            Seq(2, "a2", 20.0),
+            Seq(22, "c2", 220.0)
+          )
+          // Range on a second indexed column keeps only the middle file.
+          checkAnswer(s"select id, name, price from $tableName where price >= 110 and price < 210")(
+            Seq(11, "b1", 110.0),
+            Seq(12, "b2", 120.0)
+          )
+          // Predicate that matches nothing -> all files pruned, empty result.
+          checkAnswer(s"select id, name, price from $tableName where id = 999")()
+
+          // Data skipping must not change results compared to a full scan.
+          withSQLConf("hoodie.enable.data.skipping" -> "false") {
+            checkAnswer(s"select id, name, price from $tableName where id in (2, 22)")(
+              Seq(2, "a2", 20.0),
+              Seq(22, "c2", 220.0)
+            )
+          }
+        }
+      }
+    }
+  }
 }