[core][spark][flink] Support TRUNCATE TABLE on format tables

[sundapeng]

Purpose

TRUNCATE TABLE did not work on a Format Table. TRUNCATE TABLE t was rejected in the planner
because PaimonFormatTable does not implement TruncatableTable, and TRUNCATE TABLE t PARTITION (...) reached PaimonPartitionManagement, which serves only FileStoreTable. Both
FormatTableCommit entry points threw an empty UnsupportedOperationException, and
FlinkFormatTableSink did not implement SupportsTruncate either. The workaround was an INSERT OVERWRITE ... PARTITION (...) of an empty result, which does not extend to the whole table.

FormatTableCommit now removes the data files, reusing deletePreviousDataFile, the primitive a
static INSERT OVERWRITE already clears partition directories with. Only data files go: the
partition directories stay, and so do their catalog registrations, so SHOW PARTITIONS returns
what it returned before (SPARK-34418). Staging trees of concurrent writers are left alone, on the
same judgement FormatTableScan reads with.

Which partitions the table has is answered by whatever the table reads its partitions from: the
catalog when it manages them, the partition directories the scan parses otherwise. So truncating
neither empties nor registers a directory still waiting for MSCK REPAIR TABLE, and never deletes
a file the table cannot read.

What the emptied partitions hold is reported to the catalog as an exact zero, carried with
replaceStatistics, so a truncated partition stops describing files that are gone. An overwrite
reports only the files it removed itself, so that concurrent writers do not each claim the whole
subtree; truncation states that the partition holds nothing whoever deleted the files, so one that
was already empty reports zero as well. A Format Table has no snapshot to make the whole truncation
atomic, so a failure part-way reports what it emptied before propagating.

SupportsTruncate arrived in Flink 1.18, so FlinkFormatTableSink is split the way FlinkTableSink
already is and 1.16 and 1.17 get one that does not implement it. Flink has no TRUNCATE TABLE ... PARTITION, so only whole-table truncation is wired there.

One consequence worth a second opinion: on the Spark versions whose strategy plans an unconditional
DELETE FROM as a truncate, implementing TruncatableTable makes DELETE FROM t empty a Format
Table as well. Happy to block it instead.

Tests

FormatTableCommitTest and FormatTableCommitStatisticsTest cover both entry points: what is
emptied and what is left alone, prefix specs, the value-only default partition, a failure part-way
through, and the zero statistics that are reported. FormatTablePartitionManagementTest,
CatalogManagedPartitionTest and FormatTableTestBase cover the Spark statements end to end, and
FormatTableITCase covers Flink.

[JingsongLi]

Inline review comments for the four selected findings.

[JingsongLi]

[P1] Fail when an advertised deletion did not happen. Could we make the truncate path fail when deletePreviousDataFile sees FileIO.delete(...) == false while the file still exists? The FileIO contract says false means deletion was unsuccessful. The helper currently only omits that path from clearedPartitionPaths; execution continues and this method records exact-zero statistics below (or a table truncate returns normally), so TRUNCATE can report success while old rows remain readable. Please check exists after a false result and tolerate only the concurrent-already-deleted case, with a regression FileIO that returns false without removing the file.

[JingsongLi]

[P2] Batch complete partition lookups. Spark resolves a partial spec to complete identifiers and the new Spark wrapper already performs a batched listPartitionsByNames existence check. This loop then issues one additional listPartitions call per leaf, so a broad partial spec produces N+2 sequential catalog reads and can hit latency, throttling, or timeout limits. Please batch complete specs through listPartitionsByNames and reserve prefix listing for genuinely incomplete specs; a catalog-call-count test with multiple identifiers would cover this.

[JingsongLi]

[P2] Remove or implement this empty-overwrite workaround. This does not currently clear a Format Table partition: FormatTableSinkWriter.close() calls tableCommit.commit(...) only when prepareCommit() returns a non-empty message list. An empty result therefore performs no overwrite commit or deletion. I reproduced this with a static INSERT OVERWRITE ... SELECT ... WHERE false; the pre-existing row remained. Please either make zero-row static overwrites execute the overwrite commit and add an IT, or remove this recommendation.

[JingsongLi]

[P2] Preserve the primary deletion failure. If a filesystem deletion failure has already been saved in failure and this reconciliation call also throws, the catalog exception escapes and masks the original partition-specific deletion error. Please catch the reporting exception, attach it as suppressed when failure is non-null, and rethrow the original deletion failure; a dual-failure test would lock down the diagnostic contract.