JingsongLi commented on code in PR #9330:
URL: https://github.com/apache/paimon/pull/9330#discussion_r3828256518
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java:
##########
@@ -459,12 +466,126 @@ private Set<Path> deletePreviousDataFile(Path
partitionPath, int partitionLevels
@Override
public void truncateTable() {
- throw new UnsupportedOperationException();
+ // Data files only. The partition directories stay, and so do their
catalog registrations:
+ // emptying a table does not redefine which partitions it has.
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ try {
+ deletePreviousDataFile(new Path(location), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate table %s.",
tableIdentifier.getFullName()),
+ e);
+ }
+ return;
+ }
+ // Emptying the table is emptying every partition it has, and which
those are is answered
+ // by whatever the table reads its partitions from.
+ if (partitionManager != null) {
+ truncate(registeredPartitions(Collections.emptyMap()));
+ return;
+ }
+ // Filesystem partition discovery: the partition directories the scan
reads are the table.
+ // A directory that does not parse into the partition keys is not one
of them, so
+ // truncating leaves it alone.
+ for (Pair<LinkedHashMap<String, String>, Path> partition :
+ PartitionPathUtils.searchPartSpecAndPaths(
+ fileIO,
+ new Path(location),
+ partitionKeys.size(),
+ partitionKeys,
+ formatTablePartitionOnlyValueInPath,
+ null,
+ null,
+ defaultPartName)) {
+ try {
+ deletePreviousDataFile(partition.getRight(), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate partition %s of table %s.",
+ partition.getLeft(),
tableIdentifier.getFullName()),
+ e);
+ }
+ }
}
@Override
public void truncatePartitions(List<Map<String, String>> partitionSpecs) {
- throw new UnsupportedOperationException();
+ if (partitionManager == null) {
+ truncate(partitionSpecs);
+ return;
+ }
+ List<Map<String, String>> partitions = new ArrayList<>();
+ partitionSpecs.forEach(spec ->
partitions.addAll(registeredPartitions(spec)));
+ truncate(partitions);
+ }
+
+ /**
+ * The registered partitions named by {@code prefix}, which may name only
the leading partition
+ * keys, or none of them. The catalog says which partitions a
catalog-managed table has, so
+ * truncating neither empties nor registers a directory still waiting for
MSCK REPAIR TABLE.
+ */
+ private List<Map<String, String>> registeredPartitions(Map<String, String>
prefix) {
+ return partitionManager.listPartitions(prefix, null).stream()
+ .map(Partition::spec)
+ .collect(Collectors.toList());
+ }
+
+ private void truncate(List<Map<String, String>> partitionSpecs) {
+ long truncateTime = System.currentTimeMillis();
+ Set<Path> clearedPartitionPaths = new HashSet<>();
+ // Statistics are keyed by the spec that named the partition, so only
a complete one can
+ // seed them; a prefix reaches here only for a table with nowhere to
report to.
+ Map<Map<String, String>, PartitionStatistics> emptied = new
LinkedHashMap<>();
+ RuntimeException failure = null;
+ for (Map<String, String> partitionSpec : partitionSpecs) {
+ Path partitionPath =
+ buildPartitionPath(
+ location,
+ partitionSpec,
+ formatTablePartitionOnlyValueInPath,
+ partitionKeys);
+ try {
+ clearedPartitionPaths.addAll(
+ deletePreviousDataFile(
+ partitionPath, partitionKeys.size() -
partitionSpec.size()));
+ } catch (Exception e) {
+ failure =
+ new RuntimeException(
+ String.format(
+ "Failed to truncate partition %s of
table %s.",
+ partitionSpec,
tableIdentifier.getFullName()),
+ e);
+ break;
+ }
+ if (partitionSpec.size() == partitionKeys.size()) {
+ emptied.put(partitionSpec, emptyStatistics(partitionSpec,
truncateTime));
+ }
+ }
+ if (partitionManager != null) {
+ // Truncating states that the partition holds nothing, whoever
deleted the files, so
+ // one that was already empty reports zero as well. An overwrite
reports only what it
+ // removed itself, so that concurrent writers do not each claim
the whole subtree;
+ // truncation makes the claim on purpose. What a failed truncation
emptied is reported
+ // too, so the catalog stops describing files that are gone.
+ reportPartitions(
Review Comment:
**[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.
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java:
##########
@@ -459,12 +466,126 @@ private Set<Path> deletePreviousDataFile(Path
partitionPath, int partitionLevels
@Override
public void truncateTable() {
- throw new UnsupportedOperationException();
+ // Data files only. The partition directories stay, and so do their
catalog registrations:
+ // emptying a table does not redefine which partitions it has.
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ try {
+ deletePreviousDataFile(new Path(location), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate table %s.",
tableIdentifier.getFullName()),
+ e);
+ }
+ return;
+ }
+ // Emptying the table is emptying every partition it has, and which
those are is answered
+ // by whatever the table reads its partitions from.
+ if (partitionManager != null) {
+ truncate(registeredPartitions(Collections.emptyMap()));
+ return;
+ }
+ // Filesystem partition discovery: the partition directories the scan
reads are the table.
+ // A directory that does not parse into the partition keys is not one
of them, so
+ // truncating leaves it alone.
+ for (Pair<LinkedHashMap<String, String>, Path> partition :
+ PartitionPathUtils.searchPartSpecAndPaths(
+ fileIO,
+ new Path(location),
+ partitionKeys.size(),
+ partitionKeys,
+ formatTablePartitionOnlyValueInPath,
+ null,
+ null,
+ defaultPartName)) {
+ try {
+ deletePreviousDataFile(partition.getRight(), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate partition %s of table %s.",
+ partition.getLeft(),
tableIdentifier.getFullName()),
+ e);
+ }
+ }
}
@Override
public void truncatePartitions(List<Map<String, String>> partitionSpecs) {
- throw new UnsupportedOperationException();
+ if (partitionManager == null) {
+ truncate(partitionSpecs);
+ return;
+ }
+ List<Map<String, String>> partitions = new ArrayList<>();
+ partitionSpecs.forEach(spec ->
partitions.addAll(registeredPartitions(spec)));
Review Comment:
**[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.
##########
docs/docs/flink/sql-write.mdx:
##########
@@ -142,6 +142,14 @@ TRUNCATE TABLE my_table;
</Tabs>
+On a Format Table read through Paimon (`format-table.implementation = paimon`,
the default),
+`TRUNCATE TABLE` deletes the data files and keeps the partitions: their
directories remain, and
+with `metastore.partitioned-table = true` so do their catalog registrations,
whose statistics are
+replaced with zero. That setting also makes the catalog the answer to which
partitions the table
+has, so truncating empties those and leaves an unregistered directory alone.
Flink has no
+`TRUNCATE TABLE ... PARTITION`; to empty one partition, overwrite it with an
empty result under a
+static `PARTITION` clause.
Review Comment:
**[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.
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java:
##########
@@ -459,12 +466,126 @@ private Set<Path> deletePreviousDataFile(Path
partitionPath, int partitionLevels
@Override
public void truncateTable() {
- throw new UnsupportedOperationException();
+ // Data files only. The partition directories stay, and so do their
catalog registrations:
+ // emptying a table does not redefine which partitions it has.
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ try {
+ deletePreviousDataFile(new Path(location), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate table %s.",
tableIdentifier.getFullName()),
+ e);
+ }
+ return;
+ }
+ // Emptying the table is emptying every partition it has, and which
those are is answered
+ // by whatever the table reads its partitions from.
+ if (partitionManager != null) {
+ truncate(registeredPartitions(Collections.emptyMap()));
+ return;
+ }
+ // Filesystem partition discovery: the partition directories the scan
reads are the table.
+ // A directory that does not parse into the partition keys is not one
of them, so
+ // truncating leaves it alone.
+ for (Pair<LinkedHashMap<String, String>, Path> partition :
+ PartitionPathUtils.searchPartSpecAndPaths(
+ fileIO,
+ new Path(location),
+ partitionKeys.size(),
+ partitionKeys,
+ formatTablePartitionOnlyValueInPath,
+ null,
+ null,
+ defaultPartName)) {
+ try {
+ deletePreviousDataFile(partition.getRight(), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate partition %s of table %s.",
+ partition.getLeft(),
tableIdentifier.getFullName()),
+ e);
+ }
+ }
}
@Override
public void truncatePartitions(List<Map<String, String>> partitionSpecs) {
- throw new UnsupportedOperationException();
+ if (partitionManager == null) {
+ truncate(partitionSpecs);
+ return;
+ }
+ List<Map<String, String>> partitions = new ArrayList<>();
+ partitionSpecs.forEach(spec ->
partitions.addAll(registeredPartitions(spec)));
+ truncate(partitions);
+ }
+
+ /**
+ * The registered partitions named by {@code prefix}, which may name only
the leading partition
+ * keys, or none of them. The catalog says which partitions a
catalog-managed table has, so
+ * truncating neither empties nor registers a directory still waiting for
MSCK REPAIR TABLE.
+ */
+ private List<Map<String, String>> registeredPartitions(Map<String, String>
prefix) {
+ return partitionManager.listPartitions(prefix, null).stream()
+ .map(Partition::spec)
+ .collect(Collectors.toList());
+ }
+
+ private void truncate(List<Map<String, String>> partitionSpecs) {
+ long truncateTime = System.currentTimeMillis();
+ Set<Path> clearedPartitionPaths = new HashSet<>();
+ // Statistics are keyed by the spec that named the partition, so only
a complete one can
+ // seed them; a prefix reaches here only for a table with nowhere to
report to.
+ Map<Map<String, String>, PartitionStatistics> emptied = new
LinkedHashMap<>();
+ RuntimeException failure = null;
+ for (Map<String, String> partitionSpec : partitionSpecs) {
+ Path partitionPath =
+ buildPartitionPath(
+ location,
+ partitionSpec,
+ formatTablePartitionOnlyValueInPath,
+ partitionKeys);
+ try {
+ clearedPartitionPaths.addAll(
+ deletePreviousDataFile(
+ partitionPath, partitionKeys.size() -
partitionSpec.size()));
Review Comment:
**[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
