voonhous commented on code in PR #19404:
URL: https://github.com/apache/hudi/pull/19404#discussion_r3689921552


##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.table.HoodieTableConfig
+import 
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{recordsToStrings, 
DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH}
+import org.apache.hudi.config.HoodieWriteConfig
+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
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.CsvSource
+
+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), so the
+ * COW and MOR cases together exercise {@code filterFileSlices} with {@code 
includeLogFiles} both
+ * off and on, and the table version selects V1 vs V2 in {@code DefaultSource}.
+ *
+ * Existing glob coverage ({@code TestCOWDataSourceStorage}) 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.
+ */
+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
+
+  @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 opts = 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")
+
+    // 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.
+    writeBatch("001", numFirst, firstPartition, opts, SaveMode.Overwrite)
+    writeBatch("002", numSecond, secondPartition, opts, SaveMode.Append)

Review Comment:
   Addressed. The MOR arms now write with the in-memory index 
(`canIndexLogFiles = true`), so both insert commits land in log-only file 
slices -- verified via `inputFiles`: zero base files on the MOR arms. Also 
added a third UPSERT commit updating 3 first-partition records, asserting 
exactly 3 rows carry `rider-003` under the `/2016/*/*/*` glob, plus a guard 
that MOR `inputFiles` are all `.log.` so a future index-default change cannot 
silently drop this coverage. Scaladoc claim rewritten to match.



##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -330,7 +333,7 @@ class TestStreamingSource extends StreamTest {
         addDataToQuery(tablePath,

Review Comment:
   Done. `addDataToQuery` now threads `enableInlineCluster`, and the last write 
in `testLegacyIncrementalStreamSource` clusters inline. Verified the 
replacecommit lands inside the final `getBatch` span on all four arms 
(`commitsToReturn` logs show it), so the replaced-file filtering executes; rows 
1-3 do not reappear and `CheckAnswerRows` is unchanged.



##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.table.HoodieTableConfig
+import 
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{recordsToStrings, 
DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH}
+import org.apache.hudi.config.HoodieWriteConfig
+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
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.CsvSource
+
+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), so the
+ * COW and MOR cases together exercise {@code filterFileSlices} with {@code 
includeLogFiles} both
+ * off and on, and the table version selects V1 vs V2 in {@code DefaultSource}.
+ *
+ * Existing glob coverage ({@code TestCOWDataSourceStorage}) 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.
+ */
+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
+
+  @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 = {

Review Comment:
   Added `testDefaultFallbackBehaviorAcrossTableVersions` to 
`TestIncrementalReadWithFullTableScan`, where the clean/archival machinery and 
the exception helper already live. Two corrections to the comment above, found 
while implementing:
   - The v6 failure mode is a `SparkException` wrapping `FileNotFoundException` 
at execution time, not `HoodieIncrementalPathNotFoundException` -- that 
exception only exists in the standalone `IncrementalRelationV1/V2`, which the 
batch path never constructs.
   - A span starting at `000` does **not** discriminate: with an archived 
start, V1 keeps only the latest merged file slice per file group and the record 
filter degenerates to `> "000"`, so both versions return everything. The 
discriminating query is a narrow span over cleaned-but-active commits; the test 
pins both cases (v6 throws, v8 silently full-scans) and documents why.
   
   Also, `INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN_FOR_NON_EXISTING_FILES` is a 
literal alias of the same `ConfigProperty` (same key, same default), which the 
test comment now notes.



##########
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`,

Review Comment:
   Deferred to #19443 (item 1) -- extending `TestLegacyParquetReadPath` with 
the fallback-after-clean variant is out of scope for this PR.



##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -348,11 +351,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") {

Review Comment:
   Done. Added an `AssertOnQuery` requiring `Scan ExistingRDD` and forbidding 
`FileScan` in the executed plan (verified that is the actual plan shape on all 
four legacy arms), so these tests fail if the legacy branch is removed and the 
source silently falls back to the FGR path.



##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.table.HoodieTableConfig
+import 
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{recordsToStrings, 
DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH}
+import org.apache.hudi.config.HoodieWriteConfig
+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
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.CsvSource
+
+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), so the
+ * COW and MOR cases together exercise {@code filterFileSlices} with {@code 
includeLogFiles} both
+ * off and on, and the table version selects V1 vs V2 in {@code DefaultSource}.
+ *
+ * Existing glob coverage ({@code TestCOWDataSourceStorage}) 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.
+ */
+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
+
+  @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",

Review Comment:
   Kept as an explicit v8 pin per the note; with the MOR arms now on log-only 
slices the matrix rows are no longer pairwise redundant.



##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.table.HoodieTableConfig
+import 
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{recordsToStrings, 
DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH}
+import org.apache.hudi.config.HoodieWriteConfig
+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
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.CsvSource
+
+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), so the
+ * COW and MOR cases together exercise {@code filterFileSlices} with {@code 
includeLogFiles} both
+ * off and on, and the table version selects V1 vs V2 in {@code DefaultSource}.
+ *
+ * Existing glob coverage ({@code TestCOWDataSourceStorage}) 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.
+ */
+class TestIncrementalReadWithPathGlob extends HoodieSparkClientTestBase {

Review Comment:
   Keeping `HoodieSparkClientTestBase` for consistency with the sibling 
incremental-read classes (`TestIncrementalReadWithFullTableScan`, 
`TestIncrementalReadByStateTransitionTime`); not worth diverging from the 
family to save 3 context restarts.



##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.table.HoodieTableConfig
+import 
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{recordsToStrings, 
DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH}
+import org.apache.hudi.config.HoodieWriteConfig
+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
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.CsvSource
+
+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), so the
+ * COW and MOR cases together exercise {@code filterFileSlices} with {@code 
includeLogFiles} both
+ * off and on, and the table version selects V1 vs V2 in {@code DefaultSource}.
+ *
+ * Existing glob coverage ({@code TestCOWDataSourceStorage}) 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.
+ */
+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
+
+  @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 opts = 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")
+
+    // 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.
+    writeBatch("001", numFirst, firstPartition, opts, SaveMode.Overwrite)
+    writeBatch("002", numSecond, secondPartition, opts, SaveMode.Append)
+
+    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.
+    assertPartitionCounts(incrementalRead(None), Map(firstPartition -> 
numFirst, secondPartition -> numSecond))
+
+    // Glob restricted to the first partition returns only its records.
+    assertPartitionCounts(incrementalRead(Some("/2016/*/*/*")), 
Map(firstPartition -> numFirst))
+
+    // Glob restricted to the second partition returns only its records.
+    assertPartitionCounts(incrementalRead(Some("/2015/*/*/*")), 
Map(secondPartition -> numSecond))
+
+    // Glob that matches no partition exercises the empty-result branch of the 
relation.
+    assertEquals(0, incrementalRead(Some("/9999/*/*/*")).count(),
+      "a glob matching no partition path must yield an empty result")
+  }
+
+  private def writeBatch(instant: String, n: Int, partition: String,

Review Comment:
   Renamed to `seed`; the helper scaladoc now states it only feeds field values 
and does not control commit times.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to