hudi-agent commented on code in PR #19974:
URL: https://github.com/apache/hudi/pull/19974#discussion_r4043921451
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/BaseProcedure.scala:
##########
@@ -137,4 +137,15 @@ abstract class BaseProcedure extends Procedure {
results
}
}
+
+ protected def hasFilter(filter: String): Boolean = filter != null &&
filter.trim.nonEmpty
+
+ /**
+ * Filters first and truncates afterwards, so `limit` bounds the matching
rows rather than the rows the
+ * filter gets to see. Truncating first makes `limit => n, filter => ...`
return the matches among the
+ * first n rows, which is empty whenever the matches all sit past the
cut-off.
+ */
+ protected def applyFilterAndLimit(results: Seq[Row], filter: String, schema:
StructType, limit: Int): Seq[Row] = {
+ applyFilter(results, filter, schema).take(limit)
+ }
}
Review Comment:
🤖 nit: the `limit.map(_.asInstanceOf[Int]).getOrElse(Int.MaxValue)` pattern
shows up identically in ShowFsPathDetailProcedure, ShowInvalidParquetProcedure
and ShowMetadataTableFilesProcedure — might be worth adding a small helper here
(e.g. `resolveLimit(limit: Option[Any]): Int`) alongside `hasFilter`.
<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/ShowHoodieLogFileRecordsProcedure.scala:
##########
@@ -63,6 +63,9 @@ class ShowHoodieLogFileRecordsProcedure extends BaseProcedure
with ProcedureBuil
val filter = getArgValueOrDefault(args,
parameters(5)).get.asInstanceOf[String]
validateFilter(filter, outputType)
+ // `limit` bounds how many records are read out of the log files, 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`.
+ val scanLimit = if (hasFilter(filter)) Int.MaxValue else limit
Review Comment:
🤖 With a filter this now buffers every record from every matched log file
into the on-heap `allRecords` list before filtering (in merge mode the scanner
already spills, but the materialised `IndexedRecord`s no longer do). On a large
file group a plain `filter => ...` call could OOM the driver where it used to
return quickly. Have you considered evaluating the filter incrementally (e.g.
per data block / per batch of rows via `applyFilter`) and stopping once `limit`
matches are collected, so the fix stays bounded?
<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]
