This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c9dbde0f4553 test(spark): add incremental read-path relation coverage
(#19404)
c9dbde0f4553 is described below
commit c9dbde0f455314c29aacf6e1acbd80a7a8694800
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Jul 31 05:08:52 2026 -0700
test(spark): add incremental read-path relation coverage (#19404)
* test(spark): add incremental read-path relation coverage
* Drop redundant per-partition sum assertion
The assertEquals map comparison already implies the per-partition sums are
equal, so the follow-up assertTrue can never catch a distinct failure. Remove
it and the now-unused assertTrue import.
* test(spark): address review comments on incremental relation coverage
- MOR glob arms write log-only file slices via the in-memory index and
assert merged upsert values plus inputFiles paths per glob
- thread enableInlineCluster through addDataToQuery and cluster in the
legacy streaming test to reach the replaced-file filtering of the
legacy incremental relations
- pin the legacy (non file-group-reader) plan shape in the streaming test
- add a TestIncrementalReadWithFullTableScan case pinning the divergent
v6/v8 default of incr.fallback.fulltablescan.enable on a cleaned span
---------
Co-authored-by: voon <[email protected]>
---
.../TestIncrementalReadWithFullTableScan.scala | 105 ++++++++++-
.../TestIncrementalReadWithPathGlob.scala | 205 +++++++++++++++++++++
.../hudi/functional/TestStreamingSource.scala | 57 ++++--
3 files changed, 351 insertions(+), 16 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
index dc6a8b5d760d..6c26e4fe1062 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
@@ -34,7 +34,7 @@ import org.junit.jupiter.api.{AfterEach, BeforeEach}
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows,
assertTrue}
import org.junit.jupiter.api.function.Executable
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.EnumSource
+import org.junit.jupiter.params.provider.{CsvSource, EnumSource}
import java.time.Instant
import java.util.Date
@@ -158,6 +158,109 @@ class TestIncrementalReadWithFullTableScan extends
HoodieSparkClientTestBase {
runIncrementalQueryAndCompare(startUncleanedCompletionTs,
endUncleanedCompletionTs, 1, false)
}
+ /**
+ * Pins the divergent DEFAULT of {@code
hoodie.datasource.read.incr.fallback.fulltablescan.enable}
+ * across table versions. {@code MergeOnReadIncrementalRelationV1} (selected
for table version
+ * below 8) hardcodes "false" when the option is absent, while
+ * {@code MergeOnReadIncrementalRelationV2} (table version 8 and above)
reads the ConfigProperty
+ * default, which is "true" since HUDI-8624. Batch incremental reads route
through these two
+ * relations for both COW and MOR (see the {@code
HoodieCopyOnWriteIncrementalHadoopFsRelationFactory}
+ * and {@code HoodieMergeOnReadIncrementalHadoopFsRelationFactory} V1/V2
subclasses), so the
+ * divergence is table-version driven, not table-type driven. Note that
+ * {@code INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN_FOR_NON_EXISTING_FILES} is
an
+ * alias for the very same ConfigProperty (same key, same default), so a
single key governs both
+ * versions and only the in-code default differs.
+ *
+ * The timeline shape here is the one proven by {@code
testFailEarlyForIncrViewQueryForNonExistingFiles}:
+ * 10 insert commits with cleaning and archival configured so that the
earliest commits are both
+ * cleaned and archived. The discriminating query is a narrow span over the
two oldest commits that
+ * are still on the active timeline but whose data files have already been
cleaned; there the
+ * default decides between failing and silently degrading into a full table
scan.
+ *
+ * A span that starts at "000" does NOT discriminate: both versions return
every record. With an
+ * archived start instant V1 keeps only the latest merged file slice of each
affected file group
+ * (which still exists) and its record filter degenerates to {@code
_hoodie_commit_time > "000"},
+ * so no file is missing and no fallback is needed. That case is asserted
too, so nobody
+ * "simplifies" the discriminating span away and silently loses the pin.
+ */
+ @ParameterizedTest
+ @CsvSource(value = Array(
+ "COPY_ON_WRITE,6",
+ "COPY_ON_WRITE,8",
+ "MERGE_ON_READ,6",
+ "MERGE_ON_READ,8"))
+ def testDefaultFallbackBehaviorAcrossTableVersions(tableType: String,
tableVersion: Int): Unit = {
+ val commonOpts = Map(
+ "hoodie.insert.shuffle.parallelism" -> "4",
+ "hoodie.upsert.shuffle.parallelism" -> "4",
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "timestamp",
+ HoodieWriteConfig.TBL_NAME.key -> "hoodie_test",
+ HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key -> "1",
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType,
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key -> tableVersion.toString,
+ HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> "false"
+ )
+
+ val numCommits = 10
+ for (i <- 1 to numCommits) {
+ val records = recordsToStrings(dataGen.generateInserts("%05d".format(i),
perBatchSize)).asScala.toList
+ val inputDF = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ inputDF.write.format("org.apache.hudi")
+ .options(commonOpts)
+ .option("hoodie.clean.commits.retained", "3")
+ .option("hoodie.keep.min.commits", "4")
+ .option("hoodie.keep.max.commits", "7")
+ .option(DataSourceWriteOptions.OPERATION.key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ }
+
+ val metaClient = createMetaClient(spark, basePath)
+ assertEquals(tableVersion,
metaClient.getTableConfig.getTableVersion.versionCode(),
+ "table should be written at the requested version to select the V1/V2
relation")
+
assertTrue(metaClient.getArchivedTimeline.filterCompletedInstants().countInstants()
> 0,
+ "early commits must be archived so the timeline matches
testFailEarlyForIncrViewQueryForNonExistingFiles")
+
+ // The two oldest commits still on the active timeline: not archived, but
already cleaned.
+ val completedCommits =
metaClient.getCommitsTimeline.filterCompletedInstants()
+ val cleanedStart = completedCommits.nthInstant(0).get()
+ val cleanedEnd = completedCommits.nthInstant(1).get()
+ // V1 ranges over requested time, V2 over completion time.
+ val (startTs, endTs) = if (tableVersion < 8) {
+ (cleanedStart.requestedTime, cleanedEnd.requestedTime)
+ } else {
+ (cleanedStart.getCompletionTime, cleanedEnd.getCompletionTime)
+ }
+
+ // No fallback option is set on any of these queries: what happens is
purely the in-code default
+ // of the relation that the table version selects.
+ def readSpan(start: String, end: Option[String]): Long = {
+ var reader = spark.read.format("org.apache.hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key(),
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key(), start)
+ end.foreach(e => reader =
reader.option(DataSourceReadOptions.END_COMMIT.key(), e))
+ reader.load(basePath).count()
+ }
+
+ if (tableVersion < 8) {
+ // V1 hardcodes the default to false, so nothing falls back and the
query fails on the cleaned
+ // data files of the span.
+ shouldThrowSparkExceptionIfFallbackIsFalse(() => readSpan(startTs,
Some(endTs)))
+ } else {
+ // V2 takes the config default of true, so the same span silently
degrades into a full table
+ // scan and still returns the single batch the span covers.
+ assertEquals(perBatchSize, readSpan(startTs, Some(endTs)),
+ "table version 8 must silently fall back to a full table scan for a
cleaned span")
+ }
+
+ // Both versions agree once the span starts before the timeline: see the
class comment on why
+ // this query needs no fallback at all.
+ assertEquals(numCommits * perBatchSize, readSpan("000", None),
+ "a span starting at 000 returns every record on both table versions")
+ }
+
private def runIncrementalQueryAndCompare(
startTs: String,
endTs: String,
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala
new file mode 100644
index 000000000000..8a63ed7221bb
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions}
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.HoodieTableConfig
+import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{recordsToStrings,
DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH}
+import org.apache.hudi.config.{HoodieIndexConfig, HoodieWriteConfig}
+import org.apache.hudi.index.HoodieIndex.IndexType
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+import org.junit.jupiter.api.{AfterEach, BeforeEach}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.CsvSource
+
+import java.util.{List => JList}
+
+import scala.collection.JavaConverters._
+
+/**
+ * Coverage for the {@code hoodie.datasource.read.incr.path.glob} file-slice
filtering branch of the
+ * batch incremental relations {@code MergeOnReadIncrementalRelationV1} (table
version 6) and
+ * {@code MergeOnReadIncrementalRelationV2} (table version 8). Both the COW
and the MOR batch
+ * incremental read routes build a {@code HoodieIncrementalFileIndex} backed
by these relations (see
+ * {@code HoodieCopyOnWriteIncrementalHadoopFsRelationFactoryV1/V2} and the
MOR factories), and the
+ * table version selects V1 vs V2 in {@code DefaultSource}. The COW arms cover
the base-file path of
+ * {@code filterFileSlices}, since the COW factory builds the file index with
+ * {@code includeLogFiles = false}. The MOR arms write with the in-memory
index, which can index log
+ * files, so their inserts land in log-only file slices; those arms therefore
exercise the
+ * {@code getLatestLogFile} fallback of {@code filterFileSlices} and, because
the MOR factory builds
+ * the file index with {@code includeLogFiles = true}, the log-file branch of
+ * {@code HoodieIncrementalFileIndex.inputFiles}.
+ *
+ * Existing glob coverage ({@code TestCOWDataSourceStorage}, plus {@code
TestDataSourceForBootstrap}
+ * for bootstrapped COW) exercises only COW at the default table
+ * version and asserts counts alone; this pins the per-partition result of the
glob (including the
+ * glob-matches-nothing empty branch) across both table versions and both
table types, the merged
+ * value of an upserted record inside the glob, and the file paths reported by
{@code inputFiles}.
+ */
+class TestIncrementalReadWithPathGlob extends HoodieSparkClientTestBase {
+
+ private var spark: SparkSession = _
+ // generateInsertsForPartition sets the record's partition_path field to the
given partition.
+ private val firstPartition = DEFAULT_FIRST_PARTITION_PATH // 2016/03/15
+ private val secondPartition = DEFAULT_SECOND_PARTITION_PATH // 2015/03/16
+ private val numFirst = 8
+ private val numSecond = 5
+ private val numUpdated = 3
+ // generateUpdates stamps the rider field with "rider-" + the seed it is
given.
+ private val updatedRider = "rider-003"
+
+ @BeforeEach override def setUp(): Unit = {
+ setTableName("hoodie_test")
+ initPath()
+ initSparkContexts()
+ spark = sqlContext.sparkSession
+ initTestDataGenerator()
+ initHoodieStorage()
+ }
+
+ @AfterEach override def tearDown(): Unit = {
+ spark = null
+ cleanupResources()
+ }
+
+ @ParameterizedTest
+ @CsvSource(value = Array(
+ "COPY_ON_WRITE,6",
+ "COPY_ON_WRITE,8",
+ "MERGE_ON_READ,6",
+ "MERGE_ON_READ,8"))
+ def testIncrementalPathGlobPartitionFiltering(tableType: String,
tableVersion: Int): Unit = {
+ val baseOpts = Map(
+ "hoodie.insert.shuffle.parallelism" -> "2",
+ "hoodie.upsert.shuffle.parallelism" -> "2",
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition_path",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "timestamp",
+ HoodieWriteConfig.TBL_NAME.key -> "hoodie_test",
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType,
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key -> tableVersion.toString,
+ HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> "false")
+ // The default SIMPLE index cannot index log files, so MOR inserts would
land in base files and
+ // every file slice would carry a base file. The in-memory index can,
which keeps the MOR arms
+ // on log-only file slices and drives the log-file branches of the
relation and the file index.
+ val opts = if
(DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL.equals(tableType)) {
+ baseOpts + (HoodieIndexConfig.INDEX_TYPE.key ->
IndexType.INMEMORY.toString)
+ } else {
+ baseOpts
+ }
+
+ // First commit writes only the first partition, second commit only the
second partition, so the
+ // glob boundary lines up with the commit boundary and expected counts are
deterministic.
+ val firstBatch = writeBatch("001", numFirst, firstPartition, opts,
SaveMode.Overwrite)
+ writeBatch("002", numSecond, secondPartition, opts, SaveMode.Append)
+ // Third commit upserts a subset of the first batch. Updates keep the
record key and partition,
+ // so the per-partition counts must stay the same and the updated rows
must be merged in place.
+ writeUpdateBatch("003", firstBatch.subList(0, numUpdated), opts)
+
+ val metaClient = createMetaClient(spark, basePath)
+ assertEquals(tableVersion,
metaClient.getTableConfig.getTableVersion.versionCode(),
+ "table should be written at the requested version to select the V1/V2
relation")
+
+ // Without a glob the incremental span returns every record from both
partitions.
+ val allDf = incrementalRead(None)
+ assertPartitionCounts(allDf, Map(firstPartition -> numFirst,
secondPartition -> numSecond))
+ val allInputFiles = allDf.inputFiles
+ if (DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL.equals(tableType)) {
+ // Guards the coverage this test is here for: if MOR inserts ever stop
landing in log files the
+ // getLatestLogFile fallback of filterFileSlices and the log branch of
inputFiles go untested.
+ assertTrue(allInputFiles.forall(_.contains(".log.")),
+ "MOR file slices must be log only under the in-memory index: " +
allInputFiles.mkString(","))
+ }
+ assertTrue(allInputFiles.exists(_.contains(firstPartition)),
+ "without a glob inputFiles must report files from the first partition: "
+ allInputFiles.mkString(","))
+ assertTrue(allInputFiles.exists(_.contains(secondPartition)),
+ "without a glob inputFiles must report files from the second partition:
" + allInputFiles.mkString(","))
+
+ // Glob restricted to the first partition returns only its records.
+ val firstDf = incrementalRead(Some("/2016/*/*/*"))
+ assertPartitionCounts(firstDf, Map(firstPartition -> numFirst))
+ assertEquals(numUpdated, firstDf.filter(s"rider =
'$updatedRider'").count(),
+ "the upserted records must be merged into the glob result, not
duplicated or dropped")
+ val firstInputFiles = firstDf.inputFiles
+ assertTrue(firstInputFiles.nonEmpty, "a glob matching the first partition
must report input files")
+ assertTrue(firstInputFiles.forall(_.contains(firstPartition)),
+ "every file reported for the first-partition glob must live under it: "
+ firstInputFiles.mkString(","))
+
+ // Glob restricted to the second partition returns only its records.
+ val secondDf = incrementalRead(Some("/2015/*/*/*"))
+ assertPartitionCounts(secondDf, Map(secondPartition -> numSecond))
+ val secondInputFiles = secondDf.inputFiles
+ assertTrue(secondInputFiles.nonEmpty, "a glob matching the second
partition must report input files")
+ assertTrue(secondInputFiles.forall(_.contains(secondPartition)),
+ "every file reported for the second-partition glob must live under it: "
+ secondInputFiles.mkString(","))
+
+ // Glob that matches no partition exercises the empty-result branch of the
relation.
+ val emptyDf = incrementalRead(Some("/9999/*/*/*"))
+ assertEquals(0, emptyDf.count(),
+ "a glob matching no partition path must yield an empty result")
+ assertEquals(0, emptyDf.inputFiles.length,
+ "a glob matching no partition path must report no input files")
+ }
+
+ /**
+ * Writes an insert batch and returns the generated records so that later
batches can update them.
+ * The seed only feeds the generated field values (rider-001, driver-001,
...); it does not control
+ * the commit time of the write.
+ */
+ private def writeBatch(seed: String, n: Int, partition: String,
+ opts: Map[String, String], mode: SaveMode):
JList[HoodieRecord[_]] = {
+ val records = dataGen.generateInsertsForPartition(seed, n, partition)
+ write(records, opts, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL, mode)
+ records
+ }
+
+ private def writeUpdateBatch(seed: String, baseRecords:
JList[HoodieRecord[_]],
+ opts: Map[String, String]): Unit = {
+ write(dataGen.generateUpdates(seed, baseRecords), opts,
+ DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL, SaveMode.Append)
+ }
+
+ private def write(records: JList[HoodieRecord[_]], opts: Map[String, String],
+ operation: String, mode: SaveMode): Unit = {
+ val rows = recordsToStrings(records).asScala.toList
+ val df = spark.read.json(spark.sparkContext.parallelize(rows, 2))
+ df.write.format("org.apache.hudi")
+ .options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, operation)
+ .mode(mode)
+ .save(basePath)
+ }
+
+ private def incrementalRead(pathGlob: Option[String]): DataFrame = {
+ var reader = spark.read.format("org.apache.hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, "000")
+ pathGlob.foreach(g => reader =
reader.option(DataSourceReadOptions.INCR_PATH_GLOB.key, g))
+ reader.load(basePath)
+ }
+
+ private def assertPartitionCounts(df: DataFrame, expected: Map[String,
Int]): Unit = {
+ val actual = df.groupBy("_hoodie_partition_path").count().collect()
+ .map(row => row.getString(0) -> row.getLong(1).toInt).toMap
+ assertEquals(expected, actual, "incremental path glob returned unexpected
per-partition counts")
+ }
+}
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 ffaae686eb4d..ff0ce57d2b87 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
@@ -296,27 +296,30 @@ 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.
+ * Exercises the legacy incremental streaming path in
[[HoodieStreamSourceV1]] (table version
+ * below EIGHT) and [[HoodieStreamSourceV2]] (table version EIGHT and
above), taken when the file
+ * group reader is disabled. This drives the standalone incremental relations
+ * ([[IncrementalRelationV1]] / [[MergeOnReadIncrementalRelationV1]] for
version 6,
+ * [[IncrementalRelationV2]] / [[MergeOnReadIncrementalRelationV2]] for
version 8) via `getBatch`,
+ * rather than the newer HadoopFsRelation factory path.
*/
- private def testLegacyIncrementalStreamSource(tableType: HoodieTableType):
Unit = {
+ private def testLegacyIncrementalStreamSource(tableType: HoodieTableType,
+ tableVersion:
HoodieTableVersion): Unit = {
withTempDir { inputDir =>
- val tablePath =
s"${inputDir.getCanonicalPath}/test_${tableType.name}_legacy_stream"
+ val tablePath =
s"${inputDir.getCanonicalPath}/test_${tableType.name}_v${tableVersion.versionCode}_legacy_stream"
HoodieTableMetaClient.newTableBuilder()
.setTableType(tableType)
.setTableName(getTableName(tablePath))
- .setTableVersion(HoodieTableVersion.SIX)
+ .setTableVersion(tableVersion)
.setRecordKeyFields("id")
.setOrderingFields("ts")
.initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()),
tablePath)
- addData(tablePath, Seq(("1", "a1", "10", "000")), tableVersion =
HoodieTableVersion.SIX)
+ addData(tablePath, Seq(("1", "a1", "10", "000")), tableVersion =
tableVersion)
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)
+ .option(WRITE_TABLE_VERSION.key, tableVersion.versionCode().toString)
+ .option(STREAMING_READ_TABLE_VERSION.key,
tableVersion.versionCode().toString)
// force the legacy (non file-group-reader) incremental relation path
.option(HoodieReaderConfig.FILE_GROUP_READER_ENABLED.key, "false")
.load(tablePath)
@@ -330,16 +333,31 @@ class TestStreamingSource extends StreamTest {
addDataToQuery(tablePath,
Seq(("2", "a2", "12", "000"),
("3", "a3", "12", "000")),
- tableVersion = HoodieTableVersion.SIX),
+ tableVersion = tableVersion),
StartStream(),
AssertOnQuery { q => q.processAllAvailable(); true },
+ // The legacy branch of getBatch materializes the micro batch from an
RDD via
+ // internalCreateDataFrame, so the physical plan is a "Scan
ExistingRDD"; this fails if the
+ // legacy branch is dropped and the source silently falls back to the
file group reader
+ // path, which scans a HadoopFsRelation ("FileScan" / "Scan parquet")
instead.
+ AssertOnQuery { q =>
+ val plan = q.lastExecution.executedPlan.toString
+ assertTrue(plan.contains("Scan ExistingRDD"),
+ "expected the legacy RDD-backed incremental batch, but got plan: "
+ plan)
+ assertTrue(!plan.contains("FileScan"),
+ "expected no file-group-reader HadoopFsRelation scan, but got
plan: " + plan)
+ 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),
+ // inline clustering on this write lands a replacecommit inside the
next getBatch span, which
+ // reaches the replaced-file-group filtering of the legacy incremental
relations
+ addDataToQuery(tablePath, Seq(("4", "a4", "13", "000")),
enableInlineCluster = true,
+ tableVersion = tableVersion),
StartStream(),
AssertOnQuery { q => q.processAllAvailable(); true },
CheckAnswerRows(Seq(Row("4", "a4", "13", "000")), lastOnly = true,
isSorted = false)
@@ -348,11 +366,19 @@ class TestStreamingSource extends StreamTest {
}
test("test cow stream source with legacy file group reader disabled") {
- testLegacyIncrementalStreamSource(COPY_ON_WRITE)
+ testLegacyIncrementalStreamSource(COPY_ON_WRITE, HoodieTableVersion.SIX)
}
test("test mor stream source with legacy file group reader disabled") {
- testLegacyIncrementalStreamSource(MERGE_ON_READ)
+ testLegacyIncrementalStreamSource(MERGE_ON_READ, HoodieTableVersion.SIX)
+ }
+
+ test("test cow stream source with legacy file group reader disabled on table
version 8") {
+ testLegacyIncrementalStreamSource(COPY_ON_WRITE, HoodieTableVersion.EIGHT)
+ }
+
+ test("test mor stream source with legacy file group reader disabled on table
version 8") {
+ testLegacyIncrementalStreamSource(MERGE_ON_READ, HoodieTableVersion.EIGHT)
}
private def testCheckpointTranslation(tableName: String,
@@ -475,9 +501,10 @@ class TestStreamingSource extends StreamTest {
private def addDataToQuery(inputPath: String,
rows: Seq[(String, String, String, String)],
+ enableInlineCluster: Boolean = false,
tableVersion: HoodieTableVersion =
HoodieTableVersion.current): AssertOnQuery = {
AssertOnQuery { _=>
- addData(inputPath, rows, tableVersion = tableVersion)
+ addData(inputPath, rows, enableInlineCluster = enableInlineCluster,
tableVersion = tableVersion)
true
}
}
