Skip to content

[core][spark] Report skipped manifests, resulted file size and record count in scan metrics - #9657

Open
zhuxiangyi wants to merge 1 commit into
apache:masterfrom
zhuxiangyi:scan-metrics
Open

[core][spark] Report skipped manifests, resulted file size and record count in scan metrics#9657
zhuxiangyi wants to merge 1 commit into
apache:masterfrom
zhuxiangyi:scan-metrics

Conversation

@zhuxiangyi

@zhuxiangyi zhuxiangyi commented Sep 5, 2026

Copy link
Copy Markdown
Contributor

Purpose

Scan metrics today tell the user how much was read, but not whether that amount is
reasonable
. Two concrete gaps:

1. Manifest pruning efficiency cannot be computed — the numerator is reported without the denominator.

lastScannedManifests reports manifestsResult.filteredManifests.size(), i.e. the count
after manifest level filtering. The total (allManifests.size()) sits on the very same
object but is never reported, so scannedManifests = 200 is ambiguous:

total manifests scanned what it means
A 1000 200 partition pruning works well
B 200 200 pruning did not happen at all

Case B is a common production problem (a predicate that fails to push down because of a
function or a type mismatch on the partition column), and it is exactly the case the current
metrics cannot distinguish.

2. Only file counts are reported, no bytes and no records.

resultedTableFiles = 3000 does not answer "how much data will this query read". In Paimon
the relation between file count and data volume is unstable: a frequently written table may
have 3000 files holding 2 GB, while the same table after compaction may have 300 files
holding 20 GB. Deciding whether a query is slow because of data volume, or whether a table
needs compaction, requires bytes and records. Both DataFileMeta#fileSize and
DataFileMeta#rowCount are already carried by the entries in the scan result.

There is also an asymmetry with the write side: CommitStats already reports record level
counters (deltaRecordsAppended, changelogRecordsAppended), while the scan side reports
file counts only.

Changes

Three fields are added to ScanStats and three gauges to ScanMetrics:

metric meaning
lastScanSkippedManifests allManifests.size() - filteredManifests.size()
lastScanResultedTableFilesSize total size in bytes of the files to be read
lastScanResultedRecordCount total number of records in the files to be read

Together with the existing metrics this completes two axes that are currently incomplete:

  • pruning efficiency: scannedManifests, skippedManifests, skippedTableFiles
  • scan cost: resultedTableFiles, resultedTableFilesSize, resultedRecordCount

They are computed at the single existing reporting site in AbstractFileStoreScan#plan,
inside the existing if (scanMetrics != null) block, so there is no cost when metrics are
disabled:

// for DELTA and CHANGELOG scan modes the result contains both ADD and DELETE entries,
// only ADD entries will actually be read, so size and record count only count them
long resultedTableFilesSize = 0L;
long resultedRecordCount = 0L;
for (ManifestEntry entry : result) {
    if (entry.kind() == FileKind.ADD) {
        resultedTableFilesSize += entry.file().fileSize();
        resultedRecordCount += entry.file().rowCount();
    }
}

Note the deliberate asymmetry: size and record count only cover FileKind.ADD entries,
because for DELTA / CHANGELOG scan modes the result also carries DELETE entries whose
bytes will never be read. The existing resultedTableFiles keeps its current semantics (all
kinds), so no existing metric changes value.

The three metrics are also exposed as Spark custom metrics (PaimonMetrics,
SparkMetricRegistry, PaimonBaseScan#supportedCustomMetrics); the size one uses
PaimonSizeSumMetric so the SQL tab renders 18.2 GiB rather than a raw byte count. The
Flink side needs no change, FlinkMetricRegistry forwards the new gauges automatically.

This PR also fixes a typo in the planningDuration metric description (planing ->
planning), which is in the same block of code.

Compatibility

No format change. Both source values are already persisted fields
(ManifestFileMeta#_NUM_ADDED_FILES, DataFileMeta#_FILE_SIZE); nothing new is written and
the write path is never entered, so old tables, old readers and rolling Flink upgrades are
unaffected. The new gauges are purely additive — no existing metric name or semantic changes,
so existing dashboards keep working.

For reference, Iceberg reports the equivalent counters in its ScanMetrics:
TOTAL_DATA_MANIFESTS, SKIPPED_DATA_MANIFESTS and TOTAL_FILE_SIZE_IN_BYTES.

Follow-ups (not in this PR)

  • Break scanDuration down into manifest IO vs. filtering, so a slow plan can be attributed.
  • Merge-on-read cost (deletion vectors / delete rows), which the scan file count does not reflect.
  • resultedLevel0Files, to show how much un-compacted data a query has to read.

Tests

  • ScanMetricsTest — extended to cover the three new gauges (registration, initial values and
    values after each report).
  • PaimonMetricTestcheckMetrics now also asserts resultedRecordCount and that
    resultedTableFilesSize > 0, plus a new assertion that a scan without any filter cannot
    prune any manifest (skippedManifests == 0). Passes under both -Pspark3 (Scala 2.12) and
    -Pspark4 (Scala 2.13).
  • FileStoreSourceMetricsTest — unchanged and still passing, confirming the Flink side needs
    no change.

… count in scan metrics

Scan metrics reported how much was read but not whether that amount was
reasonable. lastScannedManifests only reported the count after manifest level
filtering, so the pruning ratio could not be computed, and only file counts were
reported, so the data volume of a scan could not be estimated.

Add lastScanSkippedManifests, lastScanResultedTableFilesSize and
lastScanResultedRecordCount, computed at the existing reporting site in
AbstractFileStoreScan#plan from values that are already in memory, and expose
them as Spark custom metrics.
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.

1 participant