[SPARK-27852][Spark Core] updateBytesWritten() operaton is missed

[whatlulumomo]

What changes were proposed in this pull request?

one line code maybe missed in core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala

 override def write(kvBytes: Array[Byte], offs: Int, len: Int): Unit = {
     if (!streamOpen) {
       open()
     }
     bs.write(kvBytes, offs, len)
+    updateBytesWritten()   // the function is missed
 } 

Possible Patch Link

[AmplabJenkins]

Can one of the admins verify this patch?

[srowen]

This would make it update metrics on every write. It appears this is purposely done only every 16,000 records for this reason.

[whatlulumomo]

This would make it update metrics on every write. It appears this is purposely done only every 16,000 records for this reason.

 override def write(kvBytes: Array[Byte], offs: Int, len: Int): Unit = {
     if (!streamOpen) {
       open()
     }
     bs.write(kvBytes, offs, len)
 } 

In this write function, there isn't any work to update metrics or records. bs.write(kvBytes, offs, len) doesn't do the work, either. It seems strange.

[srowen]

Look at recordWritten().

[whatlulumomo]

Look at recordWritten().

recordWritten() is never used in override def write(kvBytes: Array[Byte], offs: Int, len: Int)

Here is the function body:

 override def write(kvBytes: Array[Byte], offs: Int, len: Int): Unit = {
     if (!streamOpen) {
       open()
     }
     bs.write(kvBytes, offs, len)
 } 

recordWritten() is called only in

def write(key: Any, value: Any) {
    if (!streamOpen) {
      open()
    }

    objOut.writeKey(key)
    objOut.writeValue(value)
    recordWritten()
  }

[HyukjinKwon]

So, is the matrix not getting updated? what's the issue this PR fixes?

[srowen]

@BestOreo the caller calls write potentially may times to write a record, then calls recordWritten, where metrics are update every 16384 records. There's a write overload that writes a whole record at once and then calls recordWritten. We specifically don't want to update metrics several times for every record. That would be very slow.