rahil-c commented on code in PR #19737:
URL: https://github.com/apache/hudi/pull/19737#discussion_r3859861177
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala:
##########
@@ -691,17 +715,16 @@ object BatchedBlobReader {
// Apply mapPartitions
val result = df.mapPartitions { partition =>
- // Create storage and reader for this partition
- val storage = HoodieStorageUtils.getStorage(broadcastConf.value)
- val reader = new BatchedBlobReader(storage, maxGapBytes, lookaheadSize)
+ // Create reader for this partition
+ val reader = new BatchedBlobReader(
+ HoodieStorageUtils.getStorage(_, broadcastConf.value), maxGapBytes,
lookaheadSize)
Review Comment:
Resolving per read gives up the amortization master had here. Each blob read
now pays `ReflectionUtils.loadClass` (the `Class` is cached in `CLAZZ_CACHE`,
but `getConstructor(...).newInstance(...)` is not),
`HadoopFSUtils.prepareHadoopConf`'s full `System.getenv()` scan, and
`path.getFileSystem(conf)` -> `FileSystem.get` under the global
`FileSystem$Cache` monitor. That is once per row on the whole-file path and
once per merged range on the ranged path, where before it was once per
partition, in a reader whose purpose is amortizing work across many rows.
I think the reasoning in fe217b9b is half right: Hadoop's static cache does
stop the `FileSystem` being rebuilt, but it does not make the lookup free (it
takes a JVM-wide monitor, contended by every task thread on the executor) and
it does not stop `getStorage` reflecting a fresh wrapper each call.
Would a memo keyed by (scheme, authority) get the correctness this buys
without the per-read cost? It keeps the multi-filesystem property that dropping
the single handle was for, and unlike the earlier per-partition handle it
carries no one-filesystem assumption:
```scala
val byFs = mutable.Map.empty[(String, String), HoodieStorage]
val reader = new BatchedBlobReader(
p => byFs.getOrElseUpdate((p.toUri.getScheme, p.toUri.getAuthority),
HoodieStorageUtils.getStorage(p, broadcastConf.value)),
maxGapBytes, lookaheadSize)
TaskContext.get().addTaskCompletionListener[Unit](_ =>
byFs.values.foreach(_.close()))
```
That does reintroduce the caller-side cache and the completion listener you
removed, so happy to be told the per-read cost is small enough against
object-store latency not to be worth it. Mostly I wanted the tradeoff on the
record, since `testResolvedStorageIsClosed` asserts construction equals close
but does not pin how many constructions a scan does.
--
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]
