You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "yihua (via GitHub)" <gi...@apache.org> on 2026/07/29 06:21:15 UTC
[PR] test(spark): add incremental read-path relation coverage [hudi]
yihua opened a new pull request, #19404:
URL: https://github.com/apache/hudi/pull/19404
### Describe the issue this Pull Request addresses
The Spark incremental read-path relations are under-covered. The four relations in `hudi-spark-datasource/hudi-spark-common` sit well below the rest of the module: `IncrementalRelationV2` around 51%, `IncrementalRelationV1` around 47%, `MergeOnReadIncrementalRelationV2` around 62%, and `MergeOnReadIncrementalRelationV1` around 74%. Two branches in particular had no coverage: the `hoodie.datasource.read.incr.path.glob` file filtering used by the batch incremental relations, and the version-8 legacy streaming path that drives `IncrementalRelationV2`.
### Summary and Changelog
This is a test-only change; no production code is modified.
- Added `TestIncrementalReadWithPathGlob` (functional). It writes one partition per commit, then runs incremental queries with `incr.path.glob` restricting to each partition and to a partition that does not exist, asserting the exact per-partition row counts. It is parametrized over table type (COW, MOR) and table version (6, 8). Both the COW and MOR batch incremental read routes build a `HoodieIncrementalFileIndex` backed by `MergeOnReadIncrementalRelationV1` (version 6) / `MergeOnReadIncrementalRelationV2` (version 8), so these cases pin `filterFileSlices` and its empty-result branch for both relations, with `includeLogFiles` both off (COW) and on (MOR).
- Extended `TestStreamingSource`. The existing legacy-streaming test (file group reader disabled) was version 6 only, which reaches `IncrementalRelationV1` / `MergeOnReadIncrementalRelationV1` but never the V2 relations. Parametrized it by table version and added version-8 cases, so the version-8 legacy streaming path through `HoodieStreamSourceV2` drives `IncrementalRelationV2` (COW) and `MergeOnReadIncrementalRelationV2` (MOR).
Deliberate overlap and how duplication was avoided: `TestCOWDataSourceStorage` already exercises `incr.path.glob` for COW at the default table version and asserts total counts only. The new glob test does not reproduce that. It extends the glob coverage to table version 6 and to MOR, and pins the per-partition result (the partition of every returned row) plus the glob-matches-nothing empty branch, so it fails if the file filtering is wrong rather than merely non-empty. The streaming change folds new version-8 cases into the existing suite and helper rather than adding a near-duplicate class.
### Impact
None. Tests only.
### Risk Level
none
### Documentation Update
none
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Enough context is provided in the sections above
- [ ] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5114058841
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7226c74a7b7a8c420fb17f89b62a9897e9401057 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5140119809
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
}, {
"hash" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882",
"triggerID" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c1a9787a52e6bebcd94f05e4602542609f379d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5140157221
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
}, {
"hash" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882",
"triggerID" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"triggerType" : "PUSH"
}, {
"hash" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c1a9787a52e6bebcd94f05e4602542609f379d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882)
* 60db2bf80e3b616a995e05541f0bcff4cbb28f34 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5114116128
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/19404?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 72.50%. Comparing base ([`8e18999`](https://app.codecov.io/gh/apache/hudi/commit/8e18999a41645d4ae2fb240361615ae2430880c9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`7226c74`](https://app.codecov.io/gh/apache/hudi/commit/7226c74a7b7a8c420fb17f89b62a9897e9401057?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19404 +/- ##
=========================================
Coverage 72.49% 72.50%
- Complexity 32888 32889 +1
=========================================
Files 2574 2574
Lines 149010 149010
Branches 18749 18749
=========================================
+ Hits 108032 108034 +2
+ Misses 32513 32512 -1
+ Partials 8465 8464 -1
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.24% <ø> (ø)` | |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.79% <ø> (+<0.01%)` | :arrow_up: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.62% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.67% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `71.22% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `15.32% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `63.49% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.62% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.57% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.44% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `64.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/19404/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.96% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/19404/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/19404/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.39% <ø> (+<0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
[see 1 file with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/19404/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5142252996
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
}, {
"hash" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882",
"triggerID" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"triggerType" : "PUSH"
}, {
"hash" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15895",
"triggerID" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"triggerType" : "PUSH"
}, {
"hash" : "bbb3ac31c681f508b69ae71c118ec91476aadc72",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15911",
"triggerID" : "bbb3ac31c681f508b69ae71c118ec91476aadc72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c1a9787a52e6bebcd94f05e4602542609f379d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882)
* 60db2bf80e3b616a995e05541f0bcff4cbb28f34 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15895)
* bbb3ac31c681f508b69ae71c118ec91476aadc72 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15911)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous merged PR #19404:
URL: https://github.com/apache/hudi/pull/19404
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5139899245
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
}, {
"hash" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7226c74a7b7a8c420fb17f89b62a9897e9401057 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810)
* 1c1a9787a52e6bebcd94f05e4602542609f379d7 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5140403085
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
}, {
"hash" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882",
"triggerID" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"triggerType" : "PUSH"
}, {
"hash" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15895",
"triggerID" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c1a9787a52e6bebcd94f05e4602542609f379d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882)
* 60db2bf80e3b616a995e05541f0bcff4cbb28f34 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15895)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #19404:
URL: https://github.com/apache/hudi/pull/19404#discussion_r3685732050
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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, assertTrue}
+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,
+ opts: Map[String, String], mode: SaveMode): Unit = {
+ val records = recordsToStrings(dataGen.generateInsertsForPartition(instant, n, partition)).asScala.toList
+ val df = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ df.write.format("org.apache.hudi")
+ .options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .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")
+ assertTrue(actual.values.sum == expected.values.sum)
+ }
+}
Review Comment:
🤖 nit: this `assertTrue` is always implied by the `assertEquals(expected, actual)` map comparison on the line above — if the maps are equal their sums are trivially equal too, so this can never catch a bug the first assertion doesn't. Could you drop it to keep the helper tight?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #19404:
URL: https://github.com/apache/hudi/pull/19404#discussion_r3688604999
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPathGlob.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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, assertTrue}
+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,
+ opts: Map[String, String], mode: SaveMode): Unit = {
+ val records = recordsToStrings(dataGen.generateInsertsForPartition(instant, n, partition)).asScala.toList
+ val df = spark.read.json(spark.sparkContext.parallelize(records, 2))
+ df.write.format("org.apache.hudi")
+ .options(opts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .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")
+ assertTrue(actual.values.sum == expected.values.sum)
+ }
+}
Review Comment:
Done. Removed the redundant assertTrue and its now-unused import.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5114019786
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7226c74a7b7a8c420fb17f89b62a9897e9401057 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19404:
URL: https://github.com/apache/hudi/pull/19404#discussion_r3689922503
##########
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
Review Comment:
Applied.
##########
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(),
Review Comment:
Added `inputFiles` assertions on all four glob cases: the empty glob asserts zero files, the partition globs assert every reported path lives under the matched partition. On the MOR arms the slices are log-only now, so the `includeLogFiles` branch of `inputFiles` is exercised too. The `Array.empty` fall-through in `HoodieIncrementalFileIndex.inputFiles` is left for a separate minor PR since this one is test-only.
##########
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
Review Comment:
Deferred to #19443 (item 2); agreed the streaming suite is the wrong home for it since its table is unpartitioned.
##########
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") {
+ testLegacyIncrementalStreamSource(COPY_ON_WRITE, HoodieTableVersion.EIGHT)
Review Comment:
Deferred to #19443 (item 3).
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5142212420
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
}, {
"hash" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882",
"triggerID" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"triggerType" : "PUSH"
}, {
"hash" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15895",
"triggerID" : "60db2bf80e3b616a995e05541f0bcff4cbb28f34",
"triggerType" : "PUSH"
}, {
"hash" : "bbb3ac31c681f508b69ae71c118ec91476aadc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bbb3ac31c681f508b69ae71c118ec91476aadc72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c1a9787a52e6bebcd94f05e4602542609f379d7 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882)
* 60db2bf80e3b616a995e05541f0bcff4cbb28f34 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15895)
* bbb3ac31c681f508b69ae71c118ec91476aadc72 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5139937344
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
}, {
"hash" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882",
"triggerID" : "1c1a9787a52e6bebcd94f05e4602542609f379d7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7226c74a7b7a8c420fb17f89b62a9897e9401057 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810)
* 1c1a9787a52e6bebcd94f05e4602542609f379d7 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15882)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19404:
URL: https://github.com/apache/hudi/pull/19404#discussion_r3689623283
##########
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:
**major**: the MOR arms never write a log file, so the `includeLogFiles` "on" half of the class-comment claim is vacuous and the MOR cases execute the identical path as COW.
Both commits are INSERTs into brand-new partitions; the default SIMPLE index has `canIndexLogFiles() == false`, so `BaseSparkDeltaCommitActionExecutor` routes inserts to base files -- a local run of this test produces 8 parquet files and zero data log files. That leaves the log-file fallback in `filterFileSlices` (`MergeOnReadIncrementalRelationV1.scala:172`, `V2.scala:166`) unreached, which is the historically fragile half: the HUDI-9540 data-loss fix (859a7ee4fb61, #13786) lives exactly there, and its regression test forces log-only slices via the `INMEMORY` index for this reason.
Please either add a third commit that upserts a subset of the first-partition keys (or set `HoodieIndexConfig.INDEX_TYPE -> INMEMORY` for the MOR arms, as `TestMORDataSource.testIncrementalQueryMORWithCompactionAndClean` does) and assert a merged column value under the `/2016/*/*/*` glob -- a count alone would not discriminate -- or narrow the class-comment claim to what is actually exercised.
##########
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:
minor: both new v8 cases are line-coverage only as written -- the same assertions pass on the FGR-enabled path, so deleting the `if (enableFileGroupReader)` branch in `HoodieStreamSourceV2.scala:148` would leave them green. The legacy branch materializes the batch via `internalCreateDataFrame` (a `LogicalRDD`) while the FGR branch scans a `HadoopFsRelation`, so the executed plan discriminates cheaply. Suggest one `AssertOnQuery` on `q.lastExecution.executedPlan` (e.g. contains "ExistingRDD") inside `testLegacyIncrementalStreamSource` -- it strengthens the v6 arms too. Otherwise, note in the comment that these cases are coverage-only.
##########
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:
**major, cheap to close here**: the replaced-file filtering of the legacy relations stays unreached after this PR, and this helper is one parameter away from covering it.
`IncrementalRelationV1.scala:164-173` / `IncrementalRelationV2.scala:149-158` filter out file groups replaced by clustering/insert_overwrite; the history is real: HUDI-2058 (2cecb751879c) introduced the filter and HUDI-9672 (17f3720a48c0) fixed incremental data duplication from skipped clustering. The existing "streaming source with clustering" test runs FGR-enabled, so it never reaches these lines.
`addData` already takes `enableInlineCluster` (line 470) but `addDataToQuery` does not thread it. Suggest: add `enableInlineCluster: Boolean = false` to `addDataToQuery`, pass it through, and set it on this call -- same `CheckAnswerRows`, near-zero extra runtime, and both the v6 and the new v8 arms then reach the replaced-file branch.
##########
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:
Follow-up suggestion: the fallback/full-table-scan blocks of these two standalone relations (`IncrementalRelationV1.scala:235-272`, `IncrementalRelationV2.scala:225-259`, including the parallelized file-existence probe from #10480) are reachable by no test in the repo, and they are most of the missing 47%/51% coverage this PR's description cites. The cheapest close is not via streaming: `TestLegacyParquetReadPath.testCowIncrementalReadEqualsFileGroupReader` already constructs both relations directly (lines 317-319); a variant that sets `INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN=true` after aggressive cleaning and asserts both still equal the file-group-reader result would cover it in ~15 lines. Fine to defer -- if so, please file the follow-up issue.
##########
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:
nit, feel free to ignore: `HoodieSparkClientTestBase` starts and stops a fresh `SparkContext` per parameterized case (4x here), while the sibling glob test `TestCOWDataSourceStorage` extends `SparkClientFunctionalTestHarness`, which shares one session across the class. The incremental-read family (`TestIncrementalReadWithFullTableScan`, `TestIncrementalReadByStateTransitionTime`) does use this base, so this is convention-consistent -- switching harness would just save 3 context restarts of CI time.
##########
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(),
Review Comment:
minor: consider also asserting `inputFiles` in these four checks, e.g. `assertEquals(0, incrementalRead(Some("/9999/*/*/*")).inputFiles.length)` here and the expected file counts above. `HoodieIncrementalFileIndex.inputFiles` has zero test coverage today and carries a suspicious fall-through (`if (fileSlices.isEmpty) { Array.empty }` discards its result, lines 58-60 -- benign only because the flatMap of an empty map is empty), and this file index diverging from the main one was a real bug class (HUDI-9621, #13593). One extra assertion pins the glob at file granularity, which is stronger than row counts.
##########
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:
**major, fine as a named follow-up**: the v6/v8 axis introduced here has an untested behavioral divergence next door: the fallback-to-full-table-scan default. `MergeOnReadIncrementalRelationV1.scala:206` hardcodes `getOrElse(..., "false")`, while `V2.scala:194-195` uses the config default, flipped to `true` by HUDI-8624 (37152c6c779a); V1 also falls back when the end instant is archived, V2 only checks the start. So on a cleaned/archived span v6 throws `HoodieIncrementalPathNotFoundException` while v8 silently full-scans -- and no test exercises either default (`TestIncrementalReadWithFullTableScan` always sets the key explicitly and never sets a table version). This branch has 5 fix commits: HUDI-2711, HUDI-3189, HUDI-7003, HUDI-8602, HUDI-8624.
Since this class already has the `(tableType, tableVersion)` matrix, one extra case that cleans/archives commit 001 and asserts the per-version default behavior would pin it. If out of scope, please file a follow-up issue so the gap is tracked.
##########
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
Review Comment:
nit: there is a second pre-existing glob test worth naming here: `TestDataSourceForBootstrap` also asserts `INCR_PATH_GLOB` (bootstrapped COW, ~line 685).
```suggestion
* Existing glob coverage ({@code TestCOWDataSourceStorage}, plus {@code TestDataSourceForBootstrap}
* for bootstrapped COW) exercises only COW at the default table
```
##########
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:
nit: this cell adds no production coverage over master -- the default table version (10) already selects the same V2 factory chain, so `TestCOWDataSourceStorage`'s existing glob test drives `MergeOnReadIncrementalRelationV2.filterFileSlices` too, and the relations have no v8-vs-v10 branching (`filterFileSlices` in V1 and V2 are byte-identical). Keeping it as an explicit v8 pin costs only ~1.4s, so feel free to leave it; the genuinely new cells are v6, MOR, and the empty-glob branch.
##########
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:
nit: `instant` does not control the commit instant -- `generateInsertsForPartition`'s first argument only seeds field values (`rider-001`, `driver-001`; `HoodieTestDataGenerator.java:459,468`), and the real commit times are auto-generated (which is also why `START_COMMIT "000"` works: it sorts below any real instant, not because the commits are named 001/002). Please rename the parameter (e.g. `seed`) so readers do not assume commit-time control.
##########
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
Review Comment:
nit: this covers the file-slice copy of the glob only. The standalone COW relations carry a second, independently-written implementation of the same option over `regularFileIdToFullPath`/`metaBootstrapFileIdToFullPath` (`IncrementalRelationV1.scala:196-207`, `V2.scala:181-192`, including a meta-bootstrap arm the MOR copy lacks), reachable only from the legacy stream sources -- and no test sets `INCR_PATH_GLOB` on that path. The streaming suite's table is unpartitioned, so the natural home is the direct-construction test in `TestLegacyParquetReadPath` (lines 317-319). A follow-up issue would do.
##########
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") {
+ testLegacyIncrementalStreamSource(COPY_ON_WRITE, HoodieTableVersion.EIGHT)
Review Comment:
nit, feel free to ignore: adjacent gap while you are raising `IncrementalRelationV2` coverage -- `INCREMENTAL_READ_SCHEMA_USE_END_INSTANTTIME` is set by no test in the repo (grep hits only `DataSourceOptions.scala` and the two relations), so the `useEndInstantSchema=true` branch (`IncrementalRelationV1.scala:121-129`, `V2.scala:106-114`, from HUDI-1301/#2125) is uncovered. One extra option on the direct-construction test in `TestLegacyParquetReadPath` would cover ~10 lines per relation. Optional / follow-up.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(spark): add incremental read-path relation coverage [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19404:
URL: https://github.com/apache/hudi/pull/19404#issuecomment-5117607553
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810",
"triggerID" : "7226c74a7b7a8c420fb17f89b62a9897e9401057",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7226c74a7b7a8c420fb17f89b62a9897e9401057 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15810)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org