rangareddy commented on code in PR #19974:
URL: https://github.com/apache/hudi/pull/19974#discussion_r4046361119
##########
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:
Good catch, fixed. You are right that this was unbounded: with a filter the
scan bound was lifted wholesale, so every record of every matched log file
landed in the on-heap list.
It now filters incrementally as you suggested. Rows go into a small batch,
`applyFilter` runs per batch, only the matches are retained, and the scan stops
as soon as `limit` matches have been found, so the heap is bounded by the batch
plus the matches no matter how much the log files hold. Since `applyFilter` is
a passthrough when no filter is set, a batch of one reproduces the previous
bound exactly, which the earlier approach could not promise for the unfiltered
path.
Covered by a new assertion in `TestHoodieLogFileProcedure` that requests
each of two records with `limit => 1`; asserting both keeps it independent of
scan order. Bounding by rows seen rather than rows matched fails it (`Expected
1, but got 0 ... matching b2`).
##########
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:
Done - added `resolveLimit(limit: Option[Any]): Int` to `BaseProcedure` next
to `hasFilter`, and the three sites now use 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
