This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 75c587bd6372895858f9a81d79ed44ab5056d82a Author: Y Ethan Guo <[email protected]> AuthorDate: Sat Jul 4 09:20:17 2026 -0700 test(spark): add streaming source and writer support coverage (#19166) (cherry picked from commit f6ca15ffd011fe6a08fcee8ab9835cc4132577bf) --- .../hudi/functional/TestStreamingSource.scala | 66 +++++++++++++++++++++- .../hudi/functional/TestStructuredStreaming.scala | 46 +++++++++++++++ .../sql/hudi/feature/TestDataSkippingQuery.scala | 62 ++++++++++++++++++++ 3 files changed, 172 insertions(+), 2 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala index a35d49993bd5..ffaae686eb4d 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala +++ b/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 } } diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala index 2ae305a8ccc0..b9a672bb0882 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStructuredStreaming.scala +++ b/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", diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala index 60513bd9c028..38cc323f58c6 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala +++ b/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) + ) + } + } + } + } + } }
