hudi-agent commented on code in PR #19974:
URL: https://github.com/apache/hudi/pull/19974#discussion_r4059319837
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowTimelineProcedure.scala:
##########
@@ -137,9 +137,15 @@ class ShowTimelineProcedure extends BaseProcedure with
ProcedureBuilder with Spa
val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
- val timelineEntries = getTimelineEntries(metaClient, limit, showArchived,
startTime, endTime)
-
- applyFilter(timelineEntries, filter, outputType)
+ // getTimelineEntries applies `limit` while collecting, so with a filter
it has to be lifted here and
+ // reapplied to the matching rows; otherwise the filter only ever sees the
first `limit` instants.
+ val scanLimit = if (hasFilter(filter)) Int.MaxValue else limit
Review Comment:
🤖 With `showArchived => true` and any filter, `scanLimit = Int.MaxValue` now
flows into `loadCompletedInstantDetailsInMemory` /
`loadCompactionDetailsInMemory`, so the whole archived timeline (with commit
metadata, in METADATA load mode) is materialised on the driver before the
filter runs. For a long-lived table that's a lot more than the previous `limit`
instants. Is that acceptable here, or would the same incremental approach you
used for `show_logfile_records` (scan newest-first in chunks until `limit`
matches are found) be worth applying? Same shape in `ShowCleansProcedure` where
every archived clean's metadata is now read.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala:
##########
@@ -159,24 +159,28 @@ class ShowCleansProcedure(includePartitionMetadata:
Boolean) extends BaseProcedu
val basePath = hoodieCatalogTable.tableLocation
val metaClient = createMetaClient(jsc, basePath)
+ // `limit` bounds how many clean instants get their metadata read, so with
a filter it has to be
+ // lifted here and reapplied to the matching rows; otherwise the filter
only ever sees the newest
+ // `limit` cleans.
+ val scanLimit = if (hasFilter(filter)) Int.MaxValue else limit
val activeResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getActiveTimeline, limit)
+ getCleansWithPartitionMetadata(metaClient.getActiveTimeline, scanLimit)
} else {
- getCleans(metaClient.getActiveTimeline, limit)
+ getCleans(metaClient.getActiveTimeline, scanLimit)
}
val finalResults = if (showArchived) {
val archivedResults = if (includePartitionMetadata) {
- getCleansWithPartitionMetadata(metaClient.getArchivedTimeline, limit)
+ getCleansWithPartitionMetadata(metaClient.getArchivedTimeline,
scanLimit)
} else {
- getCleans(metaClient.getArchivedTimeline, limit)
+ getCleans(metaClient.getArchivedTimeline, scanLimit)
}
(activeResults ++ archivedResults)
.sortWith((a, b) => a.getString(0) > b.getString(0))
- .take(limit)
+ .take(scanLimit)
Review Comment:
🤖 nit: this `.take(scanLimit)` looks redundant now — `applyFilterAndLimit`
already applies `limit` below, and when `scanLimit` is `Int.MaxValue` this is a
no-op. Could we drop it?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowFsPathDetailProcedure.scala:
##########
@@ -77,20 +77,12 @@ class ShowFsPathDetailProcedure extends BaseProcedure with
ProcedureBuilder {
}
val df = spark.sqlContext.createDataFrame(rows, OUTPUT_TYPE)
- val results = if (sort) {
- if (limit.isDefined) {
-
df.orderBy(df("storage_size").desc).limit(limit.get.asInstanceOf[Int]).collect()
- } else {
- df.orderBy(df("storage_size").desc).collect()
- }
- } else {
- if (limit.isDefined) {
-
df.orderBy(df("file_num").desc).limit(limit.get.asInstanceOf[Int]).collect()
- } else {
- df.orderBy(df("file_num").desc).collect()
- }
- }
- applyFilter(results, filter, outputType)
+ val ordered = df.orderBy(df(if (sort) "storage_size" else "file_num").desc)
+ val limitValue = resolveLimit(limit)
+ // The bound has to come after the filter; limiting the DataFrame first
would show the filter only
+ // the first `limit` rows of the ordering.
+ val results = if (limit.isDefined && !hasFilter(filter))
ordered.limit(limitValue).collect() else ordered.collect()
Review Comment:
🤖 nit: this `collect-all-if-filtered else limit-then-collect` branch plus
its comment is duplicated verbatim in ShowBootstrapMappingProcedure — worth a
small helper in BaseProcedure (e.g. `collectForFilter(df, filter, limit)`) next
to `applyFilterAndLimit` so the reasoning lives in one place?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
