Repository navigation
Conversation
b88f53d to
f1a6f78
Compare
anoopj
left a comment
There was a problem hiding this comment.
The code change looks good to me. We might be missing code coverage for the non-vectorized path (RowDataReader) because the Spark read follows the vectorized path?
@anoopj Addressed by adding unit test. Also added couple of integration tests as well. Thanks! |
|
Gentle bump! |
|
@anoopj Gentle bump!! |
anoopj
left a comment
There was a problem hiding this comment.
LGTM. Thanks. We should get a committer to review/merge this as the next step.
|
cc @aokolnychyi or @anuragmantri for review please. Thanks! |
|
Gentle ping @rdblue !! |
uros-b
left a comment
There was a problem hiding this comment.
Thank you for working on this @venkata91, and thank you @anoopj for reviewing the changes! The accumulation logic looks good (monotonic += in the finally, single-threaded per reader; ScanDuration.NAME / TaskScanDuration.name() pair up). Leaving here just a couple of inline comments that should be considered before merge (the Locale.ROOT thing is pretty important, PTAL).
Also, the PR description is a bit stale now, "V1, V2 and V3 tables. 3 tests" no longer matches (five tests now, no V3 format exercised), and "Timed per next() call, not per row" is inaccurate for RowDataReader, where each next() yields a single InternalRow. Please update both.
| } else if (totalNanos < TimeUnit.SECONDS.toNanos(1)) { | ||
| return TimeUnit.NANOSECONDS.toMillis(totalNanos) + " ms"; | ||
| } else { | ||
| return String.format("%.1f s", totalNanos / (double) TimeUnit.SECONDS.toNanos(1)); |
There was a problem hiding this comment.
Locale.ROOT - ScanDuration.aggregateTaskMetrics uses String.format("%.1f s", ...) without a locale, so it renders "2,5 s" on a de/fr/pt JVM. The rest of the main source tree passes Locale.ROOT for numeric formatting; please use String.format(Locale.ROOT, "%.1f s", ...).
| } | ||
|
|
||
| @Override | ||
| public String aggregateTaskMetrics(long[] taskMetrics) { |
There was a problem hiding this comment.
aggregateTaskMetrics sums per-task nanos into a single scalar, which discards straggler-task skew. Spark's own scanTime reports total (min, med, max) and you already have the full per-task array in hand - worth considering matching that shape.
|
Thanks @uros-b for the review and your suggestions. Addressed all the review comments. |
|
@uros-b Gentle ping! Do you think we can merge this PR? |
|
@venkata91 Please ping an expert Iceberg committer for further review here ^^ |
|
@nastra @singhpk234 @manuzhang mind taking a look when you get a chance? This has been open a while and is ready for a committer review. It adds a #17820 builds on this with the write-side equivalent ( |
|
Gentle ping, please review @nastra @singhpk234 @manuzhang !! Thanks! |
|
Gentle ping, please review @nastra |
| public class TestScanDuration { | ||
|
|
||
| @Test | ||
| public void testAggregateTaskMetrics() { |
There was a problem hiding this comment.
let's avoid using test prefix in newly added tests
There was a problem hiding this comment.
Done in 51ae742. I also dropped the prefix from the two tests this PR adds to TestSparkReadMetrics, and made TestScanDuration package-private. Thanks for the review! I'll open the Spark 4.2 port once this merges.
nastra
left a comment
There was a problem hiding this comment.
LGTM with one comment that would be good to address. Also please make sure to port this to Spark 4.2
| @@ -88,7 +89,9 @@ class BatchDataReader extends BaseBatchReader<FileScanTask> | |||
| @Override | |||
| public CustomTaskMetric[] currentMetricsValues() { | |||
| return new CustomTaskMetric[] { | |||
| new TaskNumSplits(numSplits), new TaskNumDeletes(counter().get()) | |||
| new TaskNumSplits(numSplits), | |||
There was a problem hiding this comment.
Since the no-op initMEtricsValues is inherited, metrics from a later reader will override that of a previous one in the same task. Could you please fix it in a follow-up PR? Same for RowDataReader.
There was a problem hiding this comment.
Good catch. With the no-op inherited, a later reader in the same task starts from zero instead of picking up the earlier reader's value, so only the last reader's duration survives. I'll fix BatchDataReader and RowDataReader together in a follow-up and link it here.
There was a problem hiding this comment.
Fixed in #18726. It turned out Spark 4.2 is not affected, since it no longer calls initMetricsValues, so the fix is 4.1 only.
Iceberg reports scan bytes and file/manifest counts per task, but the only timer in the metrics system is totalPlanningDuration, which measures driver-side planning. There is no metric for how long a task spent actually reading data, so a scan's throughput cannot be derived from Iceberg metrics alone. Add scanDuration: wall time a task spends in BaseReader.next(), accumulated per call rather than per row so the timer costs nothing measurable relative to the read. Emitted as TaskScanDuration from BatchDataReader and RowDataReader, aggregated by ScanDuration, mirroring the existing TotalPlanningDuration pair. Values are nanoseconds: a single split can be read in well under a millisecond, so a millisecond timer would round most per-task reads to zero. ScanDuration formats the aggregate as ns/us/ms/s for the UI. Generated-by: Claude Opus 5
Generated-by: Claude Opus 5
51ae742 to
901bf2b
Compare
|
@nastra I renamed the tests as you asked and rebased onto main, so CI is green now. Could you merge this when you get a chance? I'll open the 4.2 port right after. |
Summary
totalPlanningDuration, which covers driver-side planning. There is no metric for time spent actually reading, so scan throughput can't be derived from Iceberg metrics alone.Changes
scanDuration: wall time a task spends inBaseReader.next(), covering split open and iteration. Accumulated in afinally, so a task that throws still reports what it read.TaskScanDurationfromBatchDataReaderandRowDataReader, aggregated byScanDuration, mirroring theTotalPlanningDurationpair.totalPlanningDuration, since a split can be read in well under a millisecond.total (min, med, max), matchingMetricUtils.stringValue, so a straggler task stays visible.Testing done
TestSparkReadMetrics(5): V1 and V2 tables, delete path, multi-task aggregation, and the non-vectorizedRowDataReaderpath.TestScanDuration(5): rendering, empty and straggler cases, input not mutated, locale independence.spotlessCheckandcheckstylepass.Notes
spark/v4.1. Happy to backport to 4.0 and 3.5 in a follow-up.