Skip to content

[spark] Fix read batch time for the first batch and active splits - #10835

Open
wangzhigang1999 wants to merge 1 commit into
apache:masterfrom
wangzhigang1999:fix-spark-read-batch-time
Open

wangzhigang1999 wants to merge 1 commit into
apache:masterfrom
wangzhigang1999:fix-spark-read-batch-time

Conversation

@wangzhigang1999

Copy link
Copy Markdown
Contributor

Closes #10834

Purpose

Spark's read batch time can omit the first batch and the split being read when a task finishes. PaimonRecordReaderIterator initializes its counter after the first readBatch() call, clearing that batch's elapsed time. PaimonPartitionReader only reports completed splits, while Spark takes its final metric snapshot before closing the reader. A query that stops within a split, such as LIMIT, can therefore report zero or incomplete batch time.

Initialize the counter before the first read and include the active split in metric snapshots. Use the same helper when finishing a split and closing the partition reader: close the reader, add its elapsed time to the completed total, then clear the reference. If closing throws, retain the reference and completed total so a later cleanup call can retry. Successful repeated closes do not add the time again.

The change preserves the metric name, timing scope, and per-split conversion from nanoseconds to milliseconds.

Tests

Add two tests to the existing PaimonMetricTest suite:

  • Retain the first batch's elapsed time.
  • Include the active split and preserve the completed split's time across two splits. Cover early close and EOF, each with successful close or an injected first-close failure followed by cleanup retry. Check metric values after cleanup and repeated close, and verify the underlying close counts.

The timing tests delay the first batch by 25 ms and assert a 20 ms lower bound. The EOF cases take their comparison snapshot after the EOF read, since that read can add elapsed time.

Local results for the final test source:

Profile Versions Result
spark3 Spark 3.5.8 / Scala 2.12.18 13 passed, 0 failed, 0 skipped
spark4 Spark 4.1.2 / Scala 2.13.17 13 passed, 0 failed, 0 skipped

The 13 tests comprise 11 existing tests and 2 new tests; the second new test runs all four close/EOF combinations. Both runs used JDK 17 and the normal build checks. Dependencies, including the updated common reader bundled into the Spark version module, were installed before running the target suite.

mvn -B -pl paimon-spark/paimon-spark-ut -Pspark3 \
  -DfailIfNoTests=false -Dtest=none \
  -DwildcardSuites=org.apache.paimon.spark.sql.PaimonMetricTest test

mvn -B -pl paimon-spark/paimon-spark-ut -Pspark4 \
  -DfailIfNoTests=false -Dtest=none \
  -DwildcardSuites=org.apache.paimon.spark.sql.PaimonMetricTest clean test

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Bug] Spark read batch time omits the first batch and active split

1 participant