[SPARK-25237][SQL]remove updateBytesReadWithFileSize because we use Hadoop FileSystem s…

[dujunling]

…tatistics to update the inputMetrics

What changes were proposed in this pull request?

In FileScanRdd, we will update inputMetrics's bytesRead using updateBytesRead every 1000 rows or when close the iterator.

but when close the iterator, we will invoke updateBytesReadWithFileSize to increase the inputMetrics's bytesRead with file's length.

this will result in the inputMetrics's bytesRead is wrong when run the query with limit such as select * from table limit 1.

because we do not support for Hadoop 2.5 and earlier now, we always get the bytesRead from Hadoop FileSystem statistics other than files's length.

How was this patch tested?

manual test

[dujunling]

@wzhfy

[wzhfy]

ok to test

[wzhfy]

this seems to be caused by removing support for Hadoop 2.5 and earlier? cc original authors @cloud-fan@srowen to make sure

[wzhfy]

test this please

[maropu]

If we just remove this updateBytesReadWithFileSize, the issue in the description can be solved? We need to remove updateBytesReadWithFileSize in the line 142, too?

[dujunling]

Yes, before SPARK-19464, there will only one works between updateBytesRead and updateBytesReadWithFileSize. If the hadoop version is 2.5 or earlier, updateBytesReadWithFileSize works, If the hadoop version is 2.6 or later, updateBytesRead works.

[dujunling]

When there are one more files in the partition, the inputMetrics is wrong when updateBytesReadWithFileSize in the line 142 is exist.

[maropu]

aha, I see.

[maropu]

Is it difficult to add tests for checking the metric in the case select * from t limit 1?

[maropu]

btw, can you clean up the title and the description..?

[dujunling]

While metris suites are in core test , fileScanRdd should be in sql test, it is difficult to add tests to check the input metrics in sql module

[maropu]

I'm not sure we can test the case though, for example, how about the sequence below?

import org.apache.spark.TaskContext
spark.range(10).selectExpr("id AS c0", "rand() AS c1").write.parquet("/tmp/t1")
val df = spark.read.parquet("/tmp/t1")
val fileScanRdd = df.repartition(1).queryExecution.executedPlan.children(0).children(0).execute()
fileScanRdd.mapPartitions { part =>
println(s"Initial bytesRead=${TaskContext.get.taskMetrics().inputMetrics.bytesRead}")
TaskContext.get.addTaskCompletionListener[Unit] { taskCtx =>
// Check if the metric is correct?
println(s"Total bytesRead=${TaskContext.get.taskMetrics().inputMetrics.bytesRead}")
}
part
}.collect

[srowen]

It's OK to assume Hadoop 2.6+ only. In fact 2.6 is quite old anyway.

[dujunling]

@maropu I have added a ut to check the inputMetrics

[srowen]

The data above could be made deterministic so that you can assert the bytes read more exactly. I wonder if it's important to make sure the bytes read are exact, rather than just close, given that the change above would change the metric only a little I think.

You can just track the sum rather than all values written, but it doesn't matter.

[maropu]

I think a single partition is ok for this test.

[wzhfy]

ok to test

[SparkQA]

Test build #95508 has finished for PR 22232 at commit 1c32646.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[srowen]

@dujunling I personally think this can be merged, but only if the test is tightened up

[maropu]

@srowen I could take this over or do follow-up if the author is still inactive.

[srowen]

I think that's fine @maropu ; we can always apportion credit appropriately later.