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

[zhuxiangyi]

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).
• PaimonMetricTest — checkMetrics 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.

[JingsongLi]

Pruning counts and planned bytes/records have real diagnostic value, but the new ADD-only calculation does not represent all supported read paths. Details are inline. The two ScanMetricsTest cases pass with head classes on JDK 8; they validate gauge registration/reporting but do not exercise the DELTA plan that exposes the issue. I have not rerun the Spark metrics suite.

[JingsongLi]

[P2] Include before-files when reporting the cost of change reads

The assumption that DELETE entries are never read is false for SnapshotReaderImpl.readChanges(): it uses this DELTA plan's DELETE entries as IncrementalSplit.beforeFiles, and IncrementalDiffSplitRead/IncrementalChangelogReadProvider actually read those files to produce changes. FollowUpScanner reaches this path for overwrite changes, and AllDeltaFollowUpScanner uses it for delta reads. A deletion-only change can therefore report zero resulted bytes/records while scanning the old files; a replacement underreports the before side.

Keep the metrics' documented meaning tied to the files the consumer will actually read, for example by calculating them after the normal-read versus change-read split selection. Do not unconditionally sum all DELETE entries for ordinary ADD-only reads. Add a DELTA/overwrite-changelog regression alongside the existing batch gauge tests.

[zhuxiangyi]

Thanks — you are right about readChanges(), which reads the DELETE entries as before files.

Moved the size/record fold from AbstractFileStoreScan.plan() into SnapshotReaderImpl, after the read path is chosen: read() counts ADD only (also for a DELTA plan), readChanges() counts ADD + DELETE, readIncrementalDiff() counts the ADD entries of both snapshots. ScanMetrics.reportResultedFiles() carries it; gauge names are unchanged, so Spark/Flink/docs are untouched. resultedTableFiles still counts every entry, as before.

Added ScanResultedFilesMetricsTest covering each path through the real reader and stream scan, including the deletion-only change and the overwrite path of FollowUpScanner for both append and primary key tables. On the previous commit its four change-read/diff cases fail and the four ADD-only cases pass.

[JingsongLi]

We do not want to modify the core "plan" execution path again. I am closing this PR for now; if you believe this is critical, please provide a comprehensive description outlining its business value.