AWS, GCP, Azure: Track read metrics for positioned, vectored, and accelerated reads

[dbtsai]

Problem

FileIO input streams (S3InputStream, GCSInputStream, ADLSInputStream, and the GCS analytics-core wrapper) only incremented READ_BYTES/READ_OPERATIONS in the sequential read() paths. The positioned-read methods readFully()/readTail() — and the vectored path that Parquet layers on top of them since 1.11.0 via ParquetRangeReadableInputStreamAdapter — bypassed instrumentation entirely.

Because RangeReadable.readVectored()'s default implementation loops over readFully(), and Parquet now routes most column-chunk data through vectored reads, nearly all Parquet data bytes disappeared from the read metrics that Spark consumes for task input metrics. This under-reports Hadoop read metrics significantly.

Closes #17208

Fix

Instrument readFully()/readTail() in all three object-store streams (S3InputStream, ADLSInputStream, GCSInputStream). The default RangeReadable.readVectored() loops over readFully(), so the vectored Parquet path is covered automatically for S3/ADLS/GCS.

The GCS analytics-core wrapper (AnalyticsCoreUtil) overrides readVectored() and delegates to the analytics-core stream, so it is instrumented explicitly — recording bytes per range as each range future completes successfully (failed ranges are not counted). Empty-tail reads that return -1 are guarded so they never decrement the byte counter.

Additionally, the S3 analytics-accelerator path (AnalyticsAcceleratorInputStreamWrapper, used when s3.analytics-accelerator is enabled) previously tracked no read metrics at all. It is now threaded the MetricsContext already available on S3InputFile and increments the counters from its read() methods; it is not RangeReadable, so Parquet reads flow through read(byte[], int, int) and counting there captures all data bytes.

Tests

• TestS3InputStream — testReadFullyTracksMetrics, testReadTailTracksMetrics
• TestADLSInputStream — testReadFullyTracksMetrics, testReadTailTracksMetrics
• TestGCSInputStream — testRangeReadMetrics, testReadTailEmptyObjectDoesNotDecrementMetrics
• TestAnalyticsCoreUtil — readVectoredCountsOnlyCompletedRanges (no eager counting; successful ranges counted; failed ranges not counted)
• TestAnalyticsAcceleratorInputStreamWrapper — testReadTracksMetrics

Scope notes

• OSSInputStream (Aliyun) is unaffected today because it does not implement RangeReadable; it would need the same treatment if RangeReadable support is added there.
• This covers the FileIO streams called out in the issue plus the accelerated S3/GCS paths; it is not a full audit of every custom input-stream wrapper.

Performance impact

None on the data path. This is a metrics-correctness change: the same reads happen as before, and the added work is a Counter.increment(...) (backed by LongAdder) per read call — no locks, no allocation, no extra I/O. The fix corrects the measurement (READ_BYTES/READ_OPERATIONS that Spark surfaces as task input metrics), not the reads being measured.

[viirya]

This counts inside a thenAccept callback that runs on whichever thread completes the future — an analytics-core background thread, not the caller. Every other counting site in this PR increments synchronously on the caller (task) thread.

That matters for the stated goal: under HadoopMetricsContext, READ_BYTES → FileSystem.Statistics.incrementBytesRead, and Hadoop's Statistics accumulates per-thread, with Spark attributing task input bytes from the task thread's statistics. If these futures complete on a background thread, the bytes land on the wrong thread's Statistics and may never reach Spark's task metrics — so the fix could silently no-op on this specific analytics-core path, which is the scenario #17208 is about.

TestAnalyticsCoreUtil uses DefaultMetricsContext (a plain LongAdder), so it can't surface a thread-attribution issue. Could you confirm the behavior against a real HadoopMetricsContext? If the async attribution is inherent to analytics-core, counting range.length() synchronously before delegating (accepting the failed-range imprecision this code deliberately avoids), or documenting it as a known limitation, would be worth weighing.

[dbtsai]

Good catch — you're right, and I couldn't refute it. HadoopMetricsContext maps READ_BYTES to FileSystem.Statistics.incrementBytesRead, and Hadoop's Statistics accumulates in a ThreadLocal<StatisticsData>; Spark attributes task input bytes from the task thread's slot. My thenAccept callback runs on whichever thread completes the future (an analytics-core background thread), so on this path the bytes would land on the wrong thread's Statistics and never reach Spark's task metrics — a silent no-op for exactly the scenario #17208 targets. And as you note, the DefaultMetricsContext/LongAdder test can't surface thread attribution, so it stayed green.

I'll switch this path to count synchronously on the caller thread before delegating to stream.readVectored(...), matching every other counting site in the PR. The tradeoff is that, unlike the completion-callback approach, synchronous counting can only use the requested range.length() (the delivered size isn't known until the async read finishes), so it loses the failed-range / short-read precision this code was trying to get. Given the goal is that the bytes reach Spark at all, correct-thread attribution is the more important property; I'll count range.length() synchronously and add a comment documenting the imprecision. I'll also drop the now-misleading short-read test assertion for this path.

Does counting requested length synchronously (with a comment on the imprecision) sound right to you, or would you prefer I document the async attribution as a known limitation instead and leave counting in the callback? Happy to go either way.

[viirya]

Synchronous counting of range.length() is the right call. The whole point of this metric is to reach Spark's per-thread task input statistics via HadoopMetricsContext; if it accumulates on the wrong thread it never arrives, so the number is effectively zero on this path. Over-counting a short read or a failed batch by a bounded amount is a precision issue — the magnitude is right and it lands where Spark can see it. Correct-thread attribution wins over precision here, so I wouldn't keep the callback and just document it — that would knowingly ship a broken metric on this path.

Two things to fold in when you switch it:

• Since you'll increment range.length() up front, a batch where stream.readVectored(...) throws will have already counted those bytes. That's acceptable for a metric (bounded over-count on a failure path), but please say so in the comment so the imprecision is on the record — both the short-read and the failed-range cases.
• Agreed on dropping the short-read assertion. Worth being explicit in the test/PR that the remaining DefaultMetricsContext test can't verify thread attribution at all (it's a plain LongAdder); the correctness argument here is the per-thread Statistics reasoning, not the unit test. No need to fake a thread-attribution test — just don't let the green test imply coverage it doesn't have.

Thanks for the quick turnaround on this and the EOF fix.

[szehon-ho]

+1 on synchronous range.length() counting — agreed it's the right correct-thread-vs-precision tradeoff.

Related convention point: the same "don't count on a no-data read" rule from #16055 isn't applied in readTail on the three streams — readBytes is guarded but readOperations still increments when the tail read returns -1/0 (empty object). I left inline comments; worth folding a bytesRead > 0 guard into this PR while we're here.

[dbtsai]

Done — switched this path to count synchronously on the caller thread before delegating to stream.readVectored(...), using the requested range.length(), matching every other counting site in the PR. Added a comment documenting the bounded over-count on short reads / failed ranges, and flipped the test to readVectoredCountsRequestedLengthSynchronously (asserts both counters move up front, before the futures complete). Also noted in the test that DefaultMetricsContext (a plain LongAdder) cannot verify per-thread Statistics attribution, so the correctness argument rests on the HadoopMetricsContext reasoning rather than the unit test. Thanks for catching this.

[dbtsai]

Good point — folded the bytesRead > 0 guard into readTail on all three streams in this PR.

[anoopj]

So this is incrementing readOperations on EOF (and you have a test for it). This is contradictory to your own implementation of read() above, and also to a recent bug fix we recently merged. I recommend aligning with that convention.

[dbtsai]

Good catch. Fixed. Thanks.

[anoopj]

Suggested change

	int bytesRead = this.delegate.read(b, off, len);
	if (bytesRead > 0) {
	readBytes.increment(bytesRead);
	}
	readOperations.increment();
	return bytesRead;
	int bytesRead = delegate.read(b, off, len);
	if (bytesRead != -1) {
	readBytes.increment(bytesRead);
	readOperations.increment();
	}
	return bytesRead;

[szehon-ho]

readOperations is incremented even when bytesRead == -1 (empty/EOF tail), while readBytes is guarded. That's the same EOF-counting addressed in #16055, and it's inconsistent with read()/read(byte[]) above. Suggest guarding both together:

if (bytesRead > 0) {
  readBytes.increment(bytesRead);
  readOperations.increment();
}

and flipping testReadTailEmptyObjectDoesNotDecrementMetrics to assert readOperations == 0.

[dbtsai]

Done — guarded both counters on bytesRead > 0 and flipped the empty-object test to assert readOperations == 0.

[szehon-ho]

Same convention nit: IOUtil.readRemaining returns 0 at EOF, so an empty tail read still ticks readOperations (and readBytes.increment(0)). Guard both on bytesRead > 0 to stay consistent with read() and #16055.

[dbtsai]

Done, guarded both on bytesRead > 0.

[szehon-ho]

Same as S3/GCS — an empty tail read counts an operation. Guard readBytes/readOperations on bytesRead > 0 so a no-data read doesn't increment (#16055 convention).

[dbtsai]

Done, same guard.

[szehon-ho]

The two new metrics tests re-stub getObject with their own data, so this shared stub goes unused there and would trip UnnecessaryStubbingException — hence lenient(). Reasonable, but since only testReadTailClosesTheStream needs the default stub, a cleaner option is to drop it from @BeforeEach and stub getObject inline in that one test, keeping strict stubbing everywhere. Non-blocking.

[dbtsai]

Dropped the shared stub from @BeforeEach and stubbed inline in the two tests that need it, so strict stubbing everywhere. Thanks.

[JoshRosen]

Pre-existing issue, but maybe worth fixing: these increments are unconditional, which is inconsistent with the other conditional fixes in this PR.

[JoshRosen]

Similar comment here; we should re-review this for consistency with the guarded increments in the rest of the PR / code.

[dbtsai]

Discussion: AnalyticsCoreUtil.readVectored over-counts failed / short ranges

Splitting this out from the inline threads since it is a deliberate design tradeoff, not a bug to silently fix.

The GCS analytics-core vectored path now counts range.length() synchronously on the caller thread, before delegating to stream.readVectored(...):

for (FileRange range : ranges) {
  readBytes.increment(range.length());
  readOperations.increment();
}
stream.readVectored(objectRanges, allocate);

This was the fix for @viirya's catch that counting in the range-future completion callbacks attributes bytes to an analytics-core background thread — under HadoopMetricsContext, READ_BYTES maps to FileSystem.Statistics.incrementBytesRead, which accumulates per-thread, and Spark reads task input bytes from the task thread. Background-thread counting is effectively a no-op for Spark task metrics, which is the exact scenario #17208 is about. So correct-thread attribution has to win.

The known imprecision that buys us that:

• Short read near EOF — we count the requested length, not the delivered bytes (bounded over-count).
• Failed range — if stream.readVectored(...) throws, the bytes were already counted (the delegate throws after the increment loop).

My read: for a task-input metric, magnitude-correct-and-visible beats precise-but-invisible, and both errors are bounded. But I want to surface it explicitly rather than bury it in a code comment.

Options if the over-count is a concern:

1. Keep as-is (current) — synchronous, requested-length, documented imprecision.
2. Count range.length() before delegating, but subtract on the failure path — recovers the failed-range case; short-read imprecision remains (delivered size isn't known synchronously).
3. Callback-based counting — precise per delivered bytes, but wrong-thread attribution → doesn't reach Spark. Rejected for that reason.

Preference? cc @szehon-ho @JoshRosen @viirya — I'm inclined to keep option 1 but happy to move to option 2 if we want to tighten the failed-range case.