You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "KiteSoar (via GitHub)" <gi...@apache.org> on 2026/09/27 07:31:33 UTC
[PR] fix(tools): read archived clean plans and metadata [hudi]
KiteSoar opened a new pull request, #20099:
URL: https://github.com/apache/hudi/pull/20099
### Describe the issue this Pull Request addresses
Closes #19639.
The clean procedures read archived clean instants from `metaClient.getArchivedTimeline`, but the archived timeline does not have the cleaner plan or clean metadata payloads loaded by default. As a result, `show_clean_plans` returned null plan fields, while `show_cleans` and `show_cleans_metadata` could fail with an `IOException` when they tried to deserialize archived clean metadata.
### Summary and Changelog
Users can now query archived clean plans, clean summaries, and per-partition clean metadata through the three clean procedures.
- Load the required plan or metadata payloads for archived clean instants and expose them through a temporary timeline used by the procedures.
- Decode archived records for both timeline layout V1 and V2. V1 cleaner records are converted from their Avro representation, while V2 payloads are copied from their byte buffers.
- Apply the procedure limit before loading archived payloads for `show_clean_plans` and `show_cleans`. Preserve `show_cleans_metadata` row-limit behavior because one clean instant can produce multiple partition rows.
- Add coverage for archived clean plans and metadata on table versions 6 and 9.
Validation:
- `TestArchivedTimelineV1`: 24 tests passed.
- `TestShowCleansProcedures`: 11 tests passed.
### Impact
This fixes the existing `show_clean_plans`, `show_cleans`, and `show_cleans_metadata` behavior when `show_archived` is enabled. It does not change public APIs, storage formats, or configuration. Archived payload loading is bounded by `limit` where the procedure returns one row per clean instant.
### Risk Level
Low. The change is limited to Spark SQL clean procedures and uses the existing archived timeline loader. Tests cover both legacy and current timeline layouts.
### Documentation Update
None.
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] 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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172556523
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
+ }
- // Known limitation, see #19639: getCleanerPlans collects the archived clean instants but
- // then hands every instant to processCleanPlan against the ACTIVE timeline, so the
- // archived instant's .clean.requested file is not found, the read falls back to
- // createErrorRow and every plan field comes back null. Only plan_time/state/action
- // survive, because those are taken from the instant and not from the plan. A fix has to
- // read the archived instant's own content and would flip the null assertions below to
- // non-null; merely routing to getArchivedTimeline the way the sibling ShowCleansProcedure
- // does is not enough, since that path throws today (pinned above).
- assertResult("COMPLETED")(archivedPlan.getString(1))
- assertResult("clean")(archivedPlan.getString(2))
- planFields.foreach { name =>
- assert(archivedPlan.isNullAt(archivedPlan.fieldIndex(name)),
- s"archived clean plan is expected to return a null $name today (#19639)")
+ // This table is not partitioned, so all three procedures return one row per clean.
+ assertResult(6)(rowCount("show_cleans", showArchived = true))
+ assertResult(6)(rowCount("show_cleans_metadata", showArchived = true))
+ assertResult(6)(rowCount("show_clean_plans", showArchived = true))
+
+ val procedures = Seq("show_clean_plans", "show_cleans", "show_cleans_metadata")
+ val beforeArchive = procedures.map { procedure =>
+ procedure -> spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ }.toMap
+
+ spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
+ s" retain_commits => 1, enable_metadata => false, options => 'hoodie.write.table.version=$tableVersion')").collect()
+
+ // Precondition: archival must have split the two cleans across the two timelines,
+ // otherwise the archived branch below is never exercised and the test passes vacuously.
+ val metaClient = createMetaClient(spark, tablePath)
+ assertResult(tableVersion)(metaClient.getTableConfig.getTableVersion.versionCode())
+ val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline.filterCompletedInstants
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ assert(archivedCleans.nonEmpty, s"expected archived cleans, got $archivedCleans")
+ assert(activeCleans.nonEmpty, s"expected active cleans, got $activeCleans")
+ assert(archivedCleans.max.compareTo(activeCleans.min) < 0,
+ "archived cleans must be older than active cleans")
+
+ // The active-only view must exclude the archived clean for all three procedures.
+ assertResult(activeCleans.size)(rowCount("show_cleans", showArchived = false))
+ assertResult(activeCleans.size)(rowCount("show_cleans_metadata", showArchived = false))
+ assertResult(activeCleans.size)(rowCount("show_clean_plans", showArchived = false))
+
+ procedures.foreach { procedure =>
+ val allRows = spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ assertResult(beforeArchive(procedure))(allRows)
+ val limitedRows = spark.sql(s"call $procedure(table => '$tableName', showArchived => true, limit => 1)").collect().toSeq
+ assertResult(allRows.take(1))(limitedRows)
Review Comment:
done. Added limit = activeCleans.size + 1 assertions for all three procedures and verified that the last returned row is archivedCleans.max
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-5853935890
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20099?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 80.40%. Comparing base ([`8d3b065`](https://app.codecov.io/gh/apache/hudi/commit/8d3b06570c5810210dc87dfa7ad051477b6cc175?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`1f64c42`](https://app.codecov.io/gh/apache/hudi/commit/1f64c42e0a82ddca90027a8a3137afdf0b6d1120?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20099 +/- ##
============================================
- Coverage 80.41% 80.40% -0.01%
- Complexity 34869 34870 +1
============================================
Files 2546 2546
Lines 142888 142901 +13
Branches 17373 17378 +5
============================================
+ Hits 114900 114905 +5
- Misses 20074 20080 +6
- Partials 7914 7916 +2
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20099/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/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.96% <ø> (ø)` | |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.48% <ø> (+0.01%)` | :arrow_up: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.73% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.83% <ø> (-0.02%)` | :arrow_down: |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.14% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.05% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.98% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.99% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.21% <ø> (-0.32%)` | :arrow_down: |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.30% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20099/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20099/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/20099/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/20099/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.99% <ø> (+<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.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20099?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...i/command/procedures/ShowCleansPlanProcedure.scala](https://app.codecov.io/gh/apache/hudi/pull/20099?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2Fhudi%2Fcommand%2Fprocedures%2FShowCleansPlanProcedure.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9zcGFyay9zcWwvaHVkaS9jb21tYW5kL3Byb2NlZHVyZXMvU2hvd0NsZWFuc1BsYW5Qcm9jZWR1cmUuc2NhbGE=) | `89.90% <ø> (-0.93%)` | :arrow_down: |
| [.../hudi/command/procedures/ShowCleansProcedure.scala](https://app.codecov.io/gh/apache/hudi/pull/20099?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2Fhudi%2Fcommand%2Fprocedures%2FShowCleansProcedure.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9zcGFyay9zcWwvaHVkaS9jb21tYW5kL3Byb2NlZHVyZXMvU2hvd0NsZWFuc1Byb2NlZHVyZS5zY2FsYQ==) | `94.11% <ø> (-0.12%)` | :arrow_down: |
... and [2 files with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20099/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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-5967956675
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308",
"triggerID" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"triggerType" : "PUSH"
}, {
"hash" : "b231fe4c56126d0eba3921ff5eedef79694d6e30",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17438",
"triggerID" : "b231fe4c56126d0eba3921ff5eedef79694d6e30",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f64c42e0a82ddca90027a8a3137afdf0b6d1120 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308)
* b231fe4c56126d0eba3921ff5eedef79694d6e30 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17438)
<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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4408105232
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ArchivedCleanTimelineUtils.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.common.table.timeline;
+
+import org.apache.hudi.common.avro.HoodieAvroUtils;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.ACTION_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.METADATA_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.PLAN_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
+/** Utilities for reading archived cleaner plans and metadata in either timeline layout. */
+public class ArchivedCleanTimelineUtils {
+
+ private ArchivedCleanTimelineUtils() {
+ }
+
+ /**
+ * Creates a timeline for the selected completed cleans, supplied in descending requested-time order.
+ * Payloads are loaded lazily in windows of at most {@code batchSize} selected instants; only the
+ * current window is retained. This lets callers stop after enough partition rows, including when
+ * some cleans have empty partition metadata, without loading the entire archived history.
+ */
+ public static HoodieTimeline getTimeline(HoodieTableMetaClient metaClient, List<HoodieInstant> selectedInstants,
+ boolean loadPlans, int batchSize) {
+ checkArgument(batchSize > 0, "Batch size must be positive");
+ List<HoodieInstant> instants = new ArrayList<>(selectedInstants);
+ TimelineFactory factory = metaClient.getTableFormat().getTimelineFactory();
+ Map<String, Integer> positions = new HashMap<>();
+ for (int i = 0; i < instants.size(); i++) {
+ positions.put(instants.get(i).requestedTime(), i);
+ }
+ boolean legacy = metaClient.getTimelineLayoutVersion().getVersion() < TimelineLayoutVersion.VERSION_2;
+ String actionField = legacy ? "actionType" : ACTION_ARCHIVED_META_FIELD;
+ String contentField = legacy ? (loadPlans ? "hoodieCleanerPlan" : "hoodieCleanMetadata")
+ : (loadPlans ? PLAN_ARCHIVED_META_FIELD : METADATA_ARCHIVED_META_FIELD);
+ HoodieArchivedTimeline.LoadMode loadMode = loadPlans ? HoodieArchivedTimeline.LoadMode.PLAN : HoodieArchivedTimeline.LoadMode.METADATA;
+ HoodieInstantReader reader = new HoodieInstantReader() {
+ private final Map<String, byte[]> contents = new ConcurrentHashMap<>();
+
+ @Override
+ public synchronized InputStream getContentStream(HoodieInstant instant) {
+ String instantTime = instant.requestedTime();
+ if (!contents.containsKey(instantTime)) {
+ Integer position = positions.get(instantTime);
+ checkArgument(position != null, "Clean instant was not selected: " + instantTime);
+ List<HoodieInstant> batch = instants.subList(position, position + Math.min(batchSize, instants.size() - position));
+ Set<String> batchTimes = new HashSet<>(batch.stream().map(HoodieInstant::requestedTime).collect(Collectors.toList()));
+ HoodieArchivedTimeline.TimeRangeFilter range = new HoodieArchivedTimeline.ClosedClosedTimeRangeFilter(
+ batch.get(batch.size() - 1).requestedTime(), batch.get(0).requestedTime());
+ contents.clear();
+ factory.createArchivedTimelineLoader().loadInstants(metaClient, legacy ? null : range, loadMode,
Review Comment:
[P2] Avoid repeatedly scanning the same V2 archive file across batches. In addition to the V1 rescan concern already discussed, V2's getFilteredFiles only prunes whole files by time range. loadInstants then iterates every record in each overlapping file and applies the range check after iterator.next(). Consequently, if a compacted file contains 320 selected cleans, consuming them in the callers' 32-instant windows reopens and traverses that file 10 times. With show_cleans_metadata(limit => 1) and many empty cleans, the one-instant windows can make this substantially worse. The mocked fixture.loaded assertions count accepted payloads and do not detect these repeated physical reads. Could we follow up with file-aware iteration/caching or reader-level range pruning, and a test that counts file opens or records visited across multiple windows in one V2 archive file? This is a performance concern inferred from the loader, not a measured benchmark.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +249,37 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getCleanTimeline(metaClient: HoodieTableMetaClient,
+ showArchived: Boolean,
+ loadPlans: Boolean,
+ instantLimit: Int,
+ batchSize: Int): HoodieTimeline = {
+ val activeTimeline = metaClient.getActiveTimeline
+ val activeCleans = if (loadPlans) activeTimeline.getCleanerTimeline
+ else activeTimeline.getCleanerTimeline.filterCompletedInstants
+ val activeInstants = activeCleans.getInstants.asScala.toSeq
+ val archivedInstants = if (showArchived) {
+ metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants.getInstants.asScala.toSeq
+ } else {
+ Seq.empty[HoodieInstant]
+ }
+ val comparator = TimelineLayout.fromVersion(metaClient.getTimelineLayoutVersion)
+ .getInstantComparator.requestedTimeOrderedComparator.reversed()
+ val selected = (activeInstants.map(_ -> true) ++ archivedInstants.map(_ -> false))
+ .sortWith((a, b) => comparator.compare(a._1, b._1) < 0).take(instantLimit)
+ val selectedArchived = selected.collect { case (instant, false) => instant }
+ val archivedTimeline = ArchivedCleanTimelineUtils.getTimeline(metaClient, selectedArchived.asJava,
+ loadPlans, batchSize)
+ val activeTimes = selected.collect { case (instant, true) => instant.requestedTime() }.toSet
+ metaClient.getTableFormat.getTimelineFactory.createDefaultTimeline(
+ selected.map(_._1).reverse.asJava.stream(), new HoodieInstantReader {
Review Comment:
[P3] Simplify the repeated ordering and collection conversions. selected is already sorted descending, and this line stores it ascending, but both consumers then sort again (getSortedCleans for metadata and getReverseOrderedInstants for plans); the archived helper also sorts its selected subset. In particular, the partition-metadata path keeps all instant descriptors, so it repeats that work over the full clean history even for a small row limit. Could the row-building code consume a single selected descending sequence (or reverse the known ascending timeline list), retaining this timeline only for payload lookup? That would also let getSortedCleans shed its Java list -> array -> Scala list -> Java list -> ArrayList conversion chain. Non-blocking simplification.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172552981
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
+ val cleanInstants = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getReverseOrderedInstants.iterator().asScala.take(limit).toList
+ val contents = new ConcurrentHashMap[String, Array[Byte]]()
+ val factory = metaClient.getTableFormat.getTimelineFactory
+ if (cleanInstants.nonEmpty) {
+ val legacy = metaClient.getTimelineLayoutVersion.getVersion < TimelineLayoutVersion.VERSION_2
+ val actionField = if (legacy) "actionType" else "action"
+ val contentField = if (legacy) {
+ if (loadPlans) "hoodieCleanerPlan" else "hoodieCleanMetadata"
+ } else {
+ if (loadPlans) "plan" else "metadata"
+ }
Review Comment:
Moved the format-specific decoding into ArchivedCleanTimelineUtils in hudi-common and reused the V2 action/plan/metadata field constants in b231fe4c5612. The helper is now available for CleansCommand to reuse. CLI integration is not included in this PR, so the CLI-specific V1 issue mentioned here remains outside this fix.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -165,10 +172,13 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
getCleans(metaClient.getActiveTimeline, limit)
}
val finalResults = if (showArchived) {
+ val archivedCleanLimit = if (includePartitionMetadata) Int.MaxValue else limit
Review Comment:
Done in b231fe4c5612. Int.MaxValue now applies only to instant descriptors; archived metadata payloads are loaded lazily in bounded batches while partition rows are needed. Empty cleans continue to the next batch. Tests cover empty and multi-row cleans and verify which payloads are loaded.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-5854999603
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308",
"triggerID" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f64c42e0a82ddca90027a8a3137afdf0b6d1120 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308)
<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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4118483702
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
+ val cleanInstants = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getReverseOrderedInstants.iterator().asScala.take(limit).toList
+ val contents = new ConcurrentHashMap[String, Array[Byte]]()
+ val factory = metaClient.getTableFormat.getTimelineFactory
+ if (cleanInstants.nonEmpty) {
+ val legacy = metaClient.getTimelineLayoutVersion.getVersion < TimelineLayoutVersion.VERSION_2
+ val actionField = if (legacy) "actionType" else "action"
+ val contentField = if (legacy) {
+ if (loadPlans) "hoodieCleanerPlan" else "hoodieCleanMetadata"
+ } else {
+ if (loadPlans) "plan" else "metadata"
+ }
Review Comment:
**minor:** not blocking, but this puts both archive formats' field names and payload decoding into a Spark procedure, and hudi-cli has the same bug on V1: `CLIUtils.java:57-60` loads details through `ArchivedTimelineV1.java:346`, which stores clean payloads as JSON that `readCleanMetadata` cannot parse. Would it make sense to host this helper in hudi-common (a static on `HoodieArchivedTimeline` or `TimelineUtils`) so `CleansCommand` can share it, and to reuse `ArchivedTimelineV2.ACTION_ARCHIVED_META_FIELD` / `PLAN_ARCHIVED_META_FIELD` / `METADATA_ARCHIVED_META_FIELD` instead of the literals?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -165,10 +172,13 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
getCleans(metaClient.getActiveTimeline, limit)
}
val finalResults = if (showArchived) {
+ val archivedCleanLimit = if (includePartitionMetadata) Int.MaxValue else limit
Review Comment:
**minor:** not blocking. `Int.MaxValue` loads the metadata of every archived clean regardless of `limit`. Many rows per instant argues for fewer instants, not more; the only case that needs more than `limit` instants is a 0-row empty clean, which requires `hoodie.write.empty.clean.interval.hours` > 0 (default -1, `HoodieCleanConfig.java:254`). Could we pass `limit` here, or extend in batches only while the row count is below `limit`?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
+ val cleanInstants = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getReverseOrderedInstants.iterator().asScala.take(limit).toList
+ val contents = new ConcurrentHashMap[String, Array[Byte]]()
+ val factory = metaClient.getTableFormat.getTimelineFactory
+ if (cleanInstants.nonEmpty) {
+ val legacy = metaClient.getTimelineLayoutVersion.getVersion < TimelineLayoutVersion.VERSION_2
+ val actionField = if (legacy) "actionType" else "action"
+ val contentField = if (legacy) {
+ if (loadPlans) "hoodieCleanerPlan" else "hoodieCleanMetadata"
+ } else {
+ if (loadPlans) "plan" else "metadata"
+ }
+ val loadMode = if (loadPlans) HoodieArchivedTimeline.LoadMode.PLAN else HoodieArchivedTimeline.LoadMode.METADATA
+ factory.createArchivedTimelineLoader().loadInstants(metaClient,
+ new HoodieArchivedTimeline.ClosedClosedTimeRangeFilter(
+ cleanInstants.last.requestedTime(), cleanInstants.head.requestedTime()),
+ loadMode,
+ record => HoodieTimeline.CLEAN_ACTION == record.get(actionField).toString,
Review Comment:
**blocker:** on V1 tables the clean-only `commitsFilter` trips the loader's per-file early exit, so older archived cleans load no payload and the #19639 symptoms return (IOException from `show_cleans` / `show_cleans_metadata`, null rows from `show_clean_plans`). Evidence: `ArchivedTimelineLoaderV1.java:139` applies the filter before counting in-range instants and `:157-163` breaks at the first file that yields 0 after one yielded > 0; every archival run writes a new `.commits_.archive.N` (`HoodieLogFormatWriter.java:98-99`) and cleans archive on their own threshold (`TimelineArchiverV1.java:205-213`), so a commit-only file between two clean-bearing ones is routine, and `show_cleans_metadata` (`Int.MaxValue`) always spans it. The new test archives once, so it only ever sees one file. Could we pass `null` as the range filter when `legacy` (the break is guarded by `filter != null`) and apply `isInRange(instantTime)` in the consumer, plus a V1 case with three archival runs and a commi
t-only middle file?
<details><summary>Why `record => true` is not enough</summary>
The exit assumes archive files are time-ordered, but cleans lag commits: a newer file can hold only commits above the range while an older file still holds in-range cleans, so counting every action still breaks early. Dropping the filter on V1 costs one extra full pass, which `getArchivedTimeline` already does today for V1.
</details>
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
+ }
- // Known limitation, see #19639: getCleanerPlans collects the archived clean instants but
- // then hands every instant to processCleanPlan against the ACTIVE timeline, so the
- // archived instant's .clean.requested file is not found, the read falls back to
- // createErrorRow and every plan field comes back null. Only plan_time/state/action
- // survive, because those are taken from the instant and not from the plan. A fix has to
- // read the archived instant's own content and would flip the null assertions below to
- // non-null; merely routing to getArchivedTimeline the way the sibling ShowCleansProcedure
- // does is not enough, since that path throws today (pinned above).
- assertResult("COMPLETED")(archivedPlan.getString(1))
- assertResult("clean")(archivedPlan.getString(2))
- planFields.foreach { name =>
- assert(archivedPlan.isNullAt(archivedPlan.fieldIndex(name)),
- s"archived clean plan is expected to return a null $name today (#19639)")
+ // This table is not partitioned, so all three procedures return one row per clean.
+ assertResult(6)(rowCount("show_cleans", showArchived = true))
+ assertResult(6)(rowCount("show_cleans_metadata", showArchived = true))
+ assertResult(6)(rowCount("show_clean_plans", showArchived = true))
+
+ val procedures = Seq("show_clean_plans", "show_cleans", "show_cleans_metadata")
+ val beforeArchive = procedures.map { procedure =>
+ procedure -> spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ }.toMap
+
+ spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
+ s" retain_commits => 1, enable_metadata => false, options => 'hoodie.write.table.version=$tableVersion')").collect()
+
+ // Precondition: archival must have split the two cleans across the two timelines,
+ // otherwise the archived branch below is never exercised and the test passes vacuously.
+ val metaClient = createMetaClient(spark, tablePath)
+ assertResult(tableVersion)(metaClient.getTableConfig.getTableVersion.versionCode())
+ val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline.filterCompletedInstants
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ assert(archivedCleans.nonEmpty, s"expected archived cleans, got $archivedCleans")
+ assert(activeCleans.nonEmpty, s"expected active cleans, got $activeCleans")
+ assert(archivedCleans.max.compareTo(activeCleans.min) < 0,
+ "archived cleans must be older than active cleans")
+
+ // The active-only view must exclude the archived clean for all three procedures.
+ assertResult(activeCleans.size)(rowCount("show_cleans", showArchived = false))
+ assertResult(activeCleans.size)(rowCount("show_cleans_metadata", showArchived = false))
+ assertResult(activeCleans.size)(rowCount("show_clean_plans", showArchived = false))
+
+ procedures.foreach { procedure =>
+ val allRows = spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ assertResult(beforeArchive(procedure))(allRows)
+ val limitedRows = spark.sql(s"call $procedure(table => '$tableName', showArchived => true, limit => 1)").collect().toSeq
+ assertResult(allRows.take(1))(limitedRows)
Review Comment:
**minor:** not blocking. Both procedures sort the merged rows descending before `take(limit)`, so `limit => 1` always returns the newest active clean and the archived `take(limit)` pushdown is never exercised; taking the wrong end would still pass. Could we use `limit => activeCleans.size + 1` and assert the last row is `archivedCleans.max`?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
Review Comment:
**nit:** feel free to ignore. Both callers pass `limit`, so this default is the one line codecov reports as uncovered.
```suggestion
limit: Int): HoodieTimeline = {
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
Review Comment:
**nit:** feel free to ignore. `HoodieTableVersion.current()` is TEN, so this drops the default version the replaced test ran on (9 and 10 share layout V2, so the branch coverage is the same). Needs `import org.apache.hudi.common.table.HoodieTableVersion`.
```suggestion
Seq(6, HoodieTableVersion.current().versionCode()).foreach { tableVersion =>
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
Review Comment:
**nit:** feel free to ignore. The table version is set three ways (`withSQLConf`, the tblproperty, and `options` here). `run_clean` already picks the tblproperty up through `HoodieCLIUtils.getWriteParameters` (`HoodieCLIUtils.scala:72-75`); only `archive_commits` builds its config from `options` alone. Could we drop the `withSQLConf` entry and this `options` argument, keeping the tblproperty plus the `archive_commits` `options`?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
+ }
- // Known limitation, see #19639: getCleanerPlans collects the archived clean instants but
- // then hands every instant to processCleanPlan against the ACTIVE timeline, so the
- // archived instant's .clean.requested file is not found, the read falls back to
- // createErrorRow and every plan field comes back null. Only plan_time/state/action
- // survive, because those are taken from the instant and not from the plan. A fix has to
- // read the archived instant's own content and would flip the null assertions below to
- // non-null; merely routing to getArchivedTimeline the way the sibling ShowCleansProcedure
- // does is not enough, since that path throws today (pinned above).
- assertResult("COMPLETED")(archivedPlan.getString(1))
- assertResult("clean")(archivedPlan.getString(2))
- planFields.foreach { name =>
- assert(archivedPlan.isNullAt(archivedPlan.fieldIndex(name)),
- s"archived clean plan is expected to return a null $name today (#19639)")
+ // This table is not partitioned, so all three procedures return one row per clean.
+ assertResult(6)(rowCount("show_cleans", showArchived = true))
+ assertResult(6)(rowCount("show_cleans_metadata", showArchived = true))
+ assertResult(6)(rowCount("show_clean_plans", showArchived = true))
+
+ val procedures = Seq("show_clean_plans", "show_cleans", "show_cleans_metadata")
+ val beforeArchive = procedures.map { procedure =>
+ procedure -> spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ }.toMap
+
+ spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
+ s" retain_commits => 1, enable_metadata => false, options => 'hoodie.write.table.version=$tableVersion')").collect()
+
+ // Precondition: archival must have split the two cleans across the two timelines,
Review Comment:
**nit:** feel free to ignore. The test now creates six cleans, so "the two cleans" is stale.
```suggestion
// Precondition: archival must have split the cleans across the active and archived timelines,
```
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172557922
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
Review Comment:
done，The test matrix now uses Seq(6, HoodieTableVersion.current().versionCode()), with the required 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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub) via commits" <co...@hudi.apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4439460915
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +249,37 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getCleanTimeline(metaClient: HoodieTableMetaClient,
+ showArchived: Boolean,
+ loadPlans: Boolean,
+ instantLimit: Int,
+ batchSize: Int): HoodieTimeline = {
+ val activeTimeline = metaClient.getActiveTimeline
+ val activeCleans = if (loadPlans) activeTimeline.getCleanerTimeline
+ else activeTimeline.getCleanerTimeline.filterCompletedInstants
+ val activeInstants = activeCleans.getInstants.asScala.toSeq
+ val archivedInstants = if (showArchived) {
+ metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants.getInstants.asScala.toSeq
+ } else {
+ Seq.empty[HoodieInstant]
+ }
+ val comparator = TimelineLayout.fromVersion(metaClient.getTimelineLayoutVersion)
+ .getInstantComparator.requestedTimeOrderedComparator.reversed()
+ val selected = (activeInstants.map(_ -> true) ++ archivedInstants.map(_ -> false))
+ .sortWith((a, b) => comparator.compare(a._1, b._1) < 0).take(instantLimit)
+ val selectedArchived = selected.collect { case (instant, false) => instant }
+ val archivedTimeline = ArchivedCleanTimelineUtils.getTimeline(metaClient, selectedArchived.asJava,
+ loadPlans, batchSize)
+ val activeTimes = selected.collect { case (instant, true) => instant.requestedTime() }.toSet
+ metaClient.getTableFormat.getTimelineFactory.createDefaultTimeline(
+ selected.map(_._1).reverse.asJava.stream(), new HoodieInstantReader {
Review Comment:
done
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ArchivedCleanTimelineUtils.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.common.table.timeline;
+
+import org.apache.hudi.common.avro.HoodieAvroUtils;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.ACTION_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.METADATA_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.PLAN_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
+/** Utilities for reading archived cleaner plans and metadata in either timeline layout. */
+public class ArchivedCleanTimelineUtils {
+
+ private ArchivedCleanTimelineUtils() {
+ }
+
+ /**
+ * Creates a timeline for the selected completed cleans, supplied in descending requested-time order.
+ * Payloads are loaded lazily in windows of at most {@code batchSize} selected instants; only the
+ * current window is retained. This lets callers stop after enough partition rows, including when
+ * some cleans have empty partition metadata, without loading the entire archived history.
+ */
+ public static HoodieTimeline getTimeline(HoodieTableMetaClient metaClient, List<HoodieInstant> selectedInstants,
+ boolean loadPlans, int batchSize) {
+ checkArgument(batchSize > 0, "Batch size must be positive");
+ List<HoodieInstant> instants = new ArrayList<>(selectedInstants);
+ TimelineFactory factory = metaClient.getTableFormat().getTimelineFactory();
+ Map<String, Integer> positions = new HashMap<>();
+ for (int i = 0; i < instants.size(); i++) {
+ positions.put(instants.get(i).requestedTime(), i);
+ }
+ boolean legacy = metaClient.getTimelineLayoutVersion().getVersion() < TimelineLayoutVersion.VERSION_2;
+ String actionField = legacy ? "actionType" : ACTION_ARCHIVED_META_FIELD;
+ String contentField = legacy ? (loadPlans ? "hoodieCleanerPlan" : "hoodieCleanMetadata")
+ : (loadPlans ? PLAN_ARCHIVED_META_FIELD : METADATA_ARCHIVED_META_FIELD);
+ HoodieArchivedTimeline.LoadMode loadMode = loadPlans ? HoodieArchivedTimeline.LoadMode.PLAN : HoodieArchivedTimeline.LoadMode.METADATA;
+ HoodieInstantReader reader = new HoodieInstantReader() {
+ private final Map<String, byte[]> contents = new ConcurrentHashMap<>();
+
+ @Override
+ public synchronized InputStream getContentStream(HoodieInstant instant) {
+ String instantTime = instant.requestedTime();
+ if (!contents.containsKey(instantTime)) {
+ Integer position = positions.get(instantTime);
+ checkArgument(position != null, "Clean instant was not selected: " + instantTime);
+ List<HoodieInstant> batch = instants.subList(position, position + Math.min(batchSize, instants.size() - position));
+ Set<String> batchTimes = new HashSet<>(batch.stream().map(HoodieInstant::requestedTime).collect(Collectors.toList()));
+ HoodieArchivedTimeline.TimeRangeFilter range = new HoodieArchivedTimeline.ClosedClosedTimeRangeFilter(
+ batch.get(batch.size() - 1).requestedTime(), batch.get(0).requestedTime());
+ contents.clear();
+ factory.createArchivedTimelineLoader().loadInstants(metaClient, legacy ? null : range, loadMode,
Review Comment:
done
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-5853904105
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308",
"triggerID" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f64c42e0a82ddca90027a8a3137afdf0b6d1120 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308)
<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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172554359
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
Review Comment:
Done in b231fe4c5612. The old helper and its unused default limit were removed; the replacement helpers require explicit limits/batch sizes.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172551334
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
+ val cleanInstants = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getReverseOrderedInstants.iterator().asScala.take(limit).toList
+ val contents = new ConcurrentHashMap[String, Array[Byte]]()
+ val factory = metaClient.getTableFormat.getTimelineFactory
+ if (cleanInstants.nonEmpty) {
+ val legacy = metaClient.getTimelineLayoutVersion.getVersion < TimelineLayoutVersion.VERSION_2
+ val actionField = if (legacy) "actionType" else "action"
+ val contentField = if (legacy) {
+ if (loadPlans) "hoodieCleanerPlan" else "hoodieCleanMetadata"
+ } else {
+ if (loadPlans) "plan" else "metadata"
+ }
+ val loadMode = if (loadPlans) HoodieArchivedTimeline.LoadMode.PLAN else HoodieArchivedTimeline.LoadMode.METADATA
+ factory.createArchivedTimelineLoader().loadInstants(metaClient,
+ new HoodieArchivedTimeline.ClosedClosedTimeRangeFilter(
+ cleanInstants.last.requestedTime(), cleanInstants.head.requestedTime()),
+ loadMode,
+ record => HoodieTimeline.CLEAN_ACTION == record.get(actionField).toString,
Review Comment:
done
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172556523
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
+ }
- // Known limitation, see #19639: getCleanerPlans collects the archived clean instants but
- // then hands every instant to processCleanPlan against the ACTIVE timeline, so the
- // archived instant's .clean.requested file is not found, the read falls back to
- // createErrorRow and every plan field comes back null. Only plan_time/state/action
- // survive, because those are taken from the instant and not from the plan. A fix has to
- // read the archived instant's own content and would flip the null assertions below to
- // non-null; merely routing to getArchivedTimeline the way the sibling ShowCleansProcedure
- // does is not enough, since that path throws today (pinned above).
- assertResult("COMPLETED")(archivedPlan.getString(1))
- assertResult("clean")(archivedPlan.getString(2))
- planFields.foreach { name =>
- assert(archivedPlan.isNullAt(archivedPlan.fieldIndex(name)),
- s"archived clean plan is expected to return a null $name today (#19639)")
+ // This table is not partitioned, so all three procedures return one row per clean.
+ assertResult(6)(rowCount("show_cleans", showArchived = true))
+ assertResult(6)(rowCount("show_cleans_metadata", showArchived = true))
+ assertResult(6)(rowCount("show_clean_plans", showArchived = true))
+
+ val procedures = Seq("show_clean_plans", "show_cleans", "show_cleans_metadata")
+ val beforeArchive = procedures.map { procedure =>
+ procedure -> spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ }.toMap
+
+ spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
+ s" retain_commits => 1, enable_metadata => false, options => 'hoodie.write.table.version=$tableVersion')").collect()
+
+ // Precondition: archival must have split the two cleans across the two timelines,
+ // otherwise the archived branch below is never exercised and the test passes vacuously.
+ val metaClient = createMetaClient(spark, tablePath)
+ assertResult(tableVersion)(metaClient.getTableConfig.getTableVersion.versionCode())
+ val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline.filterCompletedInstants
+ .getInstants.asScala.map(_.requestedTime).toSeq
+ assert(archivedCleans.nonEmpty, s"expected archived cleans, got $archivedCleans")
+ assert(activeCleans.nonEmpty, s"expected active cleans, got $activeCleans")
+ assert(archivedCleans.max.compareTo(activeCleans.min) < 0,
+ "archived cleans must be older than active cleans")
+
+ // The active-only view must exclude the archived clean for all three procedures.
+ assertResult(activeCleans.size)(rowCount("show_cleans", showArchived = false))
+ assertResult(activeCleans.size)(rowCount("show_cleans_metadata", showArchived = false))
+ assertResult(activeCleans.size)(rowCount("show_clean_plans", showArchived = false))
+
+ procedures.foreach { procedure =>
+ val allRows = spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ assertResult(beforeArchive(procedure))(allRows)
+ val limitedRows = spark.sql(s"call $procedure(table => '$tableName', showArchived => true, limit => 1)").collect().toSeq
+ assertResult(allRows.take(1))(limitedRows)
Review Comment:
Done in b231fe4c5612. Added limit = activeCleans.size + 1 assertions for all three procedures and verified that the last returned row is archivedCleans.max.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172557922
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
Review Comment:
Done in b231fe4c5612. The test matrix now uses Seq(6, HoodieTableVersion.current().versionCode()), with the required 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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-5967918255
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308",
"triggerID" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"triggerType" : "PUSH"
}, {
"hash" : "b231fe4c56126d0eba3921ff5eedef79694d6e30",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b231fe4c56126d0eba3921ff5eedef79694d6e30",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f64c42e0a82ddca90027a8a3137afdf0b6d1120 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308)
* b231fe4c56126d0eba3921ff5eedef79694d6e30 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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-5853870953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f64c42e0a82ddca90027a8a3137afdf0b6d1120 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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172549999
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -165,10 +172,13 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
getCleans(metaClient.getActiveTimeline, limit)
}
val finalResults = if (showArchived) {
+ val archivedCleanLimit = if (includePartitionMetadata) Int.MaxValue else limit
Review Comment:
Done in b231fe4c5612. Archived payloads are now loaded lazily in descending-time batches, with only the current batch retained. The partition-row iterator stops at the row limit and continues past empty cleans. Added zero-row/multi-row partition-metadata tests, including limit = 3 returning two rows from one clean and one from the next, with payload-load assertions.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansPlanProcedure.scala:
##########
@@ -184,21 +184,17 @@ class ShowCleansPlanProcedure extends BaseProcedure with ProcedureBuilder with S
}
private def getCleanerPlans(metaClient: HoodieTableMetaClient, limit: Int, showArchived: Boolean): Seq[Row] = {
- val activeCleanInstants = getSortedCleanInstants(metaClient.getActiveTimeline)
- .take(limit)
-
- val cleanInstants = if (showArchived) {
- val archivedCleanInstants = getSortedCleanInstants(metaClient.getArchivedTimeline)
- .take(limit)
- (activeCleanInstants ++ archivedCleanInstants)
- .sortWith((a, b) => a.requestedTime() > b.requestedTime())
- .take(limit)
+ val activeTimeline = metaClient.getActiveTimeline
+ val activeCleanInstants = getSortedCleanInstants(activeTimeline).take(limit)
+ val activeRows = activeCleanInstants.map(processCleanPlan(metaClient, activeTimeline, _))
+
+ if (showArchived) {
+ val archivedTimeline = ShowCleansProcedure.getArchivedCleanTimeline(metaClient, loadPlans = true, limit = limit)
+ val archivedCleanInstants = getSortedCleanInstants(archivedTimeline)
+ val archivedRows = archivedCleanInstants.map(processCleanPlan(metaClient, archivedTimeline, _))
+ (activeRows ++ archivedRows).sortWith((a, b) => a.getString(0) > b.getString(0)).take(limit)
Review Comment:
Done in b231fe4c5612. Active and archived instant descriptors are merged and the final top N is selected before any plan payload is read. The archived helper accepts those selected instants. Tests verify that active plans filling the limit load no archived payloads and that a mixed result loads only the selected archived plan.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172552981
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
+ val cleanInstants = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getReverseOrderedInstants.iterator().asScala.take(limit).toList
+ val contents = new ConcurrentHashMap[String, Array[Byte]]()
+ val factory = metaClient.getTableFormat.getTimelineFactory
+ if (cleanInstants.nonEmpty) {
+ val legacy = metaClient.getTimelineLayoutVersion.getVersion < TimelineLayoutVersion.VERSION_2
+ val actionField = if (legacy) "actionType" else "action"
+ val contentField = if (legacy) {
+ if (loadPlans) "hoodieCleanerPlan" else "hoodieCleanMetadata"
+ } else {
+ if (loadPlans) "plan" else "metadata"
+ }
Review Comment:
done，moved the format-specific decoding into ArchivedCleanTimelineUtils in hudi-common
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-5968820615
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17308",
"triggerID" : "1f64c42e0a82ddca90027a8a3137afdf0b6d1120",
"triggerType" : "PUSH"
}, {
"hash" : "b231fe4c56126d0eba3921ff5eedef79694d6e30",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17438",
"triggerID" : "b231fe4c56126d0eba3921ff5eedef79694d6e30",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b231fe4c56126d0eba3921ff5eedef79694d6e30 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17438)
<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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4118335784
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansPlanProcedure.scala:
##########
@@ -184,21 +184,17 @@ class ShowCleansPlanProcedure extends BaseProcedure with ProcedureBuilder with S
}
private def getCleanerPlans(metaClient: HoodieTableMetaClient, limit: Int, showArchived: Boolean): Seq[Row] = {
- val activeCleanInstants = getSortedCleanInstants(metaClient.getActiveTimeline)
- .take(limit)
-
- val cleanInstants = if (showArchived) {
- val archivedCleanInstants = getSortedCleanInstants(metaClient.getArchivedTimeline)
- .take(limit)
- (activeCleanInstants ++ archivedCleanInstants)
- .sortWith((a, b) => a.requestedTime() > b.requestedTime())
- .take(limit)
+ val activeTimeline = metaClient.getActiveTimeline
+ val activeCleanInstants = getSortedCleanInstants(activeTimeline).take(limit)
+ val activeRows = activeCleanInstants.map(processCleanPlan(metaClient, activeTimeline, _))
+
+ if (showArchived) {
+ val archivedTimeline = ShowCleansProcedure.getArchivedCleanTimeline(metaClient, loadPlans = true, limit = limit)
+ val archivedCleanInstants = getSortedCleanInstants(archivedTimeline)
+ val archivedRows = archivedCleanInstants.map(processCleanPlan(metaClient, archivedTimeline, _))
+ (activeRows ++ archivedRows).sortWith((a, b) => a.getString(0) > b.getString(0)).take(limit)
Review Comment:
[P2] Select the final top N instants before loading and decoding their plans. This now processes up to limit active plans plus limit archived plans before taking the final limit. If the active timeline has limit newer clean instants, every archived plan payload is loaded and decoded only to be discarded; cleaner plans can contain large file lists, so the extra archive reads and allocations can be significant. A simpler flow would merge instant descriptors while retaining their source timeline, sort/take(limit), load archived payloads only for selected archived instants, and then build rows. Having the helper accept selected instants instead of its own limit would also avoid repeated selection and sorting. The new limit => 1 result assertion would still pass with these unnecessary reads, so a payload-load assertion would help protect this behavior.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -165,10 +172,13 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
getCleans(metaClient.getActiveTimeline, limit)
}
val finalResults = if (showArchived) {
+ val archivedCleanLimit = if (includePartitionMetadata) Int.MaxValue else limit
Review Comment:
[P2] Bound archived metadata loading by the rows actually needed. For show_cleans_metadata, Int.MaxValue causes getArchivedCleanTimeline to copy every archived clean's metadata into the contents map before getCleansWithPartitionMetadata applies the row limit. Thus even limit => 1 loads and retains the full archived clean metadata history, including when newer active rows already fill the result. On long-lived tables this can cause substantial archive I/O and driver-memory pressure, potentially OOM. Could this load descending-time batches and stop once enough partition rows have been produced? Merely passing limit as the instant count would not preserve behavior because a clean can have empty partition metadata and contribute no rows. A partitioned-table test with multi-row and zero-row cleans, plus an assertion on payloads loaded, would cover this.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4127237962
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
Review Comment:
🤖 nit: the misaligned continuation indentation for `loadPlans`/`limit` params stands out from the rest of the file's style — worth running through scalafmt/checkstyle before merging.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -165,10 +172,13 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
getCleans(metaClient.getActiveTimeline, limit)
}
val finalResults = if (showArchived) {
+ val archivedCleanLimit = if (includePartitionMetadata) Int.MaxValue else limit
Review Comment:
🤖 One way to get this bound without guessing a batch size: fill the temporary timeline's `HoodieInstantReader.getContentStream` lazily, loading payloads on the first miss for a descending window of instants such as the next `limit`. Then the existing `takeWhile(rowCount < limit)` in `getCleansWithPartitionMetadata` decides how many archived cleans get read. Empty (0-row) cleans just pull in the next window, so it stays exact on both points raised here and in the `limit` suggestion below. On V2, `ArchivedTimelineLoaderV2.getFilteredFiles` prunes LSM files by the time range, so each window only reads the files that overlap it. On V1 each window rescans archive files, so the window should stay reasonably large there.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172586737
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
+ }
- // Known limitation, see #19639: getCleanerPlans collects the archived clean instants but
- // then hands every instant to processCleanPlan against the ACTIVE timeline, so the
- // archived instant's .clean.requested file is not found, the read falls back to
- // createErrorRow and every plan field comes back null. Only plan_time/state/action
- // survive, because those are taken from the instant and not from the plan. A fix has to
- // read the archived instant's own content and would flip the null assertions below to
- // non-null; merely routing to getArchivedTimeline the way the sibling ShowCleansProcedure
- // does is not enough, since that path throws today (pinned above).
- assertResult("COMPLETED")(archivedPlan.getString(1))
- assertResult("clean")(archivedPlan.getString(2))
- planFields.foreach { name =>
- assert(archivedPlan.isNullAt(archivedPlan.fieldIndex(name)),
- s"archived clean plan is expected to return a null $name today (#19639)")
+ // This table is not partitioned, so all three procedures return one row per clean.
+ assertResult(6)(rowCount("show_cleans", showArchived = true))
+ assertResult(6)(rowCount("show_cleans_metadata", showArchived = true))
+ assertResult(6)(rowCount("show_clean_plans", showArchived = true))
+
+ val procedures = Seq("show_clean_plans", "show_cleans", "show_cleans_metadata")
+ val beforeArchive = procedures.map { procedure =>
+ procedure -> spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ }.toMap
+
+ spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
+ s" retain_commits => 1, enable_metadata => false, options => 'hoodie.write.table.version=$tableVersion')").collect()
+
+ // Precondition: archival must have split the two cleans across the two timelines,
Review Comment:
done
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
Review Comment:
done
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172586737
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -481,126 +479,123 @@ class TestShowCleansProcedures extends HoodieSparkProcedureTestBase {
}
}
- test("Test show_clean_plans with an archived clean instant") {
- withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false") {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = tmp.getCanonicalPath
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- | ) using hudi
- | location '$tablePath'
- | tblproperties (
- | primaryKey = 'id',
- | type = 'cow',
- | preCombineField = 'ts',
- | hoodie.metadata.enable = 'false'
- | )
- |""".stripMargin)
+ Seq(6, 9).foreach { tableVersion =>
+ test(s"Test show_clean_plans with an archived clean instant - table version $tableVersion") {
+ withSQLConf("hoodie.clean.automatic" -> "false", "hoodie.archive.automatic" -> "false",
+ "hoodie.write.table.version" -> tableVersion.toString) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ | ) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | hoodie.metadata.enable = 'false',
+ | 'hoodie.write.table.version' = '$tableVersion'
+ | )
+ |""".stripMargin)
- def rowCount(procedure: String, showArchived: Boolean): Int =
- spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
-
- // Six write commits with a clean in the middle: the first clean sits before the last
- // commit that archival will move, so it gets archived along with those commits, while
- // the second clean stays on the active timeline.
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
- spark.sql(s"update $tableName set price = 11 where id = 1")
- spark.sql(s"update $tableName set price = 12 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
- spark.sql(s"update $tableName set price = 13 where id = 1")
- spark.sql(s"update $tableName set price = 14 where id = 1")
- spark.sql(s"update $tableName set price = 15 where id = 1")
- spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1)").collect()
-
- // Nothing has been archived yet, so showArchived => true here only exercises the merge of
- // the active timeline with an empty archived one. All three procedures succeed and see
- // both cleans, which is what makes the failures asserted after archival attributable to
- // the archived instants themselves rather than to the merged read path. Two rows for
- // show_cleans_metadata as well, since that procedure emits one row per partition per
- // clean and this table is not partitioned.
- assertResult(2)(rowCount("show_cleans", showArchived = true))
- assertResult(2)(rowCount("show_cleans_metadata", showArchived = true))
- assertResult(2)(rowCount("show_clean_plans", showArchived = true))
-
- spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
- " retain_commits => 1, enable_metadata => false)").collect()
-
- // Precondition: archival must have split the two cleans across the two timelines,
- // otherwise the archived branch below is never exercised and the test passes vacuously.
- val metaClient = createMetaClient(spark, tablePath)
- val archivedCleans = metaClient.getArchivedTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- val activeCleans = metaClient.getActiveTimeline.getCleanerTimeline
- .getInstants.asScala.map(_.requestedTime).toSeq
- assert(archivedCleans.length == 1, s"expected exactly 1 archived clean, got $archivedCleans")
- assert(activeCleans.length == 1, s"expected exactly 1 active clean, got $activeCleans")
- assert(archivedCleans.head.compareTo(activeCleans.head) < 0,
- "the archived clean must be the older of the two")
-
- // Sibling-procedure controls, pinning what the other two procedures do with the very same
- // archived clean. On the active timeline all three agree: one row for the one active
- // clean. With showArchived => true they diverge, and neither sibling is correct today.
- assertResult(1)(rowCount("show_cleans", showArchived = false))
- assertResult(1)(rowCount("show_cleans_metadata", showArchived = false))
- assertResult(1)(rowCount("show_clean_plans", showArchived = false))
-
- // The other half of #19639, which covers all three clean procedures. show_cleans and its
- // show_cleans_metadata variant do route to getArchivedTimeline, but the archived instants
- // carry no content there, so readCleanMetadata cannot deserialize them and the call fails
- // outright rather than degrading to a partial row. Same missing-archived-content cause as
- // the all-null plan rows asserted below, just a harsher symptom. Pinned as observed.
- Seq("show_cleans", "show_cleans_metadata").foreach { procedure =>
- val e = intercept[IOException](rowCount(procedure, showArchived = true))
- assert(e.getMessage.contains(archivedCleans.head),
- s"$procedure over the archived timeline should fail on the archived clean" +
- s" ${archivedCleans.head}, but failed with: ${e.getMessage}")
- }
+ def rowCount(procedure: String, showArchived: Boolean): Int =
+ spark.sql(s"call $procedure(table => '$tableName', showArchived => $showArchived)").collect().length
- val plans = spark.sql(s"call show_clean_plans(table => '$tableName', showArchived => true)").collect()
- assertResult(2)(plans.length)
- val planTimes = plans.map(_.getString(0)).mkString(", ")
- val activePlan = plans.find(_.getString(0) == activeCleans.head)
- .getOrElse(fail(s"no plan row for the active clean ${activeCleans.head}, got plan_times: $planTimes"))
- val archivedPlan = plans.find(_.getString(0) == archivedCleans.head)
- .getOrElse(fail(s"no plan row for the archived clean ${archivedCleans.head}, got plan_times: $planTimes"))
-
- // Fields that come from the cleaner plan itself, resolved by name off the row schema so
- // that a change in output-schema ordering cannot silently re-point these assertions.
- // extra_metadata is left out: it is null for both rows, so it does not discriminate.
- val planFields = Seq(
- "earliest_instant_to_retain",
- "last_completed_commit_timestamp",
- "policy",
- "version",
- "total_partitions_to_clean",
- "total_partitions_to_delete")
-
- // The active clean plan is read correctly.
- assertResult("COMPLETED")(activePlan.getString(1))
- assertResult("clean")(activePlan.getString(2))
- planFields.foreach { name =>
- assert(!activePlan.isNullAt(activePlan.fieldIndex(name)), s"active clean plan should have a non-null $name")
- }
+ // V1 archives cleans using a separate count threshold, so generate enough cleans
+ // to force archival with both timeline formats.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set price = 11 where id = 1")
+ spark.sql(s"update $tableName set price = 12 where id = 1")
+ (0 until 6).foreach { i =>
+ if (i > 0) {
+ spark.sql(s"update $tableName set price = ${12 + i} where id = 1")
+ }
+ spark.sql(s"call run_clean(table => '$tableName', retain_commits => 1, " +
+ s"options => 'hoodie.write.table.version=$tableVersion')").collect()
+ }
- // Known limitation, see #19639: getCleanerPlans collects the archived clean instants but
- // then hands every instant to processCleanPlan against the ACTIVE timeline, so the
- // archived instant's .clean.requested file is not found, the read falls back to
- // createErrorRow and every plan field comes back null. Only plan_time/state/action
- // survive, because those are taken from the instant and not from the plan. A fix has to
- // read the archived instant's own content and would flip the null assertions below to
- // non-null; merely routing to getArchivedTimeline the way the sibling ShowCleansProcedure
- // does is not enough, since that path throws today (pinned above).
- assertResult("COMPLETED")(archivedPlan.getString(1))
- assertResult("clean")(archivedPlan.getString(2))
- planFields.foreach { name =>
- assert(archivedPlan.isNullAt(archivedPlan.fieldIndex(name)),
- s"archived clean plan is expected to return a null $name today (#19639)")
+ // This table is not partitioned, so all three procedures return one row per clean.
+ assertResult(6)(rowCount("show_cleans", showArchived = true))
+ assertResult(6)(rowCount("show_cleans_metadata", showArchived = true))
+ assertResult(6)(rowCount("show_clean_plans", showArchived = true))
+
+ val procedures = Seq("show_clean_plans", "show_cleans", "show_cleans_metadata")
+ val beforeArchive = procedures.map { procedure =>
+ procedure -> spark.sql(s"call $procedure(table => '$tableName', showArchived => true)").collect().toSeq
+ }.toMap
+
+ spark.sql(s"call archive_commits(table => '$tableName', min_commits => 2, max_commits => 3," +
+ s" retain_commits => 1, enable_metadata => false, options => 'hoodie.write.table.version=$tableVersion')").collect()
+
+ // Precondition: archival must have split the two cleans across the two timelines,
Review Comment:
Done.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172551334
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
+ val cleanInstants = metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants
+ .getReverseOrderedInstants.iterator().asScala.take(limit).toList
+ val contents = new ConcurrentHashMap[String, Array[Byte]]()
+ val factory = metaClient.getTableFormat.getTimelineFactory
+ if (cleanInstants.nonEmpty) {
+ val legacy = metaClient.getTimelineLayoutVersion.getVersion < TimelineLayoutVersion.VERSION_2
+ val actionField = if (legacy) "actionType" else "action"
+ val contentField = if (legacy) {
+ if (loadPlans) "hoodieCleanerPlan" else "hoodieCleanMetadata"
+ } else {
+ if (loadPlans) "plan" else "metadata"
+ }
+ val loadMode = if (loadPlans) HoodieArchivedTimeline.LoadMode.PLAN else HoodieArchivedTimeline.LoadMode.METADATA
+ factory.createArchivedTimelineLoader().loadInstants(metaClient,
+ new HoodieArchivedTimeline.ClosedClosedTimeRangeFilter(
+ cleanInstants.last.requestedTime(), cleanInstants.head.requestedTime()),
+ loadMode,
+ record => HoodieTimeline.CLEAN_ACTION == record.get(actionField).toString,
Review Comment:
Done in b231fe4c5612. The V1 loader now receives a null range filter, with range and selected-instant checks applied in the consumer. Added a V1 regression test with three separate archive files and a commit-only middle file; both older and newer clean plans and metadata are read successfully.
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172549999
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -165,10 +172,13 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
getCleans(metaClient.getActiveTimeline, limit)
}
val finalResults = if (showArchived) {
+ val archivedCleanLimit = if (includePartitionMetadata) Int.MaxValue else limit
Review Comment:
done
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansPlanProcedure.scala:
##########
@@ -184,21 +184,17 @@ class ShowCleansPlanProcedure extends BaseProcedure with ProcedureBuilder with S
}
private def getCleanerPlans(metaClient: HoodieTableMetaClient, limit: Int, showArchived: Boolean): Seq[Row] = {
- val activeCleanInstants = getSortedCleanInstants(metaClient.getActiveTimeline)
- .take(limit)
-
- val cleanInstants = if (showArchived) {
- val archivedCleanInstants = getSortedCleanInstants(metaClient.getArchivedTimeline)
- .take(limit)
- (activeCleanInstants ++ archivedCleanInstants)
- .sortWith((a, b) => a.requestedTime() > b.requestedTime())
- .take(limit)
+ val activeTimeline = metaClient.getActiveTimeline
+ val activeCleanInstants = getSortedCleanInstants(activeTimeline).take(limit)
+ val activeRows = activeCleanInstants.map(processCleanPlan(metaClient, activeTimeline, _))
+
+ if (showArchived) {
+ val archivedTimeline = ShowCleansProcedure.getArchivedCleanTimeline(metaClient, loadPlans = true, limit = limit)
+ val archivedCleanInstants = getSortedCleanInstants(archivedTimeline)
+ val archivedRows = archivedCleanInstants.map(processCleanPlan(metaClient, archivedTimeline, _))
+ (activeRows ++ archivedRows).sortWith((a, b) => a.getString(0) > b.getString(0)).take(limit)
Review Comment:
done
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4172553565
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -165,10 +172,13 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
getCleans(metaClient.getActiveTimeline, limit)
}
val finalResults = if (showArchived) {
+ val archivedCleanLimit = if (includePartitionMetadata) Int.MaxValue else limit
Review Comment:
done
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +266,48 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getArchivedCleanTimeline(metaClient: HoodieTableMetaClient,
+ loadPlans: Boolean,
+ limit: Int = Int.MaxValue): HoodieTimeline = {
Review Comment:
done
--
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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20099:
URL: https://github.com/apache/hudi/pull/20099#discussion_r4235940703
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ArchivedCleanTimelineUtils.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.common.table.timeline;
+
+import org.apache.hudi.common.avro.HoodieAvroUtils;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.ACTION_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.METADATA_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.PLAN_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
+/** Utilities for reading archived cleaner plans and metadata in either timeline layout. */
+public class ArchivedCleanTimelineUtils {
+
+ private ArchivedCleanTimelineUtils() {
+ }
+
+ /**
+ * Creates a timeline for the selected completed cleans, supplied in descending requested-time order.
+ * Payloads are loaded lazily in windows of at most {@code batchSize} selected instants; only the
+ * current window is retained. This lets callers stop after enough partition rows, including when
+ * some cleans have empty partition metadata, without loading the entire archived history.
+ */
+ public static HoodieTimeline getTimeline(HoodieTableMetaClient metaClient, List<HoodieInstant> selectedInstants,
+ boolean loadPlans, int batchSize) {
+ checkArgument(batchSize > 0, "Batch size must be positive");
+ List<HoodieInstant> instants = new ArrayList<>(selectedInstants);
+ TimelineFactory factory = metaClient.getTableFormat().getTimelineFactory();
+ Map<String, Integer> positions = new HashMap<>();
+ for (int i = 0; i < instants.size(); i++) {
+ positions.put(instants.get(i).requestedTime(), i);
+ }
+ boolean legacy = metaClient.getTimelineLayoutVersion().getVersion() < TimelineLayoutVersion.VERSION_2;
+ String actionField = legacy ? "actionType" : ACTION_ARCHIVED_META_FIELD;
+ String contentField = legacy ? (loadPlans ? "hoodieCleanerPlan" : "hoodieCleanMetadata")
+ : (loadPlans ? PLAN_ARCHIVED_META_FIELD : METADATA_ARCHIVED_META_FIELD);
+ HoodieArchivedTimeline.LoadMode loadMode = loadPlans ? HoodieArchivedTimeline.LoadMode.PLAN : HoodieArchivedTimeline.LoadMode.METADATA;
+ HoodieInstantReader reader = new HoodieInstantReader() {
+ private final Map<String, byte[]> contents = new ConcurrentHashMap<>();
+
+ @Override
+ public synchronized InputStream getContentStream(HoodieInstant instant) {
+ String instantTime = instant.requestedTime();
+ if (!contents.containsKey(instantTime)) {
+ Integer position = positions.get(instantTime);
+ checkArgument(position != null, "Clean instant was not selected: " + instantTime);
+ List<HoodieInstant> batch = instants.subList(position, position + Math.min(batchSize, instants.size() - position));
+ Set<String> batchTimes = new HashSet<>(batch.stream().map(HoodieInstant::requestedTime).collect(Collectors.toList()));
+ HoodieArchivedTimeline.TimeRangeFilter range = new HoodieArchivedTimeline.ClosedClosedTimeRangeFilter(
+ batch.get(batch.size() - 1).requestedTime(), batch.get(0).requestedTime());
+ contents.clear();
+ factory.createArchivedTimelineLoader().loadInstants(metaClient, legacy ? null : range, loadMode,
Review Comment:
🤖 non-blocking: on V1 the range is dropped, so every window and every archived clean whose payload is missing (like old archives with no requested record) rescans the whole `.commits_.archive*` set; could you remember the batch times you already scanned so a miss throws without loading again?
```java
private final Set<String> scanned = ConcurrentHashMap.newKeySet();
...
if (!scanned.contains(instantTime)) { contents.clear(); scanned.clear(); scanned.addAll(batchTimes); /* load */ }
```
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ArchivedCleanTimelineUtils.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.common.table.timeline;
+
+import org.apache.hudi.common.avro.HoodieAvroUtils;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.ACTION_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.METADATA_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.PLAN_ARCHIVED_META_FIELD;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
+/** Utilities for reading archived cleaner plans and metadata in either timeline layout. */
+public class ArchivedCleanTimelineUtils {
+
+ private ArchivedCleanTimelineUtils() {
+ }
+
+ /**
+ * Creates a timeline for the selected completed cleans, supplied in descending requested-time order.
+ * Payloads are loaded lazily in windows of at most {@code batchSize} selected instants; only the
+ * current window is retained. This lets callers stop after enough partition rows, including when
+ * some cleans have empty partition metadata, without loading the entire archived history.
+ */
+ public static HoodieTimeline getTimeline(HoodieTableMetaClient metaClient, List<HoodieInstant> selectedInstants,
+ boolean loadPlans, int batchSize) {
+ checkArgument(batchSize > 0, "Batch size must be positive");
+ List<HoodieInstant> instants = new ArrayList<>(selectedInstants);
+ TimelineFactory factory = metaClient.getTableFormat().getTimelineFactory();
+ Map<String, Integer> positions = new HashMap<>();
+ for (int i = 0; i < instants.size(); i++) {
+ positions.put(instants.get(i).requestedTime(), i);
+ }
+ boolean legacy = metaClient.getTimelineLayoutVersion().getVersion() < TimelineLayoutVersion.VERSION_2;
+ String actionField = legacy ? "actionType" : ACTION_ARCHIVED_META_FIELD;
+ String contentField = legacy ? (loadPlans ? "hoodieCleanerPlan" : "hoodieCleanMetadata")
+ : (loadPlans ? PLAN_ARCHIVED_META_FIELD : METADATA_ARCHIVED_META_FIELD);
+ HoodieArchivedTimeline.LoadMode loadMode = loadPlans ? HoodieArchivedTimeline.LoadMode.PLAN : HoodieArchivedTimeline.LoadMode.METADATA;
+ HoodieInstantReader reader = new HoodieInstantReader() {
+ private final Map<String, byte[]> contents = new ConcurrentHashMap<>();
+
+ @Override
+ public synchronized InputStream getContentStream(HoodieInstant instant) {
Review Comment:
🤖 nit: could you move the batch-loading body of `getContentStream` into a private helper, such as `loadBatch(...)`, plus a small `extractPayloadBytes(content, legacy)`? The anonymous reader nests about five levels deep inside a roughly 60-line method, which makes the windowing logic hard to follow.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -256,6 +249,37 @@ object ShowCleansProcedure {
val NAME = "show_cleans"
def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false)
+
+ private[procedures] def getCleanTimeline(metaClient: HoodieTableMetaClient,
+ showArchived: Boolean,
+ loadPlans: Boolean,
+ instantLimit: Int,
+ batchSize: Int): HoodieTimeline = {
+ val activeTimeline = metaClient.getActiveTimeline
+ val activeCleans = if (loadPlans) activeTimeline.getCleanerTimeline
+ else activeTimeline.getCleanerTimeline.filterCompletedInstants
+ val activeInstants = activeCleans.getInstants.asScala.toSeq
+ val archivedInstants = if (showArchived) {
+ metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants.getInstants.asScala.toSeq
+ } else {
+ Seq.empty[HoodieInstant]
+ }
+ val comparator = TimelineLayout.fromVersion(metaClient.getTimelineLayoutVersion)
+ .getInstantComparator.requestedTimeOrderedComparator.reversed()
+ val selected = (activeInstants.map(_ -> true) ++ archivedInstants.map(_ -> false))
Review Comment:
🤖 nit: the `_ -> true` / `_ -> false` tags and the `case (instant, false)` matches read as magic booleans; would a named tag (for example `isActive`) or two separate collections make the active/archived split clearer?
<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] fix(tools): read archived clean plans and metadata [hudi]
Posted by "KiteSoar (via GitHub)" <gi...@apache.org>.
KiteSoar commented on PR #20099:
URL: https://github.com/apache/hudi/pull/20099#issuecomment-6407876129
Hi，@voonhous @danny0405 when you have a moment, could you take another look?
--
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