You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org> on 2025/08/28 01:52:39 UTC
[PR] Hudi 9765 support for plans along with partition metadata in all the spark action procedures [hudi]
vamshikrishnakyatham opened a new pull request, #13793:
URL: https://github.com/apache/hudi/pull/13793
### Change Logs
Consolidated multiple Hudi Spark SQL procedures into unified, comprehensive procedures that support both completed and pending instants with partition-specific metadata.
### Impact
**Public API Changes**:
- Unified Show Procedures for actions like `ShowCommitsProcedure`, `ShowClusteringProcedure`, `ShowCompactionProcedure`, `ShowCleansProcedure`, `ShowRollbacksProcedure` and `ShowSavepointProcedure` that support both completed and pending instants with partition-specific and file-specific (wherever needed) metadata.
### Risk level (write none, low medium or high below)
low
### Documentation Update
- Hudi website documentation for new unified procedures
- Migration guide from legacy to new procedures
- Updated procedure reference with enhanced schemas
- Deprecation notices and migration examples
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3235094715
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fa38e71eb89ffe5571e9f64cc45407e6319c62b2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3235366955
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fa38e71eb89ffe5571e9f64cc45407e6319c62b2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223)
* 0b4dffdd4e409045edd57e38b0376f0cf98452a2 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3233947443
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2317615953
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -138,128 +171,245 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
StructField("failed_delete_files", DataTypes.IntegerType, nullable = true, Metadata.empty),
StructField("is_partition_deleted", DataTypes.BooleanType, nullable = true, Metadata.empty),
StructField("time_taken_in_millis", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty)
+ StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("earliest_commit_to_retain", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("last_completed_commit_timestamp", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("version", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_clean", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_delete", DataTypes.IntegerType, nullable = true, Metadata.empty)
))
def parameters: Array[ProcedureParameter] = PARAMETERS
- def outputType: StructType = if (includePartitionMetadata) PARTITION_METADATA_OUTPUT_TYPE else OUTPUT_TYPE
+ def outputType: StructType = OUTPUT_TYPE
override def call(args: ProcedureArgs): Seq[Row] = {
super.checkArgs(PARAMETERS, args)
- val table = getArgValueOrDefault(args, PARAMETERS(0)).get.asInstanceOf[String]
- val limit = getArgValueOrDefault(args, PARAMETERS(1)).get.asInstanceOf[Int]
- val showArchived = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Boolean]
- val filter = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[String]
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
+ val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
validateFilter(filter, outputType)
- val hoodieCatalogTable = HoodieCLIUtils.getHoodieCatalogTable(sparkSession, table)
- val basePath = hoodieCatalogTable.tableLocation
+ val basePath = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val activeResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit)
- } else {
- getCleans(metaClient.getActiveTimeline, limit)
- }
+ val activeResults = getCombinedCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
val finalResults = if (showArchived) {
- val archivedResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit)
+ val archivedResults = getCombinedCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
} else {
- getCleans(metaClient.getArchivedTimeline, limit)
+ combinedResults.take(limit)
}
- (activeResults ++ archivedResults)
- .sortWith((a, b) => a.getString(0) > b.getString(0))
- .take(limit)
} else {
- activeResults
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
}
applyFilter(finalResults, filter, outputType)
}
- override def build: Procedure = new ShowCleansProcedure(includePartitionMetadata)
+ override def build: Procedure = new ShowCleansProcedure()
- private def getCleansWithPartitionMetadata(timeline: HoodieTimeline,
- limit: Int): Seq[Row] = {
+ private def getCombinedCleansWithPartitionMetadata(timeline: HoodieTimeline, limit: Int, metaClient: HoodieTableMetaClient, startTime: String, endTime: String, timelineType: String): Seq[Row] = {
import scala.collection.JavaConverters._
+ import scala.util.{Failure, Success, Try}
+
+ val filteredCleanInstants = timeline.getCleanerTimeline.getInstants.asScala.toSeq
+ .filter { instant =>
+ val instantTime = instant.requestedTime()
+ val withinStartTime = if (startTime.nonEmpty) instantTime >= startTime else true
+ val withinEndTime = if (endTime.nonEmpty) instantTime <= endTime else true
+ withinStartTime && withinEndTime
+ }
+ val allCleanInstants = if (startTime.nonEmpty && endTime.nonEmpty) {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime())
+ } else {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime()).take(limit)
+ }
- val (rows: util.ArrayList[Row], cleanInstants: util.ArrayList[HoodieInstant]) = getSortedCleans(timeline)
-
- var rowCount = 0
-
- cleanInstants.asScala.takeWhile(_ => rowCount < limit).foreach { cleanInstant =>
- val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
-
- cleanMetadata.getPartitionMetadata.entrySet.asScala.takeWhile(_ => rowCount < limit).foreach { partitionMetadataEntry =>
- val partitionPath = partitionMetadataEntry.getKey
- val partitionMetadata = partitionMetadataEntry.getValue
-
- rows.add(Row(
- cleanInstant.requestedTime(),
- cleanInstant.getCompletionTime,
- cleanInstant.getAction,
- cleanMetadata.getStartCleanTime,
- partitionPath,
- partitionMetadata.getPolicy,
- partitionMetadata.getDeletePathPatterns.size(),
- partitionMetadata.getSuccessDeleteFiles.size(),
- partitionMetadata.getFailedDeleteFiles.size(),
- partitionMetadata.getIsPartitionDeleted,
- cleanMetadata.getTimeTakenInMillis,
- cleanMetadata.getTotalFilesDeleted
- ))
- rowCount += 1
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ allCleanInstants.foreach { cleanInstant =>
+ if (cleanInstant.getState == HoodieInstant.State.COMPLETED) {
+ Try {
+ val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
+
+ cleanMetadata.getPartitionMetadata.entrySet.asScala.foreach { partitionMetadataEntry =>
+ val partitionPath = partitionMetadataEntry.getKey
+ val partitionMetadata = partitionMetadataEntry.getValue
+
+ val row = Row(
+ cleanInstant.requestedTime(),
Review Comment:
fixed
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316667736
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
+ * - `limit`: Optional. Maximum number of clean operations to return (default: 10, ignored if time range specified)
* - `showArchived`: Optional. Whether to include archived clean operations (default: false)
* - `filter`: Optional. SQL expression to filter results (default: empty string)
+ * - `startTime`: Optional. Start time for clean operations (format: yyyyMMddHHmmss, default: empty)
Review Comment:
Sure, @yihua just FYI we are using camel cased style for the input parameters of all the show procedures to be consistent.
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316667736
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
+ * - `limit`: Optional. Maximum number of clean operations to return (default: 10, ignored if time range specified)
* - `showArchived`: Optional. Whether to include archived clean operations (default: false)
* - `filter`: Optional. SQL expression to filter results (default: empty string)
+ * - `startTime`: Optional. Start time for clean operations (format: yyyyMMddHHmmss, default: empty)
Review Comment:
Sure, @yihua just FYI we are using camrl cased style for the input parameters of all the show procedures to be consistent.
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3234539849
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200)
* fa38e71eb89ffe5571e9f64cc45407e6319c62b2 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231116476
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
}, {
"hash" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2b500858557a3e3d917e42959d66237dd1d4599c Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191)
* 9c6fa69d9084d835fab01d1bffd42d5a1b3e453a 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231853157
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
}, {
"hash" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194",
"triggerID" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"triggerType" : "PUSH"
}, {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9c6fa69d9084d835fab01d1bffd42d5a1b3e453a Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194)
* d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231577224
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
}, {
"hash" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194",
"triggerID" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2b500858557a3e3d917e42959d66237dd1d4599c Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191)
* 9c6fa69d9084d835fab01d1bffd42d5a1b3e453a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2308551108
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
Review Comment:
Ah, my bad, its optional, either of the table name or path is required, we do that verification in the `getBaseFile` method
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231665976
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
}, {
"hash" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194",
"triggerID" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9c6fa69d9084d835fab01d1bffd42d5a1b3e453a Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231969554
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
}, {
"hash" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194",
"triggerID" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"triggerType" : "PUSH"
}, {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3235484628
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b4dffdd4e409045edd57e38b0376f0cf98452a2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3235369717
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fa38e71eb89ffe5571e9f64cc45407e6319c62b2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223)
* 0b4dffdd4e409045edd57e38b0376f0cf98452a2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238696376
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ec3e1cff11c11ea9bafed719a2598bcba95d0da2 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274)
* 6e153692e2f3f03236bd9951882ac0d170401e8f 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238768232
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
}, {
"hash" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277",
"triggerID" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6e153692e2f3f03236bd9951882ac0d170401e8f Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275)
* a60e8bead700c96e9848131a3c621d03e1bbaf06 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238853526
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
}, {
"hash" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277",
"triggerID" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60e8bead700c96e9848131a3c621d03e1bbaf06 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316144715
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -138,128 +171,245 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
StructField("failed_delete_files", DataTypes.IntegerType, nullable = true, Metadata.empty),
StructField("is_partition_deleted", DataTypes.BooleanType, nullable = true, Metadata.empty),
StructField("time_taken_in_millis", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty)
+ StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("earliest_commit_to_retain", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("last_completed_commit_timestamp", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("version", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_clean", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_delete", DataTypes.IntegerType, nullable = true, Metadata.empty)
))
def parameters: Array[ProcedureParameter] = PARAMETERS
- def outputType: StructType = if (includePartitionMetadata) PARTITION_METADATA_OUTPUT_TYPE else OUTPUT_TYPE
+ def outputType: StructType = OUTPUT_TYPE
override def call(args: ProcedureArgs): Seq[Row] = {
super.checkArgs(PARAMETERS, args)
- val table = getArgValueOrDefault(args, PARAMETERS(0)).get.asInstanceOf[String]
- val limit = getArgValueOrDefault(args, PARAMETERS(1)).get.asInstanceOf[Int]
- val showArchived = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Boolean]
- val filter = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[String]
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
+ val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
validateFilter(filter, outputType)
- val hoodieCatalogTable = HoodieCLIUtils.getHoodieCatalogTable(sparkSession, table)
- val basePath = hoodieCatalogTable.tableLocation
+ val basePath = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val activeResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit)
- } else {
- getCleans(metaClient.getActiveTimeline, limit)
- }
+ val activeResults = getCombinedCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
val finalResults = if (showArchived) {
- val archivedResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit)
+ val archivedResults = getCombinedCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
} else {
- getCleans(metaClient.getArchivedTimeline, limit)
+ combinedResults.take(limit)
}
- (activeResults ++ archivedResults)
- .sortWith((a, b) => a.getString(0) > b.getString(0))
- .take(limit)
} else {
- activeResults
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
}
applyFilter(finalResults, filter, outputType)
}
- override def build: Procedure = new ShowCleansProcedure(includePartitionMetadata)
+ override def build: Procedure = new ShowCleansProcedure()
- private def getCleansWithPartitionMetadata(timeline: HoodieTimeline,
- limit: Int): Seq[Row] = {
+ private def getCombinedCleansWithPartitionMetadata(timeline: HoodieTimeline, limit: Int, metaClient: HoodieTableMetaClient, startTime: String, endTime: String, timelineType: String): Seq[Row] = {
import scala.collection.JavaConverters._
+ import scala.util.{Failure, Success, Try}
+
+ val filteredCleanInstants = timeline.getCleanerTimeline.getInstants.asScala.toSeq
+ .filter { instant =>
+ val instantTime = instant.requestedTime()
+ val withinStartTime = if (startTime.nonEmpty) instantTime >= startTime else true
+ val withinEndTime = if (endTime.nonEmpty) instantTime <= endTime else true
+ withinStartTime && withinEndTime
+ }
+ val allCleanInstants = if (startTime.nonEmpty && endTime.nonEmpty) {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime())
+ } else {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime()).take(limit)
+ }
- val (rows: util.ArrayList[Row], cleanInstants: util.ArrayList[HoodieInstant]) = getSortedCleans(timeline)
-
- var rowCount = 0
-
- cleanInstants.asScala.takeWhile(_ => rowCount < limit).foreach { cleanInstant =>
- val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
-
- cleanMetadata.getPartitionMetadata.entrySet.asScala.takeWhile(_ => rowCount < limit).foreach { partitionMetadataEntry =>
- val partitionPath = partitionMetadataEntry.getKey
- val partitionMetadata = partitionMetadataEntry.getValue
-
- rows.add(Row(
- cleanInstant.requestedTime(),
- cleanInstant.getCompletionTime,
- cleanInstant.getAction,
- cleanMetadata.getStartCleanTime,
- partitionPath,
- partitionMetadata.getPolicy,
- partitionMetadata.getDeletePathPatterns.size(),
- partitionMetadata.getSuccessDeleteFiles.size(),
- partitionMetadata.getFailedDeleteFiles.size(),
- partitionMetadata.getIsPartitionDeleted,
- cleanMetadata.getTimeTakenInMillis,
- cleanMetadata.getTotalFilesDeleted
- ))
- rowCount += 1
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ allCleanInstants.foreach { cleanInstant =>
+ if (cleanInstant.getState == HoodieInstant.State.COMPLETED) {
+ Try {
+ val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
+
+ cleanMetadata.getPartitionMetadata.entrySet.asScala.foreach { partitionMetadataEntry =>
+ val partitionPath = partitionMetadataEntry.getKey
+ val partitionMetadata = partitionMetadataEntry.getValue
+
+ val row = Row(
+ cleanInstant.requestedTime(),
Review Comment:
Remove "()"s to make consistent.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -138,128 +171,245 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
StructField("failed_delete_files", DataTypes.IntegerType, nullable = true, Metadata.empty),
StructField("is_partition_deleted", DataTypes.BooleanType, nullable = true, Metadata.empty),
StructField("time_taken_in_millis", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty)
+ StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("earliest_commit_to_retain", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("last_completed_commit_timestamp", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("version", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_clean", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_delete", DataTypes.IntegerType, nullable = true, Metadata.empty)
))
def parameters: Array[ProcedureParameter] = PARAMETERS
- def outputType: StructType = if (includePartitionMetadata) PARTITION_METADATA_OUTPUT_TYPE else OUTPUT_TYPE
+ def outputType: StructType = OUTPUT_TYPE
override def call(args: ProcedureArgs): Seq[Row] = {
super.checkArgs(PARAMETERS, args)
- val table = getArgValueOrDefault(args, PARAMETERS(0)).get.asInstanceOf[String]
- val limit = getArgValueOrDefault(args, PARAMETERS(1)).get.asInstanceOf[Int]
- val showArchived = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Boolean]
- val filter = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[String]
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
+ val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
validateFilter(filter, outputType)
- val hoodieCatalogTable = HoodieCLIUtils.getHoodieCatalogTable(sparkSession, table)
- val basePath = hoodieCatalogTable.tableLocation
+ val basePath = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val activeResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit)
- } else {
- getCleans(metaClient.getActiveTimeline, limit)
- }
+ val activeResults = getCombinedCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
val finalResults = if (showArchived) {
- val archivedResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit)
+ val archivedResults = getCombinedCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
} else {
- getCleans(metaClient.getArchivedTimeline, limit)
+ combinedResults.take(limit)
}
- (activeResults ++ archivedResults)
- .sortWith((a, b) => a.getString(0) > b.getString(0))
- .take(limit)
} else {
- activeResults
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
}
applyFilter(finalResults, filter, outputType)
}
- override def build: Procedure = new ShowCleansProcedure(includePartitionMetadata)
+ override def build: Procedure = new ShowCleansProcedure()
- private def getCleansWithPartitionMetadata(timeline: HoodieTimeline,
- limit: Int): Seq[Row] = {
+ private def getCombinedCleansWithPartitionMetadata(timeline: HoodieTimeline, limit: Int, metaClient: HoodieTableMetaClient, startTime: String, endTime: String, timelineType: String): Seq[Row] = {
import scala.collection.JavaConverters._
+ import scala.util.{Failure, Success, Try}
+
+ val filteredCleanInstants = timeline.getCleanerTimeline.getInstants.asScala.toSeq
+ .filter { instant =>
+ val instantTime = instant.requestedTime()
+ val withinStartTime = if (startTime.nonEmpty) instantTime >= startTime else true
+ val withinEndTime = if (endTime.nonEmpty) instantTime <= endTime else true
+ withinStartTime && withinEndTime
+ }
+ val allCleanInstants = if (startTime.nonEmpty && endTime.nonEmpty) {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime())
+ } else {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime()).take(limit)
+ }
- val (rows: util.ArrayList[Row], cleanInstants: util.ArrayList[HoodieInstant]) = getSortedCleans(timeline)
-
- var rowCount = 0
-
- cleanInstants.asScala.takeWhile(_ => rowCount < limit).foreach { cleanInstant =>
- val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
-
- cleanMetadata.getPartitionMetadata.entrySet.asScala.takeWhile(_ => rowCount < limit).foreach { partitionMetadataEntry =>
- val partitionPath = partitionMetadataEntry.getKey
- val partitionMetadata = partitionMetadataEntry.getValue
-
- rows.add(Row(
- cleanInstant.requestedTime(),
- cleanInstant.getCompletionTime,
- cleanInstant.getAction,
- cleanMetadata.getStartCleanTime,
- partitionPath,
- partitionMetadata.getPolicy,
- partitionMetadata.getDeletePathPatterns.size(),
- partitionMetadata.getSuccessDeleteFiles.size(),
- partitionMetadata.getFailedDeleteFiles.size(),
- partitionMetadata.getIsPartitionDeleted,
- cleanMetadata.getTimeTakenInMillis,
- cleanMetadata.getTotalFilesDeleted
- ))
- rowCount += 1
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ allCleanInstants.foreach { cleanInstant =>
+ if (cleanInstant.getState == HoodieInstant.State.COMPLETED) {
+ Try {
+ val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
+
+ cleanMetadata.getPartitionMetadata.entrySet.asScala.foreach { partitionMetadataEntry =>
+ val partitionPath = partitionMetadataEntry.getKey
+ val partitionMetadata = partitionMetadataEntry.getValue
+
+ val row = Row(
+ cleanInstant.requestedTime(),
+ cleanInstant.getCompletionTime,
+ cleanInstant.getState.name(),
+ cleanInstant.getAction,
+ timelineType,
+ cleanMetadata.getStartCleanTime,
+ partitionPath,
+ partitionMetadata.getPolicy,
+ partitionMetadata.getDeletePathPatterns.size(),
+ partitionMetadata.getSuccessDeleteFiles.size(),
+ partitionMetadata.getFailedDeleteFiles.size(),
+ partitionMetadata.getIsPartitionDeleted,
+ cleanMetadata.getTimeTakenInMillis,
+ cleanMetadata.getTotalFilesDeleted,
+ cleanMetadata.getEarliestCommitToRetain,
+ cleanMetadata.getLastCompletedCommitTimestamp,
+ cleanMetadata.getVersion,
+ null,
+ null
+ )
+ allRows += row
+ }
+ if (cleanMetadata.getPartitionMetadata.isEmpty) {
Review Comment:
```suggestion
if (cleanMetadata.getPartitionMetadata.isEmpty) {
...
} else {
cleanMetadata.getPartitionMetadata.entrySet.asScala.foreach {
...
}
}
```
Which is easier to understood
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
+ * - `limit`: Optional. Maximum number of clean operations to return (default: 10, ignored if time range specified)
* - `showArchived`: Optional. Whether to include archived clean operations (default: false)
* - `filter`: Optional. SQL expression to filter results (default: empty string)
+ * - `startTime`: Optional. Start time for clean operations (format: yyyyMMddHHmmss, default: empty)
+ * - `endTime`: Optional. End time for clean operations (format: yyyyMMddHHmmss, default: empty)
*
* == Output Schema ==
* - `clean_time`: Timestamp when the clean operation was performed
* - `state_transition_time`: Time when the clean transitioned to completed state
+ * - `state`: Operation state (COMPLETED, INFLIGHT, REQUESTED)
* - `action`: The action type (always 'clean')
* - `start_clean_time`: When the clean operation started
+ * - `partition_path`: Partition path for the clean operation
+ * - `policy`: Clean policy used (KEEP_LATEST_COMMITS, etc.)
+ * - `delete_path_patterns`: Number of delete path patterns
+ * - `success_delete_files`: Number of successfully deleted files
+ * - `failed_delete_files`: Number of files that failed to delete
+ * - `is_partition_deleted`: Whether the entire partition was deleted
* - `time_taken_in_millis`: Duration of the clean operation in milliseconds
* - `total_files_deleted`: Total number of files deleted during the clean
* - `earliest_commit_to_retain`: The earliest commit that was retained
* - `last_completed_commit_timestamp`: The last completed commit at clean time
* - `version`: Version of the clean operation metadata
- * - Additional partition-level metadata columns when using `show_cleans_metadata`
+ * - `total_partitions_to_clean`: Total partitions to clean (for pending operations)
+ * - `total_partitions_to_delete`: Total partitions to delete (for pending operations)
*
- * == Error Handling ==
- * - Throws `IllegalArgumentException` for invalid filter expressions
- * - Throws `HoodieException` for table access issues
- * - Returns empty result set if no clean plans match the criteria
+ * == Data Availability by Operation State ==
+ * - **COMPLETED operations**: All execution and partition metadata fields are populated
+ * - **PENDING operations**: Plan fields are populated, execution fields are null (graceful handling)
*
* == Filter Support ==
* The `filter` parameter supports SQL expressions for filtering results.
*
* === Common Filter Examples ===
* {{{
+ * -- Show only completed operations (equivalent to old show_cleans)
+ * CALL show_cleans(
+ * table => 'my_table',
+ * filter => "state = 'COMPLETED'"
+ * )
+ *
+ * -- Show only pending operations (equivalent to old show_clean_plans)
+ * CALL show_cleans(
+ * table => 'my_table',
+ * filter => "state IN ('REQUESTED', 'INFLIGHT')"
Review Comment:
Are these states case sensitive?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
+ * - `limit`: Optional. Maximum number of clean operations to return (default: 10, ignored if time range specified)
* - `showArchived`: Optional. Whether to include archived clean operations (default: false)
* - `filter`: Optional. SQL expression to filter results (default: empty string)
+ * - `startTime`: Optional. Start time for clean operations (format: yyyyMMddHHmmss, default: empty)
Review Comment:
Interesting though. Do we have a way to let a Hudi commiter know this rule?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowClusteringProcedure.scala:
##########
@@ -53,32 +133,219 @@ class ShowClusteringProcedure extends BaseProcedure with ProcedureBuilder with S
val tableName = getArgValueOrDefault(args, PARAMETERS(0))
val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
- val showInvolvedPartitions = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
+ validateFilter(filter, outputType)
val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val clusteringInstants = metaClient.getActiveTimeline.getInstants.iterator().asScala
+
+ val activeResults = getCombinedClusteringsWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
+ val finalResults = if (showArchived) {
+ val archivedResults = getCombinedClusteringsWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
+ } else {
+ combinedResults.take(limit)
+ }
+ } else {
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
+ }
+
+ applyFilter(finalResults, filter, outputType)
+ }
+
+ private def getCombinedClusteringsWithPartitionMetadata(timeline: HoodieTimeline,
+ limit: Int,
+ metaClient: HoodieTableMetaClient,
+ startTime: String,
+ endTime: String,
+ timelineType: String): Seq[Row] = {
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ val filteredClusteringInstants = timeline.getInstants.iterator().asScala
.filter(p => ClusteringUtils.isClusteringOrReplaceCommitAction(p.getAction))
- .toSeq
- .sortBy(f => f.requestedTime)
- .reverse
- .take(limit)
-
- val clusteringPlans = clusteringInstants.map(instant =>
- ClusteringUtils.getClusteringPlan(metaClient, instant)
- ).filter(clusteringPlan => clusteringPlan.isPresent)
-
- if (showInvolvedPartitions) {
- clusteringPlans.map { p =>
- Row(p.get().getLeft.requestedTime, p.get().getRight.getInputGroups.size(),
- p.get().getLeft.getState.name(), HoodieCLIUtils.extractPartitions(p.get().getRight.getInputGroups.asScala.toSeq))
+ .filter { instant =>
+ val instantTime = instant.requestedTime()
+ val withinStartTime = if (startTime.nonEmpty) instantTime >= startTime else true
+ val withinEndTime = if (endTime.nonEmpty) instantTime <= endTime else true
+ withinStartTime && withinEndTime
}
+ .toSeq
+ val allClusteringInstants = if (startTime.nonEmpty && endTime.nonEmpty) {
+ filteredClusteringInstants.sortBy(_.requestedTime).reverse
} else {
- clusteringPlans.map { p =>
- Row(p.get().getLeft.requestedTime, p.get().getRight.getInputGroups.size(),
- p.get().getLeft.getState.name(), "*")
+ filteredClusteringInstants.sortBy(_.requestedTime).reverse.take(limit)
+ }
+
+ allClusteringInstants.foreach { clusteringInstant =>
+ if (clusteringInstant.getState == HoodieInstant.State.COMPLETED) {
+ scala.util.Try {
+ val commitMetadata = timeline.readCommitMetadata(clusteringInstant)
+ val planStats = Try {
+ val clusteringPlanOpt = ClusteringUtils.getClusteringPlan(metaClient, clusteringInstant)
+ if (clusteringPlanOpt.isPresent) {
+ Some(extractClusteringPlanStats(clusteringPlanOpt.get().getRight))
+ } else None
+ }.getOrElse(None)
+
+ commitMetadata.getPartitionToWriteStats.entrySet.asScala.foreach { partitionEntry =>
+ val partitionPath = partitionEntry.getKey
+ val writeStats = partitionEntry.getValue.asScala
+
+ val totalInputFiles = planStats.flatMap(s => s.partitionStats.get(partitionPath).map(_.inputFiles))
+ .getOrElse(0)
+
+ val totalOutputFiles = writeStats.size
+ val totalOutputSizeBytes = writeStats.map(_.getFileSizeInBytes).sum
+ val inputGroupSize = planStats.map(_.totalInputGroups).getOrElse(writeStats.size)
+
+ val row = Row(
+ clusteringInstant.requestedTime(),
+ clusteringInstant.getCompletionTime,
+ clusteringInstant.getState.name(),
+ clusteringInstant.getAction,
+ timelineType,
+ inputGroupSize,
+ partitionPath,
+ totalInputFiles,
+ totalOutputFiles,
+ totalOutputSizeBytes
+ )
+ allRows += row
+ }
+ if (commitMetadata.getPartitionToWriteStats.isEmpty) {
+ val totalInputFiles = planStats.map(_.totalInputFiles).map(Integer.valueOf).orNull
+ val totalOutputFiles = commitMetadata.fetchTotalFilesInsert + commitMetadata.fetchTotalFilesUpdated
+ val totalOutputSizeBytes = commitMetadata.fetchTotalBytesWritten
+ val inputGroupSize = planStats.map(_.totalInputGroups).map(Integer.valueOf).orNull
+
+ val row = Row(
+ clusteringInstant.requestedTime(),
+ clusteringInstant.getCompletionTime,
+ clusteringInstant.getState.name(),
+ clusteringInstant.getAction,
+ timelineType,
+ inputGroupSize,
+ null,
+ totalInputFiles,
+ totalOutputFiles.toInt,
+ totalOutputSizeBytes
+ )
+ allRows += row
+ }
+ }.recover {
+ case e: Exception =>
+ log.warn(s"Failed to read clustering metadata for instant ${clusteringInstant.requestedTime}: ${e.getMessage}")
+ val row = createErrorRowForCompletedWithPartition(clusteringInstant, timelineType)
+ allRows += row
+ }
+ } else {
+ scala.util.Try {
+ val clusteringPlanOpt = ClusteringUtils.getClusteringPlan(metaClient, clusteringInstant)
+ if (clusteringPlanOpt.isPresent) {
+ val clusteringPlan = clusteringPlanOpt.get()
+ val plan = clusteringPlan.getRight
+ val stats = extractClusteringPlanStats(plan)
+ stats.involvedPartitions.foreach { partitionPath =>
+ val partitionStat = stats.partitionStats.get(partitionPath)
+ val row = Row(
+ clusteringInstant.requestedTime(),
+ null, // state_transition_time - not available for pending
+ clusteringInstant.getState.name(),
+ clusteringInstant.getAction,
+ timelineType,
+ stats.totalInputGroups, // input_group_size - total across all partitions
+ partitionPath,
+ partitionStat.map(_.inputFiles).getOrElse(0), // input_files_count - per partition
+ partitionStat.map(_.outputGroups).getOrElse(0), // output_files_count - per partition
+ 0L // output_size_bytes - not available for pending operations
+ )
+ allRows += row
+ }
+ } else {
+ // for inflight requests without a clustering plan
+ val row = createErrorRowForPendingWithPartition(clusteringInstant, timelineType)
+ allRows += row
+ }
+ }.recover {
+ case e: Exception =>
+ log.warn(s"Failed to read clustering plan for instant ${clusteringInstant.requestedTime}: ${e.getMessage}")
+ val row = createErrorRowForPendingWithPartition(clusteringInstant, timelineType)
+ allRows += row
+ }
}
}
+ allRows.toSeq
+ }
Review Comment:
Do we have a way to break this 100 lines of code into smaller functions for better readability?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowClusteringProcedure.scala:
##########
@@ -53,32 +133,219 @@ class ShowClusteringProcedure extends BaseProcedure with ProcedureBuilder with S
val tableName = getArgValueOrDefault(args, PARAMETERS(0))
val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
- val showInvolvedPartitions = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
+ validateFilter(filter, outputType)
val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val clusteringInstants = metaClient.getActiveTimeline.getInstants.iterator().asScala
+
+ val activeResults = getCombinedClusteringsWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
+ val finalResults = if (showArchived) {
+ val archivedResults = getCombinedClusteringsWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
+ } else {
+ combinedResults.take(limit)
+ }
+ } else {
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
+ }
+
+ applyFilter(finalResults, filter, outputType)
+ }
+
+ private def getCombinedClusteringsWithPartitionMetadata(timeline: HoodieTimeline,
+ limit: Int,
+ metaClient: HoodieTableMetaClient,
+ startTime: String,
+ endTime: String,
+ timelineType: String): Seq[Row] = {
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ val filteredClusteringInstants = timeline.getInstants.iterator().asScala
.filter(p => ClusteringUtils.isClusteringOrReplaceCommitAction(p.getAction))
Review Comment:
The javadoc of `ClusteringUtils.isClusteringOrReplaceCommitAction` is outdated. Please update it as well.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitsProcedure.scala:
##########
@@ -17,150 +17,492 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.HoodieCLIUtils
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.common.model.HoodieWriteStat
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.ClusteringUtils
+import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
+import scala.util.Try
-class ShowCommitsProcedure(includeExtraMetadata: Boolean) extends BaseProcedure with ProcedureBuilder {
- var sortByFieldParameter: ProcedureParameter = _
-
+/**
+ * Spark SQL procedure to show all commit operations for a Hudi table.
+ *
+ * This procedure provides a comprehensive view of Hudi commit operations, displaying both
+ * completed commits with execution metadata and pending commit plans.
+ * For pending operations, execution-specific fields are gracefully set to null.
+ *
+ * == Parameters ==
+ * - `table`: Required. The name of the Hudi table to query
+ * - `path`: Optional. The path of the Hudi table (alternative to table name)
+ * - `limit`: Optional. Maximum number of commit operations to return (default: 20)
+ * - `showArchived`: Optional. Whether to include archived commit operations (default: false)
+ * - `filter`: Optional. SQL expression to filter results (default: empty string)
+ * - `startTime`: Optional. Start time for commits (format: yyyyMMddHHmmss, default: empty)
+ * - `endTime`: Optional. End time for commits (format: yyyyMMddHHmmss, default: empty)
+ *
+ * == Output Schema ==
+ * - `commit_time`: Timestamp when the commit operation was initiated
+ * - `state_transition_time`: Time when the commit transitioned to completed state (null for pending)
+ * - `state`: Operation state (COMPLETED, INFLIGHT, REQUESTED)
+ * - `action`: The action type (commit/deltacommit/replacecommit)
+ * - `timeline_type`: Source timeline (ACTIVE, ARCHIVED)
+ * - `partition_path`: Partition path for the commit operation
+ * - `file_id`: ID of the file
+ * - `previous_commit`: Previous commit time for the file
+ * - `num_writes`: Number of records written
+ * - `num_inserts`: Number of records inserted
+ * - `num_deletes`: Number of records deleted
+ * - `num_update_writes`: Number of records updated
+ * - `total_log_blocks`: Number of log blocks
+ * - `total_corrupt_log_blocks`: Number of corrupt log blocks
+ * - `total_rollback_blocks`: Number of rollback blocks
+ * - `total_log_records`: Number of log records
+ * - `total_updated_records_compacted`: Number of updated records compacted
+ * - `total_files_added`: Number of files added
+ * - `total_files_updated`: Number of files updated
+ * - `total_records_written`: Number of records written
+ * - `total_records_updated`: Number of records updated
+ * - `total_bytes_written`: Total bytes written
+ * - `total_errors`: Number of errors encountered
+ * - `file_size`: Size of the file in bytes
+ * - `avg_record_size`: Average size of each record in bytes
+ * - `extra_metadata`: Additional metadata from the commit
+ */
+class ShowCommitsProcedure extends BaseProcedure with ProcedureBuilder with Logging {
private val PARAMETERS = Array[ProcedureParameter](
- ProcedureParameter.required(0, "table", DataTypes.StringType),
- ProcedureParameter.optional(1, "limit", DataTypes.IntegerType, 10)
+ ProcedureParameter.optional(0, "table", DataTypes.StringType),
+ ProcedureParameter.optional(1, "path", DataTypes.StringType),
+ ProcedureParameter.optional(2, "limit", DataTypes.IntegerType, 20),
+ ProcedureParameter.optional(3, "showArchived", DataTypes.BooleanType, false),
+ ProcedureParameter.optional(4, "showFiles", DataTypes.BooleanType, false),
+ ProcedureParameter.optional(5, "filter", DataTypes.StringType, ""),
+ ProcedureParameter.optional(6, "startTime", DataTypes.StringType, ""),
+ ProcedureParameter.optional(7, "endTime", DataTypes.StringType, "")
)
private val OUTPUT_TYPE = new StructType(Array[StructField](
StructField("commit_time", DataTypes.StringType, nullable = true, Metadata.empty),
StructField("state_transition_time", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("state", DataTypes.StringType, nullable = true, Metadata.empty),
StructField("action", DataTypes.StringType, nullable = true, Metadata.empty),
- StructField("total_bytes_written", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_files_added", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_files_updated", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_partitions_written", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_records_written", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_update_records_written", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_errors", DataTypes.LongType, nullable = true, Metadata.empty)
- ))
-
- private val METADATA_OUTPUT_TYPE = new StructType(Array[StructField](
- StructField("commit_time", DataTypes.StringType, nullable = true, Metadata.empty),
- StructField("state_transition_time", DataTypes.StringType, nullable = true, Metadata.empty),
- StructField("action", DataTypes.StringType, nullable = true, Metadata.empty),
- StructField("partition", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("timeline_type", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("partition_path", DataTypes.StringType, nullable = true, Metadata.empty),
StructField("file_id", DataTypes.StringType, nullable = true, Metadata.empty),
StructField("previous_commit", DataTypes.StringType, nullable = true, Metadata.empty),
StructField("num_writes", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("num_inserts", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("num_deletes", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("num_update_writes", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_errors", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("total_log_blocks", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("total_corrupt_log_blocks", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("total_rollback_blocks", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("total_log_records", DataTypes.LongType, nullable = true, Metadata.empty),
StructField("total_updated_records_compacted", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_bytes_written", DataTypes.LongType, nullable = true, Metadata.empty)
+ StructField("total_files_added", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("total_files_updated", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("total_records_written", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("total_records_updated", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("total_bytes_written", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("total_errors", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("file_size", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("avg_record_size", DataTypes.LongType, nullable = true, Metadata.empty),
+ StructField("extra_metadata", DataTypes.StringType, nullable = true, Metadata.empty)
))
def parameters: Array[ProcedureParameter] = PARAMETERS
- def outputType: StructType = if (includeExtraMetadata) METADATA_OUTPUT_TYPE else OUTPUT_TYPE
+ def outputType: StructType = OUTPUT_TYPE
override def call(args: ProcedureArgs): Seq[Row] = {
super.checkArgs(PARAMETERS, args)
- val table = getArgValueOrDefault(args, PARAMETERS(0)).get.asInstanceOf[String]
- val limit = getArgValueOrDefault(args, PARAMETERS(1)).get.asInstanceOf[Int]
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
+ val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val showFiles = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(7)).get.asInstanceOf[String]
+
+ validateFilter(filter, outputType)
- val hoodieCatalogTable = HoodieCLIUtils.getHoodieCatalogTable(sparkSession, table)
- val basePath = hoodieCatalogTable.tableLocation
+ val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val activeTimeline = metaClient.getActiveTimeline
- if (includeExtraMetadata) {
- getCommitsWithMetadata(activeTimeline, limit)
+ val activeResults = getCommitsWithPartitionMetadata(
+ metaClient.getActiveTimeline, limit, showFiles, startTime, endTime, metaClient, "ACTIVE")
+
+ val finalResults = if (showArchived) {
+ val archivedResults = getCommitsWithPartitionMetadata(metaClient.getArchivedTimeline, limit, showFiles, startTime, endTime, metaClient, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
+ } else {
+ combinedResults.take(limit)
+ }
} else {
- getCommits(activeTimeline, limit)
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
}
+ applyFilter(finalResults, filter, outputType)
}
- override def build: Procedure = new ShowCommitsProcedure(includeExtraMetadata)
-
- private def getCommitsWithMetadata(timeline: HoodieTimeline,
- limit: Int): Seq[Row] = {
- import scala.collection.JavaConverters._
-
- val (rows: util.ArrayList[Row], newCommits: util.ArrayList[HoodieInstant]) = getSortCommits(timeline)
- val layout = TimelineLayout.fromVersion(timeline.getTimelineLayoutVersion)
- for (i <- 0 until newCommits.size) {
- val commit = newCommits.get(i)
- val commitMetadata = timeline.readCommitMetadata(commit)
- for (partitionWriteStat <- commitMetadata.getPartitionToWriteStats.entrySet.asScala) {
- for (hoodieWriteStat <- partitionWriteStat.getValue.asScala) {
- rows.add(Row(
- commit.requestedTime, commit.getCompletionTime, commit.getAction, hoodieWriteStat.getPartitionPath,
- hoodieWriteStat.getFileId, hoodieWriteStat.getPrevCommit, hoodieWriteStat.getNumWrites,
- hoodieWriteStat.getNumInserts, hoodieWriteStat.getNumDeletes, hoodieWriteStat.getNumUpdateWrites,
- hoodieWriteStat.getTotalWriteErrors, hoodieWriteStat.getTotalLogBlocks, hoodieWriteStat.getTotalCorruptLogBlock,
- hoodieWriteStat.getTotalRollbackBlocks, hoodieWriteStat.getTotalLogRecords,
- hoodieWriteStat.getTotalUpdatedRecordsCompacted, hoodieWriteStat.getTotalWriteBytes))
+ private def getCommitsWithPartitionMetadata(timeline: HoodieTimeline,
+ limit: Int,
+ showFiles: Boolean,
+ startTime: String,
+ endTime: String,
+ metaClient: HoodieTableMetaClient,
+ timelineType: String): Seq[Row] = {
+
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ val commitsTimeline = timeline.getCommitsTimeline
+
+ val filteredCommitInstants = commitsTimeline.getInstants.asScala
+ .filter { instant =>
+ val instantTime = instant.requestedTime()
+ val withinStartTime = if (startTime.nonEmpty) instantTime >= startTime else true
+ val withinEndTime = if (endTime.nonEmpty) instantTime <= endTime else true
+ withinStartTime && withinEndTime
+ }
+ .toSeq
+
+ val allCommitInstants = if (startTime.nonEmpty && endTime.nonEmpty) {
+ filteredCommitInstants.sortBy(_.requestedTime).reverse
+ } else {
+ filteredCommitInstants.sortBy(_.requestedTime).reverse.take(limit)
+ }
+
+ allCommitInstants.foreach { commitInstant =>
+ if (commitInstant.getState == HoodieInstant.State.COMPLETED) {
+ Try {
+ val commitMetadata = if (ClusteringUtils.isClusteringOrReplaceCommitAction(commitInstant.getAction)) {
+ timeline.readReplaceCommitMetadata(commitInstant)
+ } else {
+ timeline.readCommitMetadata(commitInstant)
+ }
+
+ val extraMetadata = Option(commitMetadata.getExtraMetadata)
+ .map(_.asScala.map { case (k, v) => s"$k=$v" }.mkString(", "))
+ .orNull
+
+ if (showFiles) {
+ for (entry <- commitMetadata.getPartitionToWriteStats.entrySet.asScala) {
+ val partitionPath = entry.getKey
+ val stats = entry.getValue.asScala
+
+ for (stat <- stats) {
+ val fileId = stat.getFileId
+ val prevCommit = stat.getPrevCommit
+ val fileSize = stat.getFileSizeInBytes
+
+ val isInsert = prevCommit == HoodieWriteStat.NULL_COMMIT
+ val totalFilesAdded = if (isInsert) 1L else 0L
+ val totalFilesUpdated = if (!isInsert) 1L else 0L
+
+ val numWrites = stat.getNumWrites
+ val numInserts = stat.getNumInserts
+ val numDeletes = stat.getNumDeletes
+ val numUpdateWrites = stat.getNumUpdateWrites
+ val totalLogBlocks = stat.getTotalLogBlocks
+ val totalCorruptLogBlocks = stat.getTotalCorruptLogBlock
+ val totalRollbackBlocks = stat.getTotalRollbackBlocks
+ val totalLogRecords = stat.getTotalLogRecords
+ val totalUpdatedRecordsCompacted = stat.getTotalUpdatedRecordsCompacted
+ val totalBytesWritten = stat.getTotalWriteBytes
+ val totalErrors = stat.getTotalWriteErrors
+ val avgRecordSize = if (numWrites > 0) totalBytesWritten / numWrites else 0L
+
+ allRows += Row(
+ commitInstant.requestedTime(),
+ commitInstant.getCompletionTime,
+ commitInstant.getState.name(),
+ commitInstant.getAction,
+ timelineType,
+ partitionPath,
+ fileId,
+ prevCommit,
+ numWrites,
+ numInserts,
+ numDeletes,
+ numUpdateWrites,
+ totalLogBlocks,
+ totalCorruptLogBlocks,
+ totalRollbackBlocks,
+ totalLogRecords,
+ totalUpdatedRecordsCompacted,
+ totalFilesAdded,
+ totalFilesUpdated,
+ numWrites,
+ numUpdateWrites,
+ totalBytesWritten,
+ totalErrors,
+ fileSize,
+ avgRecordSize,
+ extraMetadata
+ )
+ }
+ }
+ } else {
+ for (entry <- commitMetadata.getPartitionToWriteStats.entrySet.asScala) {
+ val partitionPath = entry.getKey
+ val stats = entry.getValue.asScala
+
+ var totalFilesAdded = 0L
+ var totalFilesUpdated = 0L
+ var totalRecordsWritten = 0L
+ var totalRecordsUpdated = 0L
+ var totalBytesWritten = 0L
+ var totalErrors = 0L
+ var totalNumWrites = 0L
+ var totalNumInserts = 0L
+ var totalNumDeletes = 0L
+ var totalNumUpdateWrites = 0L
+ var totalLogBlocks = 0L
+ var totalCorruptLogBlocks = 0L
+ var totalRollbackBlocks = 0L
+ var totalLogRecords = 0L
+ var totalUpdatedRecordsCompacted = 0L
+ var totalFileSize = 0L
+
+ for (stat <- stats) {
+ if (stat.getPrevCommit == HoodieWriteStat.NULL_COMMIT) {
+ totalFilesAdded += 1
+ } else {
+ totalFilesUpdated += 1
+ }
+ totalNumWrites += stat.getNumWrites
+ totalNumInserts += stat.getNumInserts
+ totalNumDeletes += stat.getNumDeletes
+ totalNumUpdateWrites += stat.getNumUpdateWrites
+ totalLogBlocks += stat.getTotalLogBlocks
+ totalCorruptLogBlocks += stat.getTotalCorruptLogBlock
+ totalRollbackBlocks += stat.getTotalRollbackBlocks
+ totalLogRecords += stat.getTotalLogRecords
+ totalUpdatedRecordsCompacted += stat.getTotalUpdatedRecordsCompacted
+
+ totalRecordsWritten += stat.getNumWrites
+ totalRecordsUpdated += stat.getNumUpdateWrites
+ totalBytesWritten += stat.getTotalWriteBytes
+ totalErrors += stat.getTotalWriteErrors
+ totalFileSize += stat.getFileSizeInBytes
+ }
+
+ val avgRecordSize = if (totalNumWrites > 0) totalBytesWritten / totalNumWrites else 0L
+
+ allRows += Row(
+ commitInstant.requestedTime(),
+ commitInstant.getCompletionTime,
+ commitInstant.getState.name(),
+ commitInstant.getAction,
+ timelineType,
+ partitionPath,
+ "*",
+ null,
+ totalNumWrites,
+ totalNumInserts,
+ totalNumDeletes,
+ totalNumUpdateWrites,
+ totalLogBlocks,
+ totalCorruptLogBlocks,
+ totalRollbackBlocks,
+ totalLogRecords,
+ totalUpdatedRecordsCompacted,
+ totalFilesAdded,
+ totalFilesUpdated,
+ totalRecordsWritten,
+ totalRecordsUpdated,
+ totalBytesWritten,
+ totalErrors,
+ totalFileSize,
+ avgRecordSize,
+ extraMetadata
+ )
+ }
+
+ if (commitMetadata.getPartitionToWriteStats.isEmpty) {
+ val totalFilesAdded = commitMetadata.fetchTotalFilesInsert
+ val totalFilesUpdated = commitMetadata.fetchTotalFilesUpdated
+ val totalRecordsWritten = commitMetadata.fetchTotalRecordsWritten
+ val totalRecordsUpdated = commitMetadata.fetchTotalUpdateRecordsWritten
+ val totalBytesWritten = commitMetadata.fetchTotalBytesWritten
+ val totalErrors = commitMetadata.fetchTotalWriteErrors
+ val avgRecordSize = if (totalRecordsWritten > 0) totalBytesWritten / totalRecordsWritten else 0L
+
+ allRows += Row(
+ commitInstant.requestedTime(),
+ commitInstant.getCompletionTime,
+ commitInstant.getState.name(),
+ commitInstant.getAction,
+ timelineType,
+ null, // partitionPath
+ "*", // fileId
+ null, // prevCommit
+ totalRecordsWritten, // numWrites
+ totalFilesAdded, // numInserts
+ null, // numDeletes
+ totalRecordsUpdated, // numUpdateWrites
+ null, // totalLogBlocks
+ null, // totalCorruptLogBlocks
+ null, // totalRollbackBlocks
+ null, // totalLogRecords
+ null, // totalUpdatedRecordsCompacted
+ totalFilesAdded,
+ totalFilesUpdated,
+ totalRecordsWritten,
+ totalRecordsUpdated,
+ totalBytesWritten,
+ totalErrors,
+ null, // fileSize
+ avgRecordSize, // avgRecordSize
+ extraMetadata
+ )
+ }
+ }
+ }.recover {
+ case e: Exception =>
+ log.warn(s"Failed to read commit metadata for instant ${commitInstant.requestedTime}: ${e.getMessage}")
+ val row = createErrorRow(commitInstant)
+ allRows += row
+ }
+ } else {
+ val commitStats = extractCommitPlanStats(commitInstant, metaClient)
+ commitStats.involvedPartitions.foreach { partitionPath =>
+ allRows += Row(
+ commitInstant.requestedTime(),
+ null, // state_transition_time
+ commitInstant.getState.name(),
+ commitInstant.getAction,
+ timelineType,
+ partitionPath,
+ "*", // fileId
+ null, // prevCommit
+ 0L, // numWrites
+ 0L, // numInserts
+ 0L, // numDeletes
+ 0L, // numUpdateWrites
+ 0L, // totalLogBlocks
+ 0L, // totalCorruptLogBlocks
+ 0L, // totalRollbackBlocks
+ 0L, // totalLogRecords
+ 0L, // totalUpdatedRecordsCompacted
+ 0L, // totalFilesAdded
+ 0L, // totalFilesUpdated
+ 0L, // totalRecordsWritten
+ 0L, // totalRecordsUpdated
+ 0L, // totalBytesWritten
+ 0L, // totalErrors
+ 0L, // fileSize
+ 0L, // avgRecordSize
+ commitStats.extraMetadata
+ )
Review Comment:
Break into smaller functions will good for reviewer.
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316561090
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCompactionProcedure.scala:
##########
@@ -57,24 +130,185 @@ class ShowCompactionProcedure extends BaseProcedure with ProcedureBuilder with S
val tableName = getArgValueOrDefault(args, PARAMETERS(0))
val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
+ validateFilter(filter, outputType)
val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- assert(metaClient.getTableType == HoodieTableType.MERGE_ON_READ,
- s"Cannot show compaction on a Non Merge On Read table.")
- val compactionInstants = metaClient.getActiveTimeline.getInstants.iterator().asScala
+ if (metaClient.getTableType != HoodieTableType.MERGE_ON_READ) {
+ throw new IllegalArgumentException("Cannot show compaction on a Non Merge On Read table.")
+ }
+
+ val activeResults = getCombinedCompactionsWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
+ val finalResults = if (showArchived) {
+ val archivedResults = getCombinedCompactionsWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
+ } else {
+ combinedResults.take(limit)
+ }
+ } else {
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
+ }
+ applyFilter(finalResults, filter, outputType)
+ }
+
+ private def getCombinedCompactionsWithPartitionMetadata(timeline: HoodieTimeline,
+ limit: Int,
+ metaClient: HoodieTableMetaClient,
+ startTime: String,
+ endTime: String,
+ timelineType: String): Seq[Row] = {
+ import scala.collection.mutable.ListBuffer
+
+ val allRows = ListBuffer[Row]()
+
+ val filteredCompactionInstants = timeline.getInstants.iterator().asScala
.filter(p => p.getAction == HoodieTimeline.COMPACTION_ACTION || p.getAction == HoodieTimeline.COMMIT_ACTION)
+ .filter { instant =>
+ val instantTime = instant.requestedTime()
+ val withinStartTime = if (startTime.nonEmpty) instantTime >= startTime else true
+ val withinEndTime = if (endTime.nonEmpty) instantTime <= endTime else true
+ withinStartTime && withinEndTime
+ }
.toSeq
- .sortBy(f => f.requestedTime)
- .reverse
- .take(limit)
-
- compactionInstants.map(instant =>
- (instant, CompactionUtils.getCompactionPlan(metaClient, instant.requestedTime))
- ).map { case (instant, plan) =>
- Row(instant.requestedTime, plan.getOperations.size(), instant.getState.name())
+ val allCompactionInstants = if (startTime.nonEmpty && endTime.nonEmpty) {
+ filteredCompactionInstants.sortBy(_.requestedTime).reverse
+ } else {
+ filteredCompactionInstants.sortBy(_.requestedTime).reverse.take(limit)
+ }
+
+ allCompactionInstants.foreach { compactionInstant =>
+ if (compactionInstant.getState == HoodieInstant.State.COMPLETED) {
+ Try {
+ val compactionMetadata = timeline.readCommitMetadata(compactionInstant)
+ compactionMetadata.getPartitionToWriteStats.entrySet.asScala.foreach { partitionEntry =>
+ val partitionPath = partitionEntry.getKey
+ val writeStats = partitionEntry.getValue.asScala
+ val operationSize = Try {
+ val compactionPlan = CompactionUtils.getCompactionPlan(metaClient, compactionInstant.requestedTime)
+ compactionPlan.getOperations.size()
+ }.getOrElse(null)
+
+ writeStats.foreach { writeStat =>
+ val row = Row(
+ compactionInstant.requestedTime(),
+ compactionInstant.getCompletionTime,
+ compactionInstant.getState.name(),
+ compactionInstant.getAction,
+ timelineType,
+ operationSize,
+ partitionPath,
+ writeStat.getTotalLogFilesCompacted,
+ writeStat.getTotalUpdatedRecordsCompacted,
+ writeStat.getTotalLogSizeCompacted,
+ writeStat.getTotalWriteBytes
+ )
+ allRows += row
+ }
+ }
+ if (compactionMetadata.getPartitionToWriteStats.isEmpty) {
+ val operationSize = Try {
+ val compactionPlan = CompactionUtils.getCompactionPlan(metaClient, compactionInstant.requestedTime)
+ compactionPlan.getOperations.size()
+ }.getOrElse(null)
+
+ val totalLogFilesCompacted = compactionMetadata.getTotalLogFilesCompacted
+ val totalUpdatedRecordsCompacted = compactionMetadata.fetchTotalUpdateRecordsWritten
+ val totalLogSizeCompacted = compactionMetadata.getTotalLogFilesSize
+ val totalWriteBytes = compactionMetadata.fetchTotalBytesWritten
+
+ val row = Row(
+ compactionInstant.requestedTime(),
+ compactionInstant.getCompletionTime,
+ compactionInstant.getState.name(),
+ compactionInstant.getAction,
+ timelineType,
+ operationSize,
+ null,
+ totalLogFilesCompacted,
+ totalUpdatedRecordsCompacted,
+ totalLogSizeCompacted,
+ totalWriteBytes
+ )
+ allRows += row
+ }
+ }.recover {
+ case e: Exception =>
+ log.warn(s"Failed to read compaction metadata for instant ${compactionInstant.requestedTime}: ${e.getMessage}")
+ val row = createErrorRowForCompletedWithPartition(compactionInstant, timelineType)
+ allRows += row
+ }
+ } else {
+ Try {
+ val compactionPlan = CompactionUtils.getCompactionPlan(metaClient, compactionInstant.requestedTime)
+ val stats = extractCompactionPlanStats(compactionPlan)
+ val operations = compactionPlan.getOperations.asScala
+ stats.involvedPartitions.foreach { partitionPath =>
+ val partitionOps = operations.filter(_.getPartitionPath == partitionPath)
+ val partitionDeltaFiles = partitionOps.map(_.getDeltaFilePaths.size()).sum
+ val row = Row(
+ compactionInstant.requestedTime(),
+ null,
+ compactionInstant.getState.name(),
+ compactionInstant.getAction,
+ timelineType,
+ partitionOps.size,
+ partitionPath,
+ partitionDeltaFiles.toLong,
+ 0L,
+ 0L,
+ 0L
+ )
+ allRows += row
+ }
+ }.recover {
+ case e: Exception =>
+ log.warn(s"Failed to read compaction plan for instant ${compactionInstant.requestedTime}: ${e.getMessage}")
+ val row = createErrorRowForPendingWithPartition(compactionInstant, timelineType)
+ allRows += row
+ }
+ }
}
+ allRows.toSeq
+ }
+
+ private def extractCompactionPlanStats(compactionPlan: org.apache.hudi.avro.model.HoodieCompactionPlan): CompactionPlanStatistics = {
+ val operations = Option(compactionPlan.getOperations)
+ .map(_.asScala.toSeq)
+ .getOrElse(Seq.empty)
+
+ val involvedPartitions = operations.map(_.getPartitionPath).toSet
+
+ CompactionPlanStatistics(
+ involvedPartitions = involvedPartitions
+ )
+ }
+
+ private case class CompactionPlanStatistics(involvedPartitions: Set[String])
Review Comment:
The name of this class indicates bigger scope than it has.
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3237884697
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b4dffdd4e409045edd57e38b0376f0cf98452a2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239)
* 7fe006c3f887de640c9fe7f35d1dcc880567b40c Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231316590
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
}, {
"hash" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2b500858557a3e3d917e42959d66237dd1d4599c Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191)
* 9c6fa69d9084d835fab01d1bffd42d5a1b3e453a 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238335555
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ec3e1cff11c11ea9bafed719a2598bcba95d0da2 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231857083
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8191",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
}, {
"hash" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194",
"triggerID" : "9c6fa69d9084d835fab01d1bffd42d5a1b3e453a",
"triggerType" : "PUSH"
}, {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9c6fa69d9084d835fab01d1bffd42d5a1b3e453a Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8194)
* d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238243202
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7fe006c3f887de640c9fe7f35d1dcc880567b40c Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271)
* ec3e1cff11c11ea9bafed719a2598bcba95d0da2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238734565
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6e153692e2f3f03236bd9951882ac0d170401e8f Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238700908
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ec3e1cff11c11ea9bafed719a2598bcba95d0da2 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274)
* 6e153692e2f3f03236bd9951882ac0d170401e8f Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "linliu-code (via GitHub)" <gi...@apache.org>.
linliu-code commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316550502
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCompactionProcedure.scala:
##########
@@ -57,24 +130,185 @@ class ShowCompactionProcedure extends BaseProcedure with ProcedureBuilder with S
val tableName = getArgValueOrDefault(args, PARAMETERS(0))
val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
+ validateFilter(filter, outputType)
val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- assert(metaClient.getTableType == HoodieTableType.MERGE_ON_READ,
- s"Cannot show compaction on a Non Merge On Read table.")
- val compactionInstants = metaClient.getActiveTimeline.getInstants.iterator().asScala
+ if (metaClient.getTableType != HoodieTableType.MERGE_ON_READ) {
+ throw new IllegalArgumentException("Cannot show compaction on a Non Merge On Read table.")
+ }
+
+ val activeResults = getCombinedCompactionsWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
+ val finalResults = if (showArchived) {
+ val archivedResults = getCombinedCompactionsWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
+ } else {
+ combinedResults.take(limit)
+ }
+ } else {
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
+ }
+ applyFilter(finalResults, filter, outputType)
+ }
+
+ private def getCombinedCompactionsWithPartitionMetadata(timeline: HoodieTimeline,
+ limit: Int,
+ metaClient: HoodieTableMetaClient,
+ startTime: String,
+ endTime: String,
+ timelineType: String): Seq[Row] = {
+ import scala.collection.mutable.ListBuffer
+
+ val allRows = ListBuffer[Row]()
+
+ val filteredCompactionInstants = timeline.getInstants.iterator().asScala
.filter(p => p.getAction == HoodieTimeline.COMPACTION_ACTION || p.getAction == HoodieTimeline.COMMIT_ACTION)
Review Comment:
How do we differentiate the compaction commit and regular ingestion commits for COW tables?
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3247492126
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
}, {
"hash" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277",
"triggerID" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"triggerType" : "PUSH"
}, {
"hash" : "5d640dd2f82b6a5213380bac343a79e8dade4156",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8310",
"triggerID" : "5d640dd2f82b6a5213380bac343a79e8dade4156",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60e8bead700c96e9848131a3c621d03e1bbaf06 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277)
* 5d640dd2f82b6a5213380bac343a79e8dade4156 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8310)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238039235
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7fe006c3f887de640c9fe7f35d1dcc880567b40c Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2308556278
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
+ * - `limit`: Optional. Maximum number of clean operations to return (default: 10, ignored if time range specified)
* - `showArchived`: Optional. Whether to include archived clean operations (default: false)
* - `filter`: Optional. SQL expression to filter results (default: empty string)
+ * - `startTime`: Optional. Start time for clean operations (format: yyyyMMddHHmmss, default: empty)
Review Comment:
Using camel cased style for all the input parameters for show Procedures. so startTime would be better than start_time to be consistent
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2308551108
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
Review Comment:
Ah, my bad, its optional (I will change it in the description), either of the table name or path is required, we do that verification in the `getBaseFile` method
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3234546034
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200)
* fa38e71eb89ffe5571e9f64cc45407e6319c62b2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316905920
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowClusteringProcedure.scala:
##########
@@ -53,32 +133,219 @@ class ShowClusteringProcedure extends BaseProcedure with ProcedureBuilder with S
val tableName = getArgValueOrDefault(args, PARAMETERS(0))
val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
- val showInvolvedPartitions = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
+ validateFilter(filter, outputType)
val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val clusteringInstants = metaClient.getActiveTimeline.getInstants.iterator().asScala
+
+ val activeResults = getCombinedClusteringsWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
+ val finalResults = if (showArchived) {
+ val archivedResults = getCombinedClusteringsWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
+ } else {
+ combinedResults.take(limit)
+ }
+ } else {
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
+ }
+
+ applyFilter(finalResults, filter, outputType)
+ }
+
+ private def getCombinedClusteringsWithPartitionMetadata(timeline: HoodieTimeline,
+ limit: Int,
+ metaClient: HoodieTableMetaClient,
+ startTime: String,
+ endTime: String,
+ timelineType: String): Seq[Row] = {
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ val filteredClusteringInstants = timeline.getInstants.iterator().asScala
.filter(p => ClusteringUtils.isClusteringOrReplaceCommitAction(p.getAction))
Review Comment:
Okay
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3247490171
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
}, {
"hash" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277",
"triggerID" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"triggerType" : "PUSH"
}, {
"hash" : "5d640dd2f82b6a5213380bac343a79e8dade4156",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5d640dd2f82b6a5213380bac343a79e8dade4156",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60e8bead700c96e9848131a3c621d03e1bbaf06 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277)
* 5d640dd2f82b6a5213380bac343a79e8dade4156 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3247539826
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
}, {
"hash" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8277",
"triggerID" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"triggerType" : "PUSH"
}, {
"hash" : "5d640dd2f82b6a5213380bac343a79e8dade4156",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8310",
"triggerID" : "5d640dd2f82b6a5213380bac343a79e8dade4156",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d640dd2f82b6a5213380bac343a79e8dade4156 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8310)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316721561
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
+ * - `limit`: Optional. Maximum number of clean operations to return (default: 10, ignored if time range specified)
* - `showArchived`: Optional. Whether to include archived clean operations (default: false)
* - `filter`: Optional. SQL expression to filter results (default: empty string)
+ * - `startTime`: Optional. Start time for clean operations (format: yyyyMMddHHmmss, default: empty)
+ * - `endTime`: Optional. End time for clean operations (format: yyyyMMddHHmmss, default: empty)
*
* == Output Schema ==
* - `clean_time`: Timestamp when the clean operation was performed
* - `state_transition_time`: Time when the clean transitioned to completed state
+ * - `state`: Operation state (COMPLETED, INFLIGHT, REQUESTED)
* - `action`: The action type (always 'clean')
* - `start_clean_time`: When the clean operation started
+ * - `partition_path`: Partition path for the clean operation
+ * - `policy`: Clean policy used (KEEP_LATEST_COMMITS, etc.)
+ * - `delete_path_patterns`: Number of delete path patterns
+ * - `success_delete_files`: Number of successfully deleted files
+ * - `failed_delete_files`: Number of files that failed to delete
+ * - `is_partition_deleted`: Whether the entire partition was deleted
* - `time_taken_in_millis`: Duration of the clean operation in milliseconds
* - `total_files_deleted`: Total number of files deleted during the clean
* - `earliest_commit_to_retain`: The earliest commit that was retained
* - `last_completed_commit_timestamp`: The last completed commit at clean time
* - `version`: Version of the clean operation metadata
- * - Additional partition-level metadata columns when using `show_cleans_metadata`
+ * - `total_partitions_to_clean`: Total partitions to clean (for pending operations)
+ * - `total_partitions_to_delete`: Total partitions to delete (for pending operations)
*
- * == Error Handling ==
- * - Throws `IllegalArgumentException` for invalid filter expressions
- * - Throws `HoodieException` for table access issues
- * - Returns empty result set if no clean plans match the criteria
+ * == Data Availability by Operation State ==
+ * - **COMPLETED operations**: All execution and partition metadata fields are populated
+ * - **PENDING operations**: Plan fields are populated, execution fields are null (graceful handling)
*
* == Filter Support ==
* The `filter` parameter supports SQL expressions for filtering results.
*
* === Common Filter Examples ===
* {{{
+ * -- Show only completed operations (equivalent to old show_cleans)
+ * CALL show_cleans(
+ * table => 'my_table',
+ * filter => "state = 'COMPLETED'"
+ * )
+ *
+ * -- Show only pending operations (equivalent to old show_clean_plans)
+ * CALL show_cleans(
+ * table => 'my_table',
+ * filter => "state IN ('REQUESTED', 'INFLIGHT')"
Review Comment:
these are sql like filters, we support lower, upper functions to support case sensitivity.
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2316905452
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -138,128 +171,245 @@ class ShowCleansProcedure(includePartitionMetadata: Boolean) extends BaseProcedu
StructField("failed_delete_files", DataTypes.IntegerType, nullable = true, Metadata.empty),
StructField("is_partition_deleted", DataTypes.BooleanType, nullable = true, Metadata.empty),
StructField("time_taken_in_millis", DataTypes.LongType, nullable = true, Metadata.empty),
- StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty)
+ StructField("total_files_deleted", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("earliest_commit_to_retain", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("last_completed_commit_timestamp", DataTypes.StringType, nullable = true, Metadata.empty),
+ StructField("version", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_clean", DataTypes.IntegerType, nullable = true, Metadata.empty),
+ StructField("total_partitions_to_delete", DataTypes.IntegerType, nullable = true, Metadata.empty)
))
def parameters: Array[ProcedureParameter] = PARAMETERS
- def outputType: StructType = if (includePartitionMetadata) PARTITION_METADATA_OUTPUT_TYPE else OUTPUT_TYPE
+ def outputType: StructType = OUTPUT_TYPE
override def call(args: ProcedureArgs): Seq[Row] = {
super.checkArgs(PARAMETERS, args)
- val table = getArgValueOrDefault(args, PARAMETERS(0)).get.asInstanceOf[String]
- val limit = getArgValueOrDefault(args, PARAMETERS(1)).get.asInstanceOf[Int]
- val showArchived = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Boolean]
- val filter = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[String]
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
+ val limit = getArgValueOrDefault(args, PARAMETERS(2)).get.asInstanceOf[Int]
+ val showArchived = getArgValueOrDefault(args, PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val filter = getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String]
+ val startTime = getArgValueOrDefault(args, PARAMETERS(5)).get.asInstanceOf[String]
+ val endTime = getArgValueOrDefault(args, PARAMETERS(6)).get.asInstanceOf[String]
validateFilter(filter, outputType)
- val hoodieCatalogTable = HoodieCLIUtils.getHoodieCatalogTable(sparkSession, table)
- val basePath = hoodieCatalogTable.tableLocation
+ val basePath = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val activeResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit)
- } else {
- getCleans(metaClient.getActiveTimeline, limit)
- }
+ val activeResults = getCombinedCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit, metaClient, startTime, endTime, "ACTIVE")
val finalResults = if (showArchived) {
- val archivedResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit)
+ val archivedResults = getCombinedCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit, metaClient, startTime, endTime, "ARCHIVED")
+ val combinedResults = (activeResults ++ archivedResults)
+ .sortWith((a, b) => a.getString(0) > b.getString(0))
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ combinedResults
} else {
- getCleans(metaClient.getArchivedTimeline, limit)
+ combinedResults.take(limit)
}
- (activeResults ++ archivedResults)
- .sortWith((a, b) => a.getString(0) > b.getString(0))
- .take(limit)
} else {
- activeResults
+ if (startTime.trim.nonEmpty && endTime.trim.nonEmpty) {
+ activeResults
+ } else {
+ activeResults.take(limit)
+ }
}
applyFilter(finalResults, filter, outputType)
}
- override def build: Procedure = new ShowCleansProcedure(includePartitionMetadata)
+ override def build: Procedure = new ShowCleansProcedure()
- private def getCleansWithPartitionMetadata(timeline: HoodieTimeline,
- limit: Int): Seq[Row] = {
+ private def getCombinedCleansWithPartitionMetadata(timeline: HoodieTimeline, limit: Int, metaClient: HoodieTableMetaClient, startTime: String, endTime: String, timelineType: String): Seq[Row] = {
import scala.collection.JavaConverters._
+ import scala.util.{Failure, Success, Try}
+
+ val filteredCleanInstants = timeline.getCleanerTimeline.getInstants.asScala.toSeq
+ .filter { instant =>
+ val instantTime = instant.requestedTime()
+ val withinStartTime = if (startTime.nonEmpty) instantTime >= startTime else true
+ val withinEndTime = if (endTime.nonEmpty) instantTime <= endTime else true
+ withinStartTime && withinEndTime
+ }
+ val allCleanInstants = if (startTime.nonEmpty && endTime.nonEmpty) {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime())
+ } else {
+ filteredCleanInstants.sortWith((a, b) => a.requestedTime() > b.requestedTime()).take(limit)
+ }
- val (rows: util.ArrayList[Row], cleanInstants: util.ArrayList[HoodieInstant]) = getSortedCleans(timeline)
-
- var rowCount = 0
-
- cleanInstants.asScala.takeWhile(_ => rowCount < limit).foreach { cleanInstant =>
- val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
-
- cleanMetadata.getPartitionMetadata.entrySet.asScala.takeWhile(_ => rowCount < limit).foreach { partitionMetadataEntry =>
- val partitionPath = partitionMetadataEntry.getKey
- val partitionMetadata = partitionMetadataEntry.getValue
-
- rows.add(Row(
- cleanInstant.requestedTime(),
- cleanInstant.getCompletionTime,
- cleanInstant.getAction,
- cleanMetadata.getStartCleanTime,
- partitionPath,
- partitionMetadata.getPolicy,
- partitionMetadata.getDeletePathPatterns.size(),
- partitionMetadata.getSuccessDeleteFiles.size(),
- partitionMetadata.getFailedDeleteFiles.size(),
- partitionMetadata.getIsPartitionDeleted,
- cleanMetadata.getTimeTakenInMillis,
- cleanMetadata.getTotalFilesDeleted
- ))
- rowCount += 1
+ val allRows = scala.collection.mutable.ListBuffer[Row]()
+
+ allCleanInstants.foreach { cleanInstant =>
+ if (cleanInstant.getState == HoodieInstant.State.COMPLETED) {
+ Try {
+ val cleanMetadata = timeline.readCleanMetadata(cleanInstant)
+
+ cleanMetadata.getPartitionMetadata.entrySet.asScala.foreach { partitionMetadataEntry =>
+ val partitionPath = partitionMetadataEntry.getKey
+ val partitionMetadata = partitionMetadataEntry.getValue
+
+ val row = Row(
+ cleanInstant.requestedTime(),
+ cleanInstant.getCompletionTime,
+ cleanInstant.getState.name(),
+ cleanInstant.getAction,
+ timelineType,
+ cleanMetadata.getStartCleanTime,
+ partitionPath,
+ partitionMetadata.getPolicy,
+ partitionMetadata.getDeletePathPatterns.size(),
+ partitionMetadata.getSuccessDeleteFiles.size(),
+ partitionMetadata.getFailedDeleteFiles.size(),
+ partitionMetadata.getIsPartitionDeleted,
+ cleanMetadata.getTimeTakenInMillis,
+ cleanMetadata.getTotalFilesDeleted,
+ cleanMetadata.getEarliestCommitToRetain,
+ cleanMetadata.getLastCompletedCommitTimestamp,
+ cleanMetadata.getVersion,
+ null,
+ null
+ )
+ allRows += row
+ }
+ if (cleanMetadata.getPartitionMetadata.isEmpty) {
Review Comment:
got it, I though of doing it but smh missed
--
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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3231103429
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2b500858557a3e3d917e42959d66237dd1d4599c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2b500858557a3e3d917e42959d66237dd1d4599c 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3233958060
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200)
<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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3237880350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b4dffdd4e409045edd57e38b0376f0cf98452a2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239)
* 7fe006c3f887de640c9fe7f35d1dcc880567b40c 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238239266
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7fe006c3f887de640c9fe7f35d1dcc880567b40c Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271)
* ec3e1cff11c11ea9bafed719a2598bcba95d0da2 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #13793:
URL: https://github.com/apache/hudi/pull/13793#issuecomment-3238767078
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8200",
"triggerID" : "d69cc1a53428aeb5ce3d3c8fd3ce8475c5d329c3",
"triggerType" : "PUSH"
}, {
"hash" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8223",
"triggerID" : "fa38e71eb89ffe5571e9f64cc45407e6319c62b2",
"triggerType" : "PUSH"
}, {
"hash" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8239",
"triggerID" : "0b4dffdd4e409045edd57e38b0376f0cf98452a2",
"triggerType" : "PUSH"
}, {
"hash" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8271",
"triggerID" : "7fe006c3f887de640c9fe7f35d1dcc880567b40c",
"triggerType" : "PUSH"
}, {
"hash" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8274",
"triggerID" : "ec3e1cff11c11ea9bafed719a2598bcba95d0da2",
"triggerType" : "PUSH"
}, {
"hash" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275",
"triggerID" : "6e153692e2f3f03236bd9951882ac0d170401e8f",
"triggerType" : "PUSH"
}, {
"hash" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a60e8bead700c96e9848131a3c621d03e1bbaf06",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6e153692e2f3f03236bd9951882ac0d170401e8f Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=8275)
* a60e8bead700c96e9848131a3c621d03e1bbaf06 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] [HUDI-9765] Support for plans along with partition metadata in all the spark action procedures [hudi]
Posted by "vamshikrishnakyatham (via GitHub)" <gi...@apache.org>.
vamshikrishnakyatham commented on code in PR #13793:
URL: https://github.com/apache/hudi/pull/13793#discussion_r2308551108
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -17,53 +17,80 @@
package org.apache.spark.sql.hudi.command.procedures
-import org.apache.hudi.{HoodieCLIUtils, SparkAdapterSupport}
-import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline, TimelineLayout}
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
-import java.util
-import java.util.Collections
import java.util.function.Supplier
import scala.collection.JavaConverters._
/**
- * Spark SQL procedure to show completed clean operations for a Hudi table.
+ * Spark SQL procedure to show all clean operations for a Hudi table.
*
- * This procedure displays information about clean operations that have been executed.
- * Clean operations remove old file versions to reclaim storage space and maintain table performance.
+ * This procedure provides a comprehensive view of Hudi clean operations.
+ * It displays completed clean operations with full partition metadata for both completed and pending operations.
*
* == Parameters ==
* - `table`: Required. The name of the Hudi table to query
- * - `limit`: Optional. Maximum number of clean operations to return (default: 10)
+ * - `path`: Optional. The path of the Hudi table (anyone of `table` or `path` must be provided)
Review Comment:
Ah, my bad, its optional (I will change it in the description), either of the table name or path is required, we do that verification in the `getBaseFile` method. fixed it!
--
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