diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html index 003d5ba4c2cb..7fdb43b8b0ea 100644 --- a/docs/layouts/shortcodes/generated/core_configuration.html +++ b/docs/layouts/shortcodes/generated/core_configuration.html @@ -218,6 +218,24 @@ Duration The TTL in rocksdb index for cross partition upsert (primary keys not contain all partition fields), this can avoid maintaining too many indexes and lead to worse and worse performance, but please note that this may also cause data duplication. + +
data-file.external-paths
+ (none) + String + The external paths where the data of this table will be written, multiple elements separated by commas. + + +
data-file.external-paths.specific-fs
+ (none) + String + The specific file system of the external path when data-file.external-paths.strategy is set to specific-fs, should be the prefix scheme of the external path, now supported are s3 and oss. + + +
data-file.external-paths.strategy
+ none +

Enum

+ The strategy of selecting an external path when writing data.

Possible values: +
data-file.path-directory
(none) @@ -930,6 +948,12 @@ String A list of commit callback classes to be called after a successful tag. Class names are connected with comma (example: com.test.CallbackA,com.sample.CallbackB). + +
tag.create-success-file
+ false + Boolean + Whether to create tag success file for new created tags. +
tag.creation-delay
0 ms @@ -1026,23 +1050,5 @@ Integer The bytes of types (CHAR, VARCHAR, BINARY, VARBINARY) devote to the zorder sort. - -
data-file.external-paths
- (none) - String - The external paths where the data of this table will be written, multiple elements separated by commas. - - -
data-file.external-paths.strategy
- none -

Enum

- The strategy of selecting an external path when writing data.

Possible values: - - -
data-file.external-paths.specific-fs
- (none) - String - The specific file system of the external path when data-file.external-paths.strategy is set to specific-fs, should be the prefix scheme of the external path, now supported are s3 and oss. - diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java index 0beea0a8f2de..3d6b0b5490fa 100644 --- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java @@ -1267,6 +1267,12 @@ public class CoreOptions implements Serializable { .withDescription( "Whether to create tag automatically. And how to generate tags."); + public static final ConfigOption TAG_CREATE_SUCCESS_FILE = + key("tag.create-success-file") + .booleanType() + .defaultValue(false) + .withDescription("Whether to create tag success file for new created tags."); + public static final ConfigOption TAG_CREATION_PERIOD = key("tag.creation-period") .enumType(TagCreationPeriod.class) @@ -2285,6 +2291,10 @@ public String tagToPartitionField() { return options.get(METASTORE_TAG_TO_PARTITION); } + public boolean tagCreateSuccessFile() { + return options.get(TAG_CREATE_SUCCESS_FILE); + } + public TagCreationMode tagToPartitionPreview() { return options.get(METASTORE_TAG_TO_PARTITION_PREVIEW); } diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index 8b24ec2a54c6..953096c0b50c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -48,6 +48,7 @@ import org.apache.paimon.table.sink.CallbackUtils; import org.apache.paimon.table.sink.CommitCallback; import org.apache.paimon.table.sink.TagCallback; +import org.apache.paimon.tag.SuccessFileTagCallback; import org.apache.paimon.tag.TagAutoManager; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.FileStorePathFactory; @@ -378,6 +379,9 @@ public List createTagCallbacks() { callbacks.add( new AddPartitionTagCallback(metastoreClientFactory.create(), partitionField)); } + if (options.tagCreateSuccessFile()) { + callbacks.add(new SuccessFileTagCallback(fileIO, newTagManager().tagDirectory())); + } return callbacks; } diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/SuccessFileTagCallback.java b/paimon-core/src/main/java/org/apache/paimon/tag/SuccessFileTagCallback.java new file mode 100644 index 000000000000..d1daa48ce8a5 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/tag/SuccessFileTagCallback.java @@ -0,0 +1,78 @@ +/* + * 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.tag; + +import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.partition.file.SuccessFile; +import org.apache.paimon.table.sink.TagCallback; + +import java.io.IOException; + +/** A {@link TagCallback} which create "{tagName}_SUCCESS" file. */ +public class SuccessFileTagCallback implements TagCallback { + private static final String SUCCESS_FILE_SUFFIX = "_SUCCESS"; + private static final String SUCCESS_FILE_DIRECTORY = "tag-success-file"; + + private final FileIO fileIO; + private final Path successFileDirectory; + + public SuccessFileTagCallback(FileIO fileIO, Path tagDirectory) { + this.fileIO = fileIO; + this.successFileDirectory = new Path(tagDirectory, SUCCESS_FILE_DIRECTORY); + try { + fileIO.checkOrMkdirs(successFileDirectory); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + @Override + public void notifyCreation(String tagName) { + Path tagSuccessPath = tagSuccessFilePath(tagName); + long currentTime = System.currentTimeMillis(); + SuccessFile successFile = new SuccessFile(currentTime, currentTime); + + try { + if (fileIO.exists(tagSuccessPath)) { + successFile = + SuccessFile.fromPath(fileIO, tagSuccessPath) + .updateModificationTime(currentTime); + } + fileIO.overwriteFileUtf8(tagSuccessPath, successFile.toJson()); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + @Override + public void notifyDeletion(String tagName) { + Path tagSuccessPath = tagSuccessFilePath(tagName); + fileIO.deleteQuietly(tagSuccessPath); + } + + @VisibleForTesting + public Path tagSuccessFilePath(String tagName) { + return new Path(successFileDirectory, tagName + SUCCESS_FILE_SUFFIX); + } + + @Override + public void close() throws Exception {} +} diff --git a/paimon-core/src/test/java/org/apache/paimon/tag/SuccessFileTagCallBackTest.java b/paimon-core/src/test/java/org/apache/paimon/tag/SuccessFileTagCallBackTest.java new file mode 100644 index 000000000000..34dd1188186b --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/tag/SuccessFileTagCallBackTest.java @@ -0,0 +1,105 @@ +/* + * 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.tag; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.FileStore; +import org.apache.paimon.catalog.PrimaryKeyTableTestBase; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.TableCommitImpl; +import org.apache.paimon.table.sink.TagCallback; +import org.apache.paimon.utils.TagManager; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.List; + +import static org.apache.paimon.CoreOptions.TAG_AUTOMATIC_CREATION; +import static org.apache.paimon.CoreOptions.TAG_CREATION_PERIOD; +import static org.apache.paimon.CoreOptions.TAG_NUM_RETAINED_MAX; +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link SuccessFileTagCallback}. */ +public class SuccessFileTagCallBackTest extends PrimaryKeyTableTestBase { + + @Test + public void testWithSuccessFile() throws IOException { + Options options = new Options(); + options.set(CoreOptions.TAG_CREATE_SUCCESS_FILE, true); + options.set(TAG_AUTOMATIC_CREATION, CoreOptions.TagCreationMode.WATERMARK); + options.set(TAG_CREATION_PERIOD, CoreOptions.TagCreationPeriod.HOURLY); + options.set(TAG_NUM_RETAINED_MAX, 3); + FileStoreTable table = this.table.copy(options.toMap()); + FileIO fileIO = table.fileIO(); + TableCommitImpl commit = table.newCommit(commitUser).ignoreEmptyCommit(false); + FileStore fileStore = table.store(); + TagManager tagManager = fileStore.newTagManager(); + SuccessFileTagCallback successFileTagCallback = null; + + List tagCallbacks = fileStore.createTagCallbacks(); + assertThat(tagCallbacks).hasAtLeastOneElementOfType(SuccessFileTagCallback.class); + + for (TagCallback tagCallback : tagCallbacks) { + if (tagCallback instanceof SuccessFileTagCallback) { + successFileTagCallback = (SuccessFileTagCallback) tagCallback; + } + } + + commit.commit(new ManifestCommittable(0, utcMills("2023-07-18T12:12:00"))); + assertThat(tagManager.allTagNames()).containsOnly("2023-07-18 11"); + assertThat(fileIO.exists(successFileTagCallback.tagSuccessFilePath("2023-07-18 11"))) + .isTrue(); + + table.deleteTag("2023-07-18 11"); + assertThat(tagManager.get("2023-07-18 11")).isEmpty(); + assertThat(fileIO.exists(successFileTagCallback.tagSuccessFilePath("2023-07-18 11"))) + .isFalse(); + } + + @Test + public void testWithoutSuccessFile() throws IOException { + Options options = new Options(); + options.set(CoreOptions.TAG_CREATE_SUCCESS_FILE, false); + options.set(TAG_AUTOMATIC_CREATION, CoreOptions.TagCreationMode.WATERMARK); + options.set(TAG_CREATION_PERIOD, CoreOptions.TagCreationPeriod.HOURLY); + options.set(TAG_NUM_RETAINED_MAX, 3); + FileStoreTable table = this.table.copy(options.toMap()); + FileIO fileIO = table.fileIO(); + TableCommitImpl commit = table.newCommit(commitUser).ignoreEmptyCommit(false); + FileStore fileStore = table.store(); + TagManager tagManager = fileStore.newTagManager(); + List tagCallbacks = fileStore.createTagCallbacks(); + assertThat(tagCallbacks).doesNotHaveAnyElementsOfTypes(SuccessFileTagCallback.class); + + commit.commit(new ManifestCommittable(0, utcMills("2023-07-18T12:12:00"))); + assertThat(tagManager.allTagNames()).containsOnly("2023-07-18 11"); + assertThat(fileIO.exists(new Path(tagManager.tagPath("2023-07-18 11"), "tag-success-file"))) + .isFalse(); + + table.deleteTag("2023-07-18 11"); + assertThat(tagManager.get("2023-07-18 11")).isEmpty(); + assertThat(fileIO.exists(new Path(tagManager.tagPath("2023-07-18 11"), "tag-success-file"))) + .isFalse(); + } +}