diff --git a/docs/docs/flink/procedures/table-operations.md b/docs/docs/flink/procedures/table-operations.md index 503796f269cf..a2b0655de6bb 100644 --- a/docs/docs/flink/procedures/table-operations.md +++ b/docs/docs/flink/procedures/table-operations.md @@ -379,6 +379,9 @@ Mark partitions as done. Arguments: - `partitions`: semicolon-separated partition specs. +A Format Table only supports the `success-file` action, which creates an empty `_SUCCESS` file in +each partition directory. + **Syntax** ```sql diff --git a/docs/docs/spark/format-table.md b/docs/docs/spark/format-table.md index 70e3ce6aa9ff..883c2b453f08 100644 --- a/docs/docs/spark/format-table.md +++ b/docs/docs/spark/format-table.md @@ -112,6 +112,42 @@ change the table's managed/external ownership. ## Writes and Maintenance +### Partition Mark Done + +With `'partition.end-input-to-done' = 'true'`, a Spark write creates an empty `_SUCCESS` file in +each partition it writes, as Hadoop jobs do, so that downstream jobs waiting for the file can +start. Unpartitioned tables are not marked: no `_SUCCESS` is written at the table root, and the +procedure rejects them. A Format Table only supports the default `partition.mark-done-action`, +`success-file`; any other action fails the statement before data is written. Only the default +`format-table.implementation = paimon` marks partitions: writes to a table that uses `engine` do +not, and the procedure rejects it. + +```sql +CREATE TABLE my_table (id INT, dt STRING) USING parquet PARTITIONED BY (dt) +TBLPROPERTIES ('partition.end-input-to-done' = 'true'); + +-- creates dt=20240101/_SUCCESS, even when the query returns no rows +INSERT OVERWRITE my_table PARTITION (dt = '20240101') SELECT id FROM source; +``` + +- An `INSERT OVERWRITE` that names every partition column marks that partition even without rows; + an `INSERT INTO` that writes no rows marks nothing. +- Before a write or `TRUNCATE` changes a partition, its `_SUCCESS` is deleted, including a + partition an overwrite clears without writing to. After the commit, the write creates it again + in the partitions it wrote, so a partition that was only cleared is left without it. If marking + fails after the commit, the statement fails with the data written; mark the partitions with the + [mark_partition_done](./procedures/metadata#mark_partition_done) procedure. +- Without `partition.end-input-to-done`, writes leave `_SUCCESS` files alone. When marking + partitions with the procedure, enable it as well, so that a later write does not leave a stale + file next to changed data. +- A write marks a partition done when that write finishes, so a partition must have one writer at + a time. Writers are not coordinated: with two concurrent writes to a partition, the first to + finish marks it while the other may still be publishing files. +- The file follows `format-table.partition-path-only-value` and is not read as data. A + catalog-managed partition at a custom location, or not registered, is skipped. +- Flink writes delete the file of the partitions they change but do not create it; run the + procedure after them. + ### Partition File Formats Catalog-managed partitions can override the table's `file.format` through the partition `options` diff --git a/docs/docs/spark/procedures/metadata.md b/docs/docs/spark/procedures/metadata.md index bddd860eb5fa..453cc7470337 100644 --- a/docs/docs/spark/procedures/metadata.md +++ b/docs/docs/spark/procedures/metadata.md @@ -88,6 +88,9 @@ Mark partitions as done. - `table` (`STRING`, required): the target table identifier. - `partitions` (`STRING`, required): partitions need to be mark done, If you specify multiple partitions, delimiter is ';'. +A Format Table only supports the `success-file` action; see +[Format Table Partition Mark Done](../format-table#partition-mark-done). + ```sql -- mark single partition done CALL sys.mark_partition_done(table => 'default.T', partitions => 'day=2024-07-01'); diff --git a/paimon-core/src/main/java/org/apache/paimon/partition/actions/PartitionMarkDoneAction.java b/paimon-core/src/main/java/org/apache/paimon/partition/actions/PartitionMarkDoneAction.java index 680aa13277f0..899194daf10a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/partition/actions/PartitionMarkDoneAction.java +++ b/paimon-core/src/main/java/org/apache/paimon/partition/actions/PartitionMarkDoneAction.java @@ -20,8 +20,10 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.PartitionMarkDone; import org.apache.paimon.table.PartitionModification; +import org.apache.paimon.table.format.FormatTableSuccessFileMarkDoneAction; import org.apache.paimon.utils.StringUtils; import java.io.Closeable; @@ -42,6 +44,13 @@ default void open(FileStoreTable fileStoreTable, CoreOptions options) {} void markDone(String partition) throws Exception; + /** Marks several partitions done; an implementation may batch them. */ + default void markDone(List partitions) throws Exception { + for (String partition : partitions) { + markDone(partition); + } + } + static List createActions( ClassLoader cl, FileStoreTable fileStoreTable, CoreOptions options) { return options.partitionMarkDoneActions().stream() @@ -82,6 +91,29 @@ static List createActions( .collect(Collectors.toList()); } + /** Creates the actions of a {@link FormatTable}, which only supports {@code success-file}. */ + static List createActions( + ClassLoader cl, FormatTable formatTable, CoreOptions options) { + return options.partitionMarkDoneActions().stream() + .map( + action -> { + if (action != CoreOptions.PartitionMarkDoneAction.SUCCESS_FILE) { + throw new UnsupportedOperationException( + String.format( + "Format Table %s only supports the '%s' partition " + + "mark done action, but '%s' is " + + "configured in %s.", + formatTable.fullName(), + CoreOptions.PartitionMarkDoneAction.SUCCESS_FILE, + action, + PARTITION_MARK_DONE_ACTION.key())); + } + return (PartitionMarkDoneAction) + new FormatTableSuccessFileMarkDoneAction(formatTable); + }) + .collect(Collectors.toList()); + } + static PartitionMarkDoneAction generateCustomMarkDoneAction( ClassLoader cl, CoreOptions options) { if (StringUtils.isNullOrWhitespaceOnly(options.partitionMarkDoneCustomClass())) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java index a57bf1fb97d6..b4d57cdc652b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java @@ -86,22 +86,28 @@ public BatchTableCommit newCommit() { table.partitionManager() != null && !table.partitionKeys().isEmpty() ? options.formatTableCommitPublishThreadNum() : 1; - return new FormatTableCommit( - table.location(), - table.partitionKeys(), - table.fileIO(), - formatTablePartitionOnlyValueInPath, - table.defaultPartName(), - overwrite, - Identifier.fromString(table.fullName()), - staticPartition, - syncHiveUri, - table.catalogContext(), - table.partitionManager(), - options.dynamicPartitionOverwrite(), - cleanupThreadNum, - publishThreadNum, - FormatTablePartitionOptions.fileFormat(table.options(), null)); + FormatTableCommit commit = + new FormatTableCommit( + table.location(), + table.partitionKeys(), + table.fileIO(), + formatTablePartitionOnlyValueInPath, + table.defaultPartName(), + overwrite, + Identifier.fromString(table.fullName()), + staticPartition, + syncHiveUri, + table.catalogContext(), + table.partitionManager(), + options.dynamicPartitionOverwrite(), + cleanupThreadNum, + publishThreadNum, + FormatTablePartitionOptions.fileFormat(table.options(), null)); + // A partition marked done must not keep its success file while its data changes. + return commit.withSuccessFileDeletion( + !table.partitionKeys().isEmpty() + && options.toConfiguration() + .get(CoreOptions.PARTITION_MARK_DONE_WHEN_END_INPUT)); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java index 2b67c267518a..49a5ef8f4c44 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java @@ -33,6 +33,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.partition.actions.SuccessFileMarkDoneAction; import org.apache.paimon.stats.Statistics; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.sink.CommitMessage; @@ -104,6 +105,9 @@ public class FormatTableCommit implements BatchTableCommit { private final int publishThreadNum; @Nullable private final String fileFormat; + private List> committedPartitions = Collections.emptyList(); + private boolean deletesSuccessFiles = false; + /** * Creates a legacy committer whose write format is unknown. Appends and overwrites reject * partitions with an explicit {@code file.format}, including values stored by earlier @@ -207,6 +211,7 @@ public FormatTableCommit( @Override public void commit(List commitMessages) { + committedPartitions = Collections.emptyList(); try { // One reading for the whole commit: stat-ing each file back costs a request per // file for a coarser number. @@ -237,6 +242,16 @@ public void commit(List commitMessages) { if (overwrite && partitionManager != null) { targetPartitions.forEach(partition -> reportTargetSpecs.add(partition.spec())); } + if (!overwrite && deletesSuccessFiles) { + Set writtenPartitionPaths = new LinkedHashSet<>(); + for (TwoPhaseCommitMessage message : messages) { + Path targetPath = message.getCommitter().targetPath(); + if (targetPath != null) { + writtenPartitionPaths.add(targetPath.getParent()); + } + } + deleteSuccessFiles(new ArrayList<>(writtenPartitionPaths), cleanupThreadNum); + } Path staticPartitionPath = null; if (staticPartitions != null && !staticPartitions.isEmpty()) { @@ -380,7 +395,7 @@ public void commit(List commitMessages) { } } } - + committedPartitions = committedPartitionSpecs(writtenPartitionSpecs); } catch (Throwable failure) { // Cleanup restores the caller's interrupt before failing. Clear it only while aborting // staging output, then restore it; an abort failure is secondary to the commit failure @@ -499,6 +514,40 @@ private List loadCommitTargetPartitions( : loadPartitionsByNames(writtenPartitionSpecs); } + /** + * Partitions written by the last successful commit, including the full static partition of an + * overwrite that wrote no rows. Empty after a failed commit or for an unpartitioned table. + */ + public List> committedPartitions() { + return committedPartitions; + } + + /** + * Whether to delete the success file of a partition before writing to it or clearing it, so + * that a stale one is never seen next to the changed data. + */ + FormatTableCommit withSuccessFileDeletion(boolean deletesSuccessFiles) { + this.deletesSuccessFiles = deletesSuccessFiles; + return this; + } + + private List> committedPartitionSpecs( + Set> writtenPartitionSpecs) { + if (partitionKeys == null || partitionKeys.isEmpty()) { + return Collections.emptyList(); + } + Set> committed = new LinkedHashSet<>(); + if (staticPartitions != null && staticPartitions.size() == partitionKeys.size()) { + LinkedHashMap orderedStaticPartition = new LinkedHashMap<>(); + for (String partitionKey : partitionKeys) { + orderedStaticPartition.put(partitionKey, staticPartitions.get(partitionKey)); + } + committed.add(orderedStaticPartition); + } + committed.addAll(writtenPartitionSpecs); + return Collections.unmodifiableList(new ArrayList<>(committed)); + } + private Set> writtenPartitionSpecs(List messages) { if (partitionKeys == null || partitionKeys.isEmpty()) { return Collections.emptySet(); @@ -914,6 +963,9 @@ private Set deletePreviousDataFile(Path partitionPath, int partitionLevels private Set deletePreviousDataFiles( List partitionPaths, int partitionLevels, int threadNum) throws IOException { + if (deletesSuccessFiles) { + deleteSuccessFiles(partitionDirectories(partitionPaths, partitionLevels), threadNum); + } Iterator dataFiles = previousDataFiles(partitionPaths, partitionLevels); Set clearedPartitionPaths = new HashSet<>(); try { @@ -1186,6 +1238,69 @@ private static Throwable unwrapUncheckedIOException(Throwable failure) { return unwrapped; } + /** The partition directories the given number of levels below each path. */ + private List partitionDirectories(List paths, int partitionLevels) + throws IOException { + if (partitionLevels == 0) { + return paths; + } + List keys = + partitionKeys.subList(partitionKeys.size() - partitionLevels, partitionKeys.size()); + List directories = new ArrayList<>(); + for (Path path : paths) { + if (!fileIO.exists(path)) { + continue; + } + for (Pair, Path> partition : + PartitionPathUtils.searchPartSpecAndPaths( + fileIO, + path, + partitionLevels, + keys, + formatTablePartitionOnlyValueInPath, + null, + null, + defaultPartName)) { + directories.add(partition.getRight()); + } + } + return directories; + } + + private void deleteSuccessFiles(List partitionPaths, int threadNum) throws IOException { + if (partitionPaths.isEmpty()) { + return; + } + try { + executeSideEffects( + COMMIT_EXECUTOR, + this::deleteSuccessFile, + partitionPaths.iterator(), + threadNum, + ignored -> {}); + } catch (UncheckedIOException e) { + throw (IOException) unwrapUncheckedIOException(e); + } + } + + private List deleteSuccessFile(Path partitionPath) { + Path successPath = new Path(partitionPath, SuccessFileMarkDoneAction.SUCCESS_FILE_NAME); + try { + if (fileIO.delete(successPath, false) || !fileIO.exists(successPath)) { + return Collections.emptyList(); + } + } catch (FileNotFoundException ignore) { + return Collections.emptyList(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + throw new UncheckedIOException( + new IOException( + String.format( + "Failed to delete success file %s of table %s.", + successPath, tableIdentifier.getFullName()))); + } + /** Deletes one listed file and reports its parent when this commit removed the file. */ private List deleteAndReportCleared(FileStatus file) { try { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSuccessFileMarkDoneAction.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSuccessFileMarkDoneAction.java new file mode 100644 index 000000000000..00b83e5bef22 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableSuccessFileMarkDoneAction.java @@ -0,0 +1,154 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.actions.PartitionMarkDoneAction; +import org.apache.paimon.partition.actions.SuccessFileMarkDoneAction; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.utils.PartitionPathUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * A {@link PartitionMarkDoneAction} which creates an empty "_SUCCESS" file in a {@link FormatTable} + * partition directory, as Hadoop jobs do. Catalog-managed partitions at a custom location or not + * registered are skipped, so the file is only written inside the table directory. + */ +public class FormatTableSuccessFileMarkDoneAction implements PartitionMarkDoneAction { + + private static final Logger LOG = + LoggerFactory.getLogger(FormatTableSuccessFileMarkDoneAction.class); + + private final FileIO fileIO; + private final Path tablePath; + private final List partitionKeys; + private final boolean onlyValueInPath; + @Nullable private final FormatTablePartitionManager partitionManager; + private final String tableName; + + public FormatTableSuccessFileMarkDoneAction(FormatTable table) { + checkArgument( + !table.partitionKeys().isEmpty(), + "Format Table %s is not partitioned, so it has no partition to mark done.", + table.fullName()); + this.fileIO = table.fileIO(); + this.tablePath = new Path(table.location()); + this.partitionKeys = table.partitionKeys(); + this.onlyValueInPath = + new CoreOptions(table.options()).formatTablePartitionOnlyValueInPath(); + this.partitionManager = table.partitionManager(); + this.tableName = table.fullName(); + } + + @Override + public void markDone(String partition) throws Exception { + markDone(Collections.singletonList(partition)); + } + + @Override + public void markDone(List partitions) throws Exception { + List> specs = new ArrayList<>(partitions.size()); + for (String partition : partitions) { + LinkedHashMap spec = + PartitionPathUtils.extractPartitionSpecFromPath( + new Path(partition), partitionKeys); + checkArgument( + spec != null, + "Partition '%s' does not name the partition keys %s of Format Table %s in " + + "order.", + partition, + partitionKeys, + tableName); + specs.add(spec); + } + + Map, Partition> registered = registeredPartitions(specs); + for (LinkedHashMap spec : specs) { + if (registered != null && !inTableDirectory(spec, registered.get(spec))) { + continue; + } + fileIO.overwriteFileUtf8( + new Path(partitionPath(spec), SuccessFileMarkDoneAction.SUCCESS_FILE_NAME), ""); + } + } + + /** One catalog lookup for all partitions, or null when partitions are not catalog-managed. */ + @Nullable + private Map, Partition> registeredPartitions( + List> specs) { + if (partitionManager == null) { + return null; + } + Map, Partition> registered = new HashMap<>(); + for (Partition partition : + partitionManager.listPartitionsByNames( + new ArrayList<>(new LinkedHashSet<>(specs)))) { + registered.put(partition.spec(), partition); + } + return registered; + } + + private boolean inTableDirectory(Map spec, @Nullable Partition partition) { + if (partition == null) { + LOG.warn( + "Skip creating the success file of partition {} of Format Table {}: the " + + "catalog does not register this partition.", + spec, + tableName); + return false; + } + String customLocation = FormatTablePartitionPathResolver.customLocation(partition); + if (customLocation != null) { + LOG.warn( + "Skip creating the success file of partition {} of Format Table {}: the " + + "partition is registered at custom location '{}', outside the " + + "table directory.", + spec, + tableName, + customLocation); + return false; + } + return true; + } + + private Path partitionPath(LinkedHashMap spec) { + return FormatTablePartitionPathResolver.defaultPartitionPath( + tablePath, spec, onlyValueInPath); + } + + @Override + public void close() {} +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitMarkDoneTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitMarkDoneTest.java new file mode 100644 index 000000000000..100c53bdaa88 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitMarkDoneTest.java @@ -0,0 +1,254 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.TwoPhaseOutputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.table.sink.CommitMessage; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests for the partition mark done support of {@link FormatTableCommit}. */ +class FormatTableCommitMarkDoneTest { + + @TempDir java.nio.file.Path tempDir; + + private final LocalFileIO fileIO = LocalFileIO.create(); + private Path tablePath; + + @BeforeEach + void setUp() throws IOException { + tablePath = new Path(tempDir.toUri()); + for (String partition : Arrays.asList("dt=1/hh=08", "dt=1/hh=09", "dt=2/hh=08")) { + fileIO.overwriteFileUtf8(new Path(tablePath, partition + "/data.csv"), "1"); + fileIO.overwriteFileUtf8(new Path(tablePath, partition + "/_SUCCESS"), "{}"); + } + // Marked done while empty. + fileIO.overwriteFileUtf8(new Path(tablePath, "dt=1/hh=10/_SUCCESS"), "{}"); + } + + @Test + void testCommittedPartitions() { + FormatTableCommit append = commit(fileIO, false, null, false); + append.commit(Arrays.asList(message("dt=3/hh=08/a"), message("dt=3/hh=09/b"))); + assertThat(append.committedPartitions()).containsExactly(spec("3", "08"), spec("3", "09")); + assertThat(new ArrayList<>(append.committedPartitions().get(0).keySet())) + .containsExactly("dt", "hh"); + + // An overwrite of a full static partition reports it even without rows. + Map staticPartition = new HashMap<>(); + staticPartition.put("hh", "00"); + staticPartition.put("dt", "4"); + FormatTableCommit emptyOverwrite = commit(fileIO, true, staticPartition, false); + emptyOverwrite.commit(Collections.emptyList()); + assertThat(emptyOverwrite.committedPartitions()).containsExactly(spec("4", "00")); + + FormatTableCommit prefixOverwrite = + commit(fileIO, true, Collections.singletonMap("dt", "3"), false); + prefixOverwrite.commit(Collections.singletonList(message("dt=3/hh=09/c"))); + assertThat(prefixOverwrite.committedPartitions()).containsExactly(spec("3", "09")); + } + + @Test + void testFailedCommitReportsNothing() throws IOException { + FormatTableCommit commit = commit(fileIO, false, null, false); + commit.commit(Collections.singletonList(message("dt=3/hh=08/a"))); + + TwoPhaseOutputStream.Committer failing = mock(TwoPhaseOutputStream.Committer.class); + when(failing.targetPath()).thenReturn(new Path(tablePath, "dt=3/hh=09/b")); + doThrow(new IOException("publish failed")).when(failing).commit(any()); + assertThatThrownBy( + () -> commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(failing)))); + assertThat(commit.committedPartitions()).isEmpty(); + } + + @Test + void testOverwriteDeletesSuccessFilesOfClearedPartitions() { + commit(fileIO, true, Collections.singletonMap("dt", "1"), true) + .commit(Collections.singletonList(message("dt=1/hh=09/new"))); + assertThat(successFile("dt=1/hh=08")).isFalse(); + assertThat(successFile("dt=1/hh=09")).isFalse(); + assertThat(successFile("dt=1/hh=10")).isFalse(); + assertThat(successFile("dt=2/hh=08")).isTrue(); + + commit(fileIO, true, null, true).commit(Collections.emptyList()); + assertThat(successFile("dt=2/hh=08")).isFalse(); + } + + @Test + void testValueOnlyOverwriteDeletesSuccessFilesOfClearedPartitions() throws IOException { + Path valueOnly = new Path(tablePath, "value_only"); + for (String partition : Arrays.asList("1/08", "1/09", "2/08")) { + fileIO.overwriteFileUtf8(new Path(valueOnly, partition + "/data.csv"), "1"); + fileIO.overwriteFileUtf8(new Path(valueOnly, partition + "/_SUCCESS"), "{}"); + } + fileIO.overwriteFileUtf8(new Path(valueOnly, "1/10/_SUCCESS"), "{}"); + + commit(valueOnly, true, fileIO, true, Collections.singletonMap("dt", "1"), true, false) + .commit(Collections.singletonList(message("value_only/1/09/new"))); + assertThat(successFile("value_only/1/08")).isFalse(); + assertThat(successFile("value_only/1/09")).isFalse(); + assertThat(successFile("value_only/1/10")).isFalse(); + assertThat(successFile("value_only/2/08")).isTrue(); + } + + @Test + void testDynamicOverwriteDeletesSuccessFilesOfWrittenPartitions() { + commit(fileIO, true, null, true, true) + .commit(Collections.singletonList(message("dt=1/hh=08/new"))); + + assertThat(successFile("dt=1/hh=08")).isFalse(); + assertThat(successFile("dt=1/hh=09")).isTrue(); + } + + @Test + void testAppendAndTruncateDeleteSuccessFiles() { + commit(fileIO, false, null, true) + .commit(Collections.singletonList(message("dt=1/hh=08/new"))); + assertThat(successFile("dt=1/hh=08")).isFalse(); + assertThat(successFile("dt=1/hh=09")).isTrue(); + + commit(fileIO, false, null, true) + .truncatePartitions(Collections.singletonList(Collections.singletonMap("dt", "1"))); + assertThat(successFile("dt=1/hh=09")).isFalse(); + assertThat(successFile("dt=2/hh=08")).isTrue(); + + commit(fileIO, false, null, true).truncateTable(); + assertThat(successFile("dt=2/hh=08")).isFalse(); + } + + @Test + void testSuccessFilesStayWhenDeletionIsOff() { + commit(fileIO, true, null, false).commit(Collections.emptyList()); + + assertThat(successFile("dt=1/hh=08")).isTrue(); + assertThat(exists("dt=1/hh=08/data.csv")).isFalse(); + } + + @Test + void testRefusedDeletionFailsBeforeDataIsDeleted() { + FileIO refusing = + new LocalFileIO() { + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + return !path.getName().equals("_SUCCESS") && super.delete(path, recursive); + } + }; + + assertThatThrownBy(() -> commit(refusing, true, null, true).commit(Collections.emptyList())) + .hasStackTraceContaining("Failed to delete success file"); + assertThat(exists("dt=1/hh=08/data.csv")).isTrue(); + } + + private FormatTableCommit commit( + FileIO fileIO, + boolean overwrite, + @Nullable Map staticPartition, + boolean deletesSuccessFiles) { + return commit(fileIO, overwrite, staticPartition, deletesSuccessFiles, false); + } + + private FormatTableCommit commit( + FileIO fileIO, + boolean overwrite, + @Nullable Map staticPartition, + boolean deletesSuccessFiles, + boolean dynamicPartitionOverwrite) { + return commit( + tablePath, + false, + fileIO, + overwrite, + staticPartition, + deletesSuccessFiles, + dynamicPartitionOverwrite); + } + + private FormatTableCommit commit( + Path location, + boolean onlyValueInPath, + FileIO fileIO, + boolean overwrite, + @Nullable Map staticPartition, + boolean deletesSuccessFiles, + boolean dynamicPartitionOverwrite) { + return new FormatTableCommit( + location.toString(), + Arrays.asList("dt", "hh"), + fileIO, + onlyValueInPath, + PARTITION_DEFAULT_NAME.defaultValue(), + overwrite, + Identifier.create("db", "events"), + staticPartition, + null, + null, + null, + dynamicPartitionOverwrite) + .withSuccessFileDeletion(deletesSuccessFiles); + } + + private CommitMessage message(String file) { + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(tablePath, file)); + return new TwoPhaseCommitMessage(committer); + } + + private boolean successFile(String partition) { + return exists(partition + "/_SUCCESS"); + } + + private boolean exists(String relative) { + try { + return fileIO.exists(new Path(tablePath, relative)); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private static Map spec(String dt, String hh) { + Map spec = new LinkedHashMap<>(); + spec.put("dt", dt); + spec.put("hh", hh); + return spec; + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableSuccessFileMarkDoneActionTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableSuccessFileMarkDoneActionTest.java new file mode 100644 index 000000000000..f334b9a0316b --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableSuccessFileMarkDoneActionTest.java @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.partition.actions.PartitionMarkDoneAction; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.PartitionPathUtils; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import javax.annotation.Nullable; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH; +import static org.apache.paimon.CoreOptions.PARTITION_MARK_DONE_ACTION; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Tests for {@link FormatTableSuccessFileMarkDoneAction}. */ +class FormatTableSuccessFileMarkDoneActionTest { + + @TempDir java.nio.file.Path tempDir; + + private final LocalFileIO fileIO = LocalFileIO.create(); + + @Test + void testWritesAnEmptySuccessFile() throws Exception { + FormatTable table = table(Collections.emptyMap(), null); + Path successPath = path(table, "dt=20240101/hh=08/_SUCCESS"); + fileIO.overwriteFileUtf8(successPath, "{\"creationTime\": 1}"); + + markDone(table, "dt=20240101/hh=08"); + + assertThat(fileIO.getFileStatus(successPath).getLen()).isZero(); + } + + @Test + void testValueOnlyAndEscapedDirectories() throws Exception { + FormatTable table = + table( + Collections.singletonMap( + FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), "true"), + null); + + LinkedHashMap spec = new LinkedHashMap<>(spec("a/b")); + + markDone(table, PartitionPathUtils.generatePartitionPath(spec)); + + Path partitionPath = + FormatTablePartitionPathResolver.defaultPartitionPath( + new Path(table.location()), spec, true); + assertThat(partitionPath.getName()).isEqualTo("a%2Fb"); + assertThat(fileIO.exists(new Path(partitionPath, "_SUCCESS"))).isTrue(); + } + + @Test + void testCatalogManagedPartitionsAreLookedUpTogether() throws Exception { + String customLocation = tempDir.resolve("elsewhere").toUri().toString(); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + when(partitionManager.listPartitionsByNames( + Arrays.asList(spec("08"), spec("09"), spec("10")))) + .thenReturn( + Arrays.asList( + partition(spec("09"), customLocation), + partition(spec("08"), null))); + FormatTable table = table(Collections.emptyMap(), partitionManager); + + markDone( + table, + "dt=20240101/hh=08", + "dt=20240101/hh=09", + "dt=20240101/hh=10", + "dt=20240101/hh=08"); + + verify(partitionManager, times(1)).listPartitionsByNames(anyList()); + assertThat(fileIO.exists(path(table, "dt=20240101/hh=08/_SUCCESS"))).isTrue(); + // Custom location and unregistered partitions are skipped. + assertThat(fileIO.exists(new Path(customLocation, "_SUCCESS"))).isFalse(); + assertThat(fileIO.exists(path(table, "dt=20240101/hh=09"))).isFalse(); + assertThat(fileIO.exists(path(table, "dt=20240101/hh=10"))).isFalse(); + } + + @Test + void testInvalidArguments() { + FormatTable table = table(Collections.emptyMap(), null); + assertThatThrownBy(() -> markDone(table, "hh=08/dt=20240101")) + .hasMessageContaining("does not name the partition keys [dt, hh]"); + + FormatTable markEvent = + table( + Collections.singletonMap(PARTITION_MARK_DONE_ACTION.key(), "mark-event"), + null); + assertThatThrownBy(() -> createActions(markEvent)) + .hasMessageContaining("only supports the 'success-file' partition mark done"); + + FormatTable unpartitioned = + FormatTable.builder() + .fileIO(fileIO) + .identifier(Identifier.create("db", "events")) + .rowType(RowType.builder().field("id", DataTypes.INT()).build()) + .partitionKeys(Collections.emptyList()) + .location(tempDir.toUri().toString()) + .format(FormatTable.Format.CSV) + .options(Collections.emptyMap()) + .build(); + assertThatThrownBy(() -> createActions(unpartitioned)) + .hasMessageContaining("not partitioned"); + } + + private void markDone(FormatTable table, String... partitions) throws Exception { + for (PartitionMarkDoneAction action : createActions(table)) { + action.markDone(Arrays.asList(partitions)); + action.close(); + } + } + + private static List createActions(FormatTable table) { + return PartitionMarkDoneAction.createActions( + FormatTableSuccessFileMarkDoneActionTest.class.getClassLoader(), + table, + CoreOptions.fromMap(table.options())); + } + + private FormatTable table( + Map options, @Nullable FormatTablePartitionManager partitionManager) { + return FormatTable.builder() + .fileIO(fileIO) + .identifier(Identifier.create("db", "events")) + .rowType( + RowType.builder() + .field("id", DataTypes.INT()) + .field("dt", DataTypes.STRING()) + .field("hh", DataTypes.STRING()) + .build()) + .partitionKeys(Arrays.asList("dt", "hh")) + .location(tempDir.resolve("events").toUri().toString()) + .format(FormatTable.Format.CSV) + .options(new HashMap<>(options)) + .partitionManager(partitionManager) + .build(); + } + + private static Path path(FormatTable table, String relative) { + return new Path(table.location(), relative); + } + + private static Map spec(String hh) { + LinkedHashMap spec = new LinkedHashMap<>(); + spec.put("dt", "20240101"); + spec.put("hh", hh); + return spec; + } + + private static Partition partition(Map spec, @Nullable String location) { + return new Partition( + spec, + 0, + 0, + 0, + 0, + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS, + false, + null, + null, + null, + null, + location == null + ? null + : Collections.singletonMap(CoreOptions.PATH.key(), location)); + } +} diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java index 18a90c294cca..829862107c07 100644 --- a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java +++ b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java @@ -23,6 +23,7 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.partition.actions.PartitionMarkDoneAction; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.Table; import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.PartitionPathUtils; @@ -56,9 +57,31 @@ public String[] call( Identifier identifier = Identifier.fromString(tableId); Table table = catalog.getTable(identifier); + if (table instanceof FormatTable) { + FormatTable formatTable = (FormatTable) table; + List actions = + PartitionMarkDoneAction.createActions( + getClass().getClassLoader(), + formatTable, + CoreOptions.fromMap(formatTable.options())); + List partitionPaths = + PartitionPathUtils.generatePartitionPaths( + getPartitions(partitionStrings), formatTable.partitionType()); + try { + for (PartitionMarkDoneAction action : actions) { + action.markDone(partitionPaths); + } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + IOUtils.closeAllQuietly(actions); + } + return new String[] {"Success"}; + } checkArgument( table instanceof FileStoreTable, - "Only FileStoreTable supports mark_partition_done procedure. The table type is '%s'.", + "Only FileStoreTable and FormatTable support mark_partition_done procedure. " + + "The table type is '%s'.", table.getClass().getName()); FileStoreTable fileStoreTable = (FileStoreTable) table; diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MarkPartitionDoneAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MarkPartitionDoneAction.java index 4f5c73248d76..aedc9f44b9a2 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MarkPartitionDoneAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MarkPartitionDoneAction.java @@ -18,8 +18,10 @@ package org.apache.paimon.flink.action; +import org.apache.paimon.CoreOptions; import org.apache.paimon.partition.actions.PartitionMarkDoneAction; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FormatTable; import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.PartitionPathUtils; @@ -31,7 +33,6 @@ /** Table partition mark done action for Flink. */ public class MarkPartitionDoneAction extends TableActionBase implements LocalAction { - private final FileStoreTable fileStoreTable; private final List> partitions; public MarkPartitionDoneAction( @@ -40,29 +41,50 @@ public MarkPartitionDoneAction( List> partitions, Map catalogConfig) { super(databaseName, tableName, catalogConfig); - if (!(table instanceof FileStoreTable)) { + if (!(table instanceof FileStoreTable) && !(table instanceof FormatTable)) { throw new UnsupportedOperationException( String.format( - "Only FileStoreTable supports mark_partition_done action. The table type is '%s'.", + "Only FileStoreTable and FormatTable support mark_partition_done " + + "action. The table type is '%s'.", table.getClass().getName())); } - this.fileStoreTable = (FileStoreTable) table; this.partitions = partitions; } @Override public void executeLocally() throws Exception { + if (table instanceof FormatTable) { + FormatTable formatTable = (FormatTable) table; + List actions = + PartitionMarkDoneAction.createActions( + getClass().getClassLoader(), + formatTable, + CoreOptions.fromMap(formatTable.options())); + List partitionPaths = + PartitionPathUtils.generatePartitionPaths( + partitions, formatTable.partitionType()); + try { + for (PartitionMarkDoneAction action : actions) { + action.markDone(partitionPaths); + } + } finally { + IOUtils.closeAllQuietly(actions); + } + return; + } + + FileStoreTable fileStoreTable = (FileStoreTable) table; List actions = PartitionMarkDoneAction.createActions( getClass().getClassLoader(), fileStoreTable, fileStoreTable.coreOptions()); - - List partitionPaths = - PartitionPathUtils.generatePartitionPaths( - partitions, fileStoreTable.store().partitionType()); - - markDone(partitionPaths, actions); - - IOUtils.closeAllQuietly(actions); + try { + markDone( + PartitionPathUtils.generatePartitionPaths( + partitions, fileStoreTable.store().partitionType()), + actions); + } finally { + IOUtils.closeAllQuietly(actions); + } } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java index 5bee1a76e27c..69cd86dd9286 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MarkPartitionDoneProcedure.java @@ -23,6 +23,7 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.partition.actions.PartitionMarkDoneAction; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.Table; import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.PartitionPathUtils; @@ -64,9 +65,31 @@ public String[] call(ProcedureContext procedureContext, String tableId, String p Identifier identifier = Identifier.fromString(tableId); Table table = catalog.getTable(identifier); + if (table instanceof FormatTable) { + FormatTable formatTable = (FormatTable) table; + List actions = + PartitionMarkDoneAction.createActions( + procedureContext.getClass().getClassLoader(), + formatTable, + CoreOptions.fromMap(formatTable.options())); + List partitionPaths = + PartitionPathUtils.generatePartitionPaths( + getPartitions(partitionStrings), formatTable.partitionType()); + try { + for (PartitionMarkDoneAction action : actions) { + action.markDone(partitionPaths); + } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + IOUtils.closeAllQuietly(actions); + } + return new String[] {"Success"}; + } checkArgument( table instanceof FileStoreTable, - "Only FileStoreTable supports mark_partition_done procedure. The table type is '%s'.", + "Only FileStoreTable and FormatTable support mark_partition_done procedure. " + + "The table type is '%s'.", table.getClass().getName()); FileStoreTable fileStoreTable = (FileStoreTable) table; diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java index 8a75b86f2717..4f79d4507cae 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java @@ -319,6 +319,32 @@ public void testTruncatePartitionedTable() throws Exception { sql("DROP TABLE %s", tableName); } + @Test + public void testMarkPartitionDoneProcedure() throws Exception { + String tableName = "mark_partition_done"; + sql( + "CREATE TABLE %s (id INT, dt STRING, hh STRING) PARTITIONED BY (dt, hh) WITH " + + "('type'='format-table','file.format'='csv'," + + "'metastore.partitioned-table'='true')", + tableName); + setDataToken(tableName); + sql("INSERT INTO %s VALUES (1, '20240101', '08'), (2, '20240101', '09')", tableName); + FormatTable table = + (FormatTable) + flinkCatalog().catalog().getTable(Identifier.create("default", tableName)); + + sql( + "CALL sys.mark_partition_done(`table` => 'default.%s', " + + "partitions => 'dt=20240101,hh=08')", + tableName); + + assertThat(table.fileIO().exists(new Path(table.location(), "dt=20240101/hh=08/_SUCCESS"))) + .isTrue(); + assertThat(table.fileIO().exists(new Path(table.location(), "dt=20240101/hh=09/_SUCCESS"))) + .isFalse(); + assertThat(sql("SELECT id FROM %s WHERE hh = '08'", tableName)).containsExactly(Row.of(1)); + } + private void setDataToken(String tableName) { restCatalogServer.setDataToken( Identifier.create("default", tableName), diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MarkPartitionDoneProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MarkPartitionDoneProcedure.java index f5cc349b9b24..e47790a1ac51 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MarkPartitionDoneProcedure.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MarkPartitionDoneProcedure.java @@ -20,18 +20,24 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.partition.actions.PartitionMarkDoneAction; +import org.apache.paimon.spark.format.PaimonFormatTable; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FormatTable; import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.PartitionPathUtils; import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.Table; import org.apache.spark.sql.connector.catalog.TableCatalog; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; +import javax.annotation.Nullable; + import java.util.List; import static org.apache.paimon.utils.ParameterUtils.getPartitions; @@ -79,6 +85,28 @@ public InternalRow[] call(InternalRow args) { String partitionStr = args.getString(1); String[] partitions = partitionStr.split(";"); + FormatTable formatTable = loadFormatTable(tableIdent); + if (formatTable != null) { + List actions = + PartitionMarkDoneAction.createActions( + getClass().getClassLoader(), + formatTable, + CoreOptions.fromMap(formatTable.options())); + List partitionPaths = + PartitionPathUtils.generatePartitionPaths( + getPartitions(partitions), formatTable.partitionType()); + try { + for (PartitionMarkDoneAction action : actions) { + action.markDone(partitionPaths); + } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + IOUtils.closeAllQuietly(actions); + } + return new InternalRow[] {newInternalRow(true)}; + } + return modifyPaimonTable( tableIdent, table -> { @@ -106,6 +134,18 @@ public InternalRow[] call(InternalRow args) { }); } + /** Returns the Format Table of the identifier, or null if it is not one. */ + @Nullable + private FormatTable loadFormatTable(Identifier ident) { + try { + Table table = tableCatalog().loadTable(ident); + return table instanceof PaimonFormatTable ? ((PaimonFormatTable) table).table() : null; + } catch (NoSuchTableException e) { + // Let the regular path report the missing table. + return null; + } + } + public static void markDone(List partitions, List actions) { for (String partition : partitions) { try { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala index c7ab592a1b48..77ab3ede97af 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala @@ -18,16 +18,22 @@ package org.apache.paimon.spark.format +import org.apache.paimon.CoreOptions +import org.apache.paimon.partition.actions.PartitionMarkDoneAction import org.apache.paimon.spark.SparkInternalRowWrapper import org.apache.paimon.spark.write.{FormatTableWriteTaskResult, V2DataWrite, WriteTaskResult} import org.apache.paimon.table.FormatTable -import org.apache.paimon.table.sink.{BatchTableWrite, BatchWriteBuilder, CommitMessage} +import org.apache.paimon.table.format.FormatTableCommit +import org.apache.paimon.table.sink.{BatchTableCommit, BatchTableWrite, BatchWriteBuilder, CommitMessage} +import org.apache.paimon.utils.{IOUtils, PartitionPathUtils} import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.connector.write.{DataWriterFactory, WriterCommitMessage} import org.apache.spark.sql.types.StructType +import java.util.{LinkedHashMap => JLinkedHashMap} + import scala.collection.JavaConverters._ /** @@ -61,22 +67,80 @@ abstract class FormatTableBatchWriteBase( (_: Int, _: Long) => new FormatTableDataWriter(batchWriteBuilder, writeSchema) } + // Once committed, a later failure (e.g. marking partitions done) must not let abort delete data. + @volatile private var committed = false + protected def commitMessages(messages: Array[WriterCommitMessage]): Unit = { logInfo(s"Committing to FormatTable ${table.name()}") val batchTableCommit = batchWriteBuilder.newCommit() val commitMessages = WriteTaskResult.merge(messages).asJava + // Created before committing, so an unsupported action fails the write before it publishes data. + val markDoneActions = createMarkDoneActions() + try { + try { + val start = System.currentTimeMillis() + batchTableCommit.commit(commitMessages) + committed = true + logInfo(s"Committed in ${System.currentTimeMillis() - start} ms") + } catch { + case e: Exception => + logError("Failed to commit FormatTable writes", e) + throw e + } + markDone(markDoneActions, batchTableCommit) + } finally { + IOUtils.closeAllQuietly(markDoneActions.asJava) + } + } + + private def createMarkDoneActions(): Seq[PartitionMarkDoneAction] = { + val coreOptions = CoreOptions.fromMap(table.options()) + if ( + table.partitionKeys().isEmpty || + !coreOptions.toConfiguration.get(CoreOptions.PARTITION_MARK_DONE_WHEN_END_INPUT) + ) { + Seq.empty + } else { + PartitionMarkDoneAction + .createActions(getClass.getClassLoader, table, coreOptions) + .asScala + .toSeq + } + } + + private def markDone( + actions: Seq[PartitionMarkDoneAction], + batchTableCommit: BatchTableCommit): Unit = { + val partitions = batchTableCommit match { + case commit: FormatTableCommit => commit.committedPartitions().asScala + case _ => Seq.empty + } + if (actions.isEmpty || partitions.isEmpty) { + return + } + val partitionPaths = partitions + .map( + partition => + PartitionPathUtils.generatePartitionPath(new JLinkedHashMap[String, String](partition))) + .asJava try { - val start = System.currentTimeMillis() - batchTableCommit.commit(commitMessages) - logInfo(s"Committed in ${System.currentTimeMillis() - start} ms") + actions.foreach(_.markDone(partitionPaths)) } catch { case e: Exception => - logError("Failed to commit FormatTable writes", e) - throw e + throw new RuntimeException( + s"Committed data to FormatTable ${table.fullName()}, but failed to mark its partitions " + + "done. Call the mark_partition_done procedure to mark them.", + e + ) } } protected def abortMessages(messages: Array[WriterCommitMessage]): Unit = { + if (committed) { + logWarning( + s"Skip aborting the write to FormatTable ${table.name()}, its data is already committed") + return + } logInfo(s"Aborting write to FormatTable ${table.name()}") val batchTableCommit = batchWriteBuilder.newCommit() val commitMessages = WriteTaskResult.merge(messages).asJava diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionMarkDoneTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionMarkDoneTest.scala new file mode 100644 index 000000000000..4d1d15de8234 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionMarkDoneTest.scala @@ -0,0 +1,176 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.format + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.fs.{Path, PositionOutputStream} +import org.apache.paimon.fs.local.LocalFileIO +import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase +import org.apache.paimon.table.FormatTable + +import org.apache.spark.sql.Row +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.connector.write.{PhysicalWriteInfo, WriterCommitMessage} +import org.apache.spark.sql.paimon.shims.SparkShimLoader +import org.apache.spark.sql.types.StructType +import org.apache.spark.unsafe.types.UTF8String + +import java.io.IOException + +class FormatTablePartitionMarkDoneTest extends PaimonSparkTestWithRestCatalogBase { + + override protected def beforeEach(): Unit = { + super.beforeEach() + sql("USE paimon") + sql("CREATE DATABASE IF NOT EXISTS test_db") + sql("USE test_db") + } + + // Unique table names: a dropped table's directory can be reused by a table of the same name. + Seq(true, false).foreach { + managed => + test(s"FormatTable mark done: writes mark their partitions (managed=$managed)") { + val name = s"mark_done_write_$managed" + withTable(name) { + createTable(name, managed, "'partition.end-input-to-done'='true'") + sql(s"INSERT INTO $name VALUES (1, '20240101', '08'), (2, '20240101', '09')") + val table = formatTable(name) + assert(successFileLength(table, "dt=20240101/hh=08") == 0) + table.fileIO().overwriteFileUtf8(successPath(table, "dt=20240101/hh=09"), "stale") + + // Clears hh=08 without writing to it, and rewrites hh=09. + sql(s"INSERT OVERWRITE $name PARTITION (dt = '20240101') SELECT 3, '09'") + assert(!marked(table, "dt=20240101/hh=08")) + assert(successFileLength(table, "dt=20240101/hh=09") == 0) + + sql( + s"INSERT OVERWRITE $name PARTITION (dt = '20240102', hh = '00') SELECT id FROM $name WHERE id < 0") + assert(marked(table, "dt=20240102/hh=00")) + checkAnswer(sql(s"SELECT id, dt, hh FROM $name"), Seq(Row(3, "20240101", "09"))) + } + } + } + + test("FormatTable mark done: a failed mark keeps the committed data") { + val name = "mark_done_failure" + withTable(name) { + createTable(name, managed = false, "'partition.end-input-to-done'='true'") + val table = formatTable(name) + // Only creating _SUCCESS fails, which happens after the data is committed. + val fileIO = new LocalFileIO { + override def newOutputStream(path: Path, overwrite: Boolean): PositionOutputStream = { + if (path.getName == "_SUCCESS") { + throw new IOException("Injected failure") + } + super.newOutputStream(path, overwrite) + } + } + val failingTable = FormatTable + .builder() + .fileIO(fileIO) + .identifier(Identifier.create("test_db", name)) + .rowType(table.rowType()) + .partitionKeys(table.partitionKeys()) + .location(table.location()) + .format(table.format()) + .options(table.options()) + .catalogContext(table.catalogContext()) + .build() + val write = SparkShimLoader.shim.createFormatTableBatchWrite( + failingTable, + None, + StructType.fromDDL("id INT, dt STRING, hh STRING")) + val writer = write + .createBatchWriterFactory(new PhysicalWriteInfo { override def numPartitions(): Int = 1 }) + .createWriter(0, 0) + writer.write(InternalRow(1, UTF8String.fromString("20240101"), UTF8String.fromString("08"))) + val messages: Array[WriterCommitMessage] = Array(writer.commit()) + + val error = intercept[RuntimeException](write.commit(messages)) + assert(error.getMessage.contains("failed to mark its partitions done")) + // Spark aborts a write whose commit throws. + write.abort(messages) + checkAnswer(sql(s"SELECT id, dt, hh FROM $name"), Row(1, "20240101", "08") :: Nil) + assert(!marked(table, "dt=20240101/hh=08")) + } + } + + test("FormatTable mark done: the procedure marks partitions") { + val name = "mark_done_procedure" + withTable(name) { + createTable(name, managed = true, "'partition.mark-done-action'='success-file'") + sql(s"INSERT INTO $name VALUES (1, '20240101', '08'), (2, '20240101', '09')") + val table = formatTable(name) + assert(!marked(table, "dt=20240101/hh=08")) + + checkAnswer( + sql( + "CALL paimon.sys.mark_partition_done(" + + s"table => 'test_db.$name', partitions => 'dt=20240101,hh=08')"), + Row(true) :: Nil) + assert(marked(table, "dt=20240101/hh=08")) + assert(!marked(table, "dt=20240101/hh=09")) + } + } + + test("FormatTable mark done: an unsupported action fails before writing") { + val name = "mark_done_unsupported" + withTable(name) { + createTable( + name, + managed = false, + "'partition.end-input-to-done'='true', 'partition.mark-done-action'='mark-event'") + val expected = "only supports the 'success-file' partition mark done" + + val writeError = intercept[Exception](sql(s"INSERT INTO $name VALUES (1, '20240101', '08')")) + assert( + causes(writeError).exists(e => e.getMessage != null && e.getMessage.contains(expected))) + checkAnswer(sql(s"SELECT id FROM $name"), Nil) + + val procedureError = intercept[Exception] { + sql( + "CALL paimon.sys.mark_partition_done(" + + s"table => 'test_db.$name', partitions => 'dt=20240101,hh=08')") + } + assert(procedureError.getMessage.contains(expected)) + } + } + + private def createTable(name: String, managed: Boolean, properties: String): Unit = { + sql( + s"CREATE TABLE $name (id INT, dt STRING, hh STRING) USING csv PARTITIONED BY (dt, hh) " + + "TBLPROPERTIES ('format-table.implementation'='paimon', " + + s"'metastore.partitioned-table'='$managed', $properties)") + } + + private def formatTable(name: String): FormatTable = + paimonCatalog.getTable(Identifier.create("test_db", name)).asInstanceOf[FormatTable] + + private def successPath(table: FormatTable, partitionDirectory: String): Path = + new Path(new Path(table.location(), partitionDirectory), "_SUCCESS") + + private def marked(table: FormatTable, partitionDirectory: String): Boolean = + table.fileIO().exists(successPath(table, partitionDirectory)) + + private def successFileLength(table: FormatTable, partitionDirectory: String): Long = + table.fileIO().getFileStatus(successPath(table, partitionDirectory)).getLen + + private def causes(e: Throwable): Iterator[Throwable] = + Iterator.iterate(e)(_.getCause).takeWhile(_ != null) +}