You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "tsreaper (via GitHub)" <gi...@apache.org> on 2024/12/26 06:44:40 UTC
[PR] [flink] Add a action/procedure to remove unexisting files from manifests [paimon]
tsreaper opened a new pull request, #4781:
URL: https://github.com/apache/paimon/pull/4781
### Purpose
This PR adds an action/procedure to remove unexisting files from manifests. It has the following use cases:
* There is currently a known case when unexisting data file might be written into manifest. Consider a write-only job (W) plus a dedicated compaction job (C):
1. W commits a snapshot with file F. Then W constantly fails and restarts, each time before we can retry the commit (or W is stopped, creating Flink savepoint S).
2. C compacts F into a larger file, so F is now deleted from the manifest.
3. Before the compact snapshot expires, but after all snapshots created by W expires, W comes back to normal. As W cannot find its previous snapshot, it assumes that this snapshot has not been committed, so file F is committed to the manifest once again.
4. When the compact snapshot expires, file F will be deleted from the file system. Now F is in the manifest, but not on the file system. With this situation, user might want to remove F from the manifest to continue reading the table.
* User deletes a data file by mistake (for example, by incorrectly setting the time threshold
for orphan files cleaning). If the user can tolerate skipping some records when consuming
this table, he can also use this action to remove the file from manifest.
Note that user is on his own risk using this procedure, which may cause data loss when used outside from the use cases above.
### Tests
Unit tests and IT cases.
### API and Format
No format changes.
### Documentation
Document is also added.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Add a action/procedure to remove unexisting files from manifests [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #4781:
URL: https://github.com/apache/paimon/pull/4781
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Add a action/procedure to remove unexisting files from manifests [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #4781:
URL: https://github.com/apache/paimon/pull/4781#discussion_r1904013134
##########
paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.ThreadPoolUtils;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
+
+/** List what data files recorded in manifests are missing from the filesystem. */
+public class ListUnexistingFiles {
+
+ private final FileStoreTable table;
+ private final FileStorePathFactory pathFactory;
+ private final ThreadPoolExecutor executor;
+
+ public ListUnexistingFiles(FileStoreTable table) {
+ this.table = table;
+ this.pathFactory = table.store().pathFactory();
+ this.executor =
+ createCachedThreadPool(
+ table.coreOptions().deleteFileThreadNum(), "LIST_UNEXISTING_FILES");
+ }
+
+ public Map<Integer, Map<String, DataFileMeta>> list(BinaryRow partition) throws Exception {
+ Map<Integer, Map<String, DataFileMeta>> result = new ConcurrentHashMap<>();
+ List<Split> splits =
+ table.newScan()
+ .withPartitionFilter(Collections.singletonList(partition))
+ .plan()
+ .splits();
+ ThreadPoolUtils.randomlyOnlyExecute(
Review Comment:
return a List to collect files
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Add a action/procedure to remove unexisting files from manifests [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #4781:
URL: https://github.com/apache/paimon/pull/4781#discussion_r1897964021
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingFilesActionFactory.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.flink.action;
+
+import java.util.Optional;
+
+/** Factory to create {@link RemoveUnexistingFilesAction}. */
+public class RemoveUnexistingFilesActionFactory implements ActionFactory {
+
+ public static final String IDENTIFIER = "remove_unexisting_files";
+ private static final String DRY_RUN = "dry_run";
+ private static final String PARALLELISM = "parallelism";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterToolAdapter params) {
+ RemoveUnexistingFilesAction action =
+ new RemoveUnexistingFilesAction(
+ params.getRequired(DATABASE),
+ params.getRequired(TABLE),
+ catalogConfigMap(params));
+
+ if (params.has(PARTITION)) {
+ action.withPartitions(getPartitions(params));
+ }
+
+ if (params.has(DRY_RUN) && Boolean.parseBoolean(params.get(DRY_RUN))) {
+ action.dryRun();
+ }
+
+ if (params.has(PARALLELISM)) {
+ action.withParallelism(Integer.parseInt(params.get(PARALLELISM)));
+ }
+
+ return Optional.of(action);
+ }
+
+ @Override
+ public void printHelp() {}
Review Comment:
Please add the usage of this action.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingFilesAction.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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.flink.action;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.operation.ListUnexistingFiles;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/**
+ * Action to remove unexisting data files from manifest entries. It has the following use cases:
+ *
+ * <ul>
+ * <li>There is currently a known case when unexisting data file might be written into manifest.
+ * Consider a write-only job (W) plus a dedicated compaction job (C):
+ * <ol>
+ * <li>W commits a snapshot with file F. Then W constantly fails and restarts, each time
+ * before we can retry the commit (or W is stopped, creating Flink savepoint S).
+ * <li>C compacts F into a larger file, so F is now deleted from the manifest.
+ * <li>Before the compact snapshot expires, but after all snapshots created by W expires, W
+ * comes back to normal (or restarted from savepoint S). As W cannot find its previous
+ * snapshot, it assumes that this snapshot has not been committed (see {@link
+ * org.apache.paimon.operation.FileStoreCommitImpl#filterCommitted} for more detail), so
+ * file F is committed to the manifest once again.
+ * <li>When the compact snapshot expires, file F will be deleted from the file system. Now F
+ * is in the manifest, but not on the file system. With this situation, user might want
+ * to remove F from the manifest to continue reading the table.
+ * </ol>
+ * <li>User deletes a data file by mistake (for example, by incorrectly setting the time threshold
+ * for orphan files cleaning). If the user can tolerate skipping some records when consuming
+ * this table, he can also use this action to remove the file from manifest.
+ * </ul>
+ *
+ * <p>Note that user is on his own risk using this procedure, which may cause data loss when used
+ * outside from the use cases above.
+ */
+public class RemoveUnexistingFilesAction extends TableActionBase {
+
+ private static final OutputTag<String> RESULT_SIDE_OUTPUT =
+ new OutputTag<>("result-side-output", BasicTypeInfo.STRING_TYPE_INFO);
+
+ @Nullable private List<Map<String, String>> partitions = null;
+ private boolean dryRun = false;
+ @Nullable private Integer parallelism = null;
+
+ public RemoveUnexistingFilesAction(
+ String databaseName, String tableName, Map<String, String> catalogConfig) {
+ super(databaseName, tableName, catalogConfig);
+ }
+
+ public RemoveUnexistingFilesAction withPartitions(List<Map<String, String>> partitions) {
+ this.partitions = partitions;
+ return this;
+ }
+
+ public RemoveUnexistingFilesAction dryRun() {
+ this.dryRun = true;
+ return this;
+ }
+
+ public RemoveUnexistingFilesAction withParallelism(int parallelism) {
+ this.parallelism = parallelism;
+ return this;
+ }
+
+ @Override
+ public void build() throws Exception {
+ buildDataStream();
+ }
+
+ public DataStream<String> buildDataStream() throws Exception {
Review Comment:
This method will not throw any Exception. So we can remove "throws Exception".
##########
paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** List what data files recorded in manifests are missing from the filesystem. */
+public class ListUnexistingFiles {
+
+ private final FileStoreTable table;
+
+ public ListUnexistingFiles(FileStoreTable table) {
+ this.table = table;
+ }
+
+ public Map<Integer, Map<String, DataFileMeta>> list(BinaryRow partition) throws Exception {
+ FileIO fileIO = table.fileIO();
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+ Map<Integer, Map<String, DataFileMeta>> result = new HashMap<>();
Review Comment:
Maybe it is better use UnexistingFiles replace result.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingFilesAction.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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.flink.action;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.operation.ListUnexistingFiles;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/**
+ * Action to remove unexisting data files from manifest entries. It has the following use cases:
+ *
+ * <ul>
+ * <li>There is currently a known case when unexisting data file might be written into manifest.
+ * Consider a write-only job (W) plus a dedicated compaction job (C):
+ * <ol>
+ * <li>W commits a snapshot with file F. Then W constantly fails and restarts, each time
+ * before we can retry the commit (or W is stopped, creating Flink savepoint S).
+ * <li>C compacts F into a larger file, so F is now deleted from the manifest.
+ * <li>Before the compact snapshot expires, but after all snapshots created by W expires, W
+ * comes back to normal (or restarted from savepoint S). As W cannot find its previous
+ * snapshot, it assumes that this snapshot has not been committed (see {@link
+ * org.apache.paimon.operation.FileStoreCommitImpl#filterCommitted} for more detail), so
+ * file F is committed to the manifest once again.
+ * <li>When the compact snapshot expires, file F will be deleted from the file system. Now F
+ * is in the manifest, but not on the file system. With this situation, user might want
+ * to remove F from the manifest to continue reading the table.
+ * </ol>
+ * <li>User deletes a data file by mistake (for example, by incorrectly setting the time threshold
+ * for orphan files cleaning). If the user can tolerate skipping some records when consuming
+ * this table, he can also use this action to remove the file from manifest.
+ * </ul>
+ *
+ * <p>Note that user is on his own risk using this procedure, which may cause data loss when used
+ * outside from the use cases above.
+ */
+public class RemoveUnexistingFilesAction extends TableActionBase {
+
+ private static final OutputTag<String> RESULT_SIDE_OUTPUT =
+ new OutputTag<>("result-side-output", BasicTypeInfo.STRING_TYPE_INFO);
+
+ @Nullable private List<Map<String, String>> partitions = null;
+ private boolean dryRun = false;
+ @Nullable private Integer parallelism = null;
+
+ public RemoveUnexistingFilesAction(
+ String databaseName, String tableName, Map<String, String> catalogConfig) {
+ super(databaseName, tableName, catalogConfig);
+ }
+
+ public RemoveUnexistingFilesAction withPartitions(List<Map<String, String>> partitions) {
+ this.partitions = partitions;
+ return this;
+ }
+
+ public RemoveUnexistingFilesAction dryRun() {
+ this.dryRun = true;
+ return this;
+ }
+
+ public RemoveUnexistingFilesAction withParallelism(int parallelism) {
+ this.parallelism = parallelism;
+ return this;
+ }
+
+ @Override
+ public void build() throws Exception {
Review Comment:
Check ExecutionOptions.RUNTIME_MODE == RuntimeExecutionMode.BATCH ?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingFilesAction.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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.flink.action;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.operation.ListUnexistingFiles;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/**
+ * Action to remove unexisting data files from manifest entries. It has the following use cases:
+ *
+ * <ul>
+ * <li>There is currently a known case when unexisting data file might be written into manifest.
+ * Consider a write-only job (W) plus a dedicated compaction job (C):
+ * <ol>
+ * <li>W commits a snapshot with file F. Then W constantly fails and restarts, each time
+ * before we can retry the commit (or W is stopped, creating Flink savepoint S).
+ * <li>C compacts F into a larger file, so F is now deleted from the manifest.
+ * <li>Before the compact snapshot expires, but after all snapshots created by W expires, W
+ * comes back to normal (or restarted from savepoint S). As W cannot find its previous
+ * snapshot, it assumes that this snapshot has not been committed (see {@link
+ * org.apache.paimon.operation.FileStoreCommitImpl#filterCommitted} for more detail), so
+ * file F is committed to the manifest once again.
+ * <li>When the compact snapshot expires, file F will be deleted from the file system. Now F
+ * is in the manifest, but not on the file system. With this situation, user might want
+ * to remove F from the manifest to continue reading the table.
+ * </ol>
+ * <li>User deletes a data file by mistake (for example, by incorrectly setting the time threshold
+ * for orphan files cleaning). If the user can tolerate skipping some records when consuming
+ * this table, he can also use this action to remove the file from manifest.
+ * </ul>
+ *
+ * <p>Note that user is on his own risk using this procedure, which may cause data loss when used
+ * outside from the use cases above.
+ */
+public class RemoveUnexistingFilesAction extends TableActionBase {
+
+ private static final OutputTag<String> RESULT_SIDE_OUTPUT =
+ new OutputTag<>("result-side-output", BasicTypeInfo.STRING_TYPE_INFO);
+
+ @Nullable private List<Map<String, String>> partitions = null;
+ private boolean dryRun = false;
+ @Nullable private Integer parallelism = null;
+
+ public RemoveUnexistingFilesAction(
+ String databaseName, String tableName, Map<String, String> catalogConfig) {
+ super(databaseName, tableName, catalogConfig);
+ }
+
+ public RemoveUnexistingFilesAction withPartitions(List<Map<String, String>> partitions) {
+ this.partitions = partitions;
+ return this;
+ }
+
+ public RemoveUnexistingFilesAction dryRun() {
+ this.dryRun = true;
+ return this;
+ }
+
+ public RemoveUnexistingFilesAction withParallelism(int parallelism) {
+ this.parallelism = parallelism;
+ return this;
+ }
+
+ @Override
+ public void build() throws Exception {
+ buildDataStream();
+ }
+
+ public DataStream<String> buildDataStream() throws Exception {
Review Comment:
This method will not throw any Exception. So we can remove "throws Exception".
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Add a action/procedure to remove unexisting files from manifests [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #4781:
URL: https://github.com/apache/paimon/pull/4781#discussion_r1898909997
##########
paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** List what data files recorded in manifests are missing from the filesystem. */
+public class ListUnexistingFiles {
+
+ private final FileStoreTable table;
+
+ public ListUnexistingFiles(FileStoreTable table) {
+ this.table = table;
+ }
+
+ public Map<Integer, Map<String, DataFileMeta>> list(BinaryRow partition) throws Exception {
+ FileIO fileIO = table.fileIO();
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+ Map<Integer, Map<String, DataFileMeta>> result = new HashMap<>();
+ List<Split> splits =
+ table.newScan()
+ .withPartitionFilter(Collections.singletonList(partition))
+ .plan()
+ .splits();
+ for (Split split : splits) {
+ DataSplit dataSplit = (DataSplit) split;
+ DataFilePathFactory dataFilePathFactory =
+ pathFactory.createDataFilePathFactory(partition, dataSplit.bucket());
+ for (DataFileMeta meta : dataSplit.dataFiles()) {
+ Path path = dataFilePathFactory.toPath(meta);
+ if (!fileIO.exists(path)) {
Review Comment:
It is better to use a thread pool, this invoking is very slow for object store.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Add a action/procedure to remove unexisting files from manifests [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #4781:
URL: https://github.com/apache/paimon/pull/4781#discussion_r1900703929
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingFilesAction.java:
##########
@@ -0,0 +1,449 @@
+/*
+ * 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.flink.action;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.flink.utils.BoundedTwoInputOperator;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.InputSelection;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/**
+ * Action to remove unexisting data files from manifest entries. It has the following use cases:
+ *
+ * <ul>
+ * <li>There is currently a known case when unexisting data file might be written into manifest.
+ * Consider a write-only job (W) plus a dedicated compaction job (C):
+ * <ol>
+ * <li>W commits a snapshot with file F. Then W constantly fails and restarts, each time
+ * before we can retry the commit (or W is stopped, creating Flink savepoint S).
+ * <li>C compacts F into a larger file, so F is now deleted from the manifest.
+ * <li>Before the compact snapshot expires, but after all snapshots created by W expires, W
+ * comes back to normal (or restarted from savepoint S). As W cannot find its previous
+ * snapshot, it assumes that this snapshot has not been committed (see {@link
+ * org.apache.paimon.operation.FileStoreCommitImpl#filterCommitted} for more detail), so
+ * file F is committed to the manifest once again.
+ * <li>When the compact snapshot expires, file F will be deleted from the file system. Now F
+ * is in the manifest, but not on the file system. With this situation, user might want
+ * to remove F from the manifest to continue reading the table.
+ * </ol>
+ * <li>User deletes a data file by mistake (for example, by incorrectly setting the time threshold
+ * for orphan files cleaning). If the user can tolerate skipping some records when consuming
+ * this table, he can also use this action to remove the file from manifest.
+ * </ul>
+ *
+ * <p>Note that user is on his own risk using this procedure, which may cause data loss when used
+ * outside from the use cases above.
+ */
+public class RemoveUnexistingFilesAction extends TableActionBase {
+
+ private static final OutputTag<String> RESULT_SIDE_OUTPUT =
+ new OutputTag<>("result-side-output", BasicTypeInfo.STRING_TYPE_INFO);
+
+ private boolean dryRun = false;
+ @Nullable private Integer parallelism = null;
+
+ public RemoveUnexistingFilesAction(
+ String databaseName, String tableName, Map<String, String> catalogConfig) {
+ super(databaseName, tableName, catalogConfig);
+ }
+
+ public RemoveUnexistingFilesAction dryRun() {
+ this.dryRun = true;
+ return this;
+ }
+
+ public RemoveUnexistingFilesAction withParallelism(int parallelism) {
+ this.parallelism = parallelism;
+ return this;
+ }
+
+ @Override
+ public void build() throws Exception {
+ buildDataStream();
+ }
+
+ public DataStream<String> buildDataStream() throws Exception {
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ flinkConf.set(ExecutionOptions.SORT_INPUTS, false);
+ flinkConf.set(ExecutionOptions.USE_BATCH_STATE_BACKEND, false);
+ if (parallelism != null) {
+ flinkConf.set(CoreOptions.DEFAULT_PARALLELISM, parallelism);
+ }
+ // Flink 1.17 introduced this config, use string to keep compatibility
+ flinkConf.setString("execution.batch.adaptive.auto-parallelism.enabled", "false");
+ env.configure(flinkConf);
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ SingleOutputStreamOperator<byte[]> source =
+ env.fromData(
+ ((FileStoreTable) table)
+ .newScan().listPartitions().stream()
+ .map(BinaryRow::toBytes)
+ .collect(Collectors.toList()),
+ PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO)
+ .name("List Partitions")
+ .forceNonParallel();
+
+ SingleOutputStreamOperator<String> files =
+ source.transform(
+ "List Buckets",
+ BasicTypeInfo.STRING_TYPE_INFO,
+ new ListBucketsInPartitionOperator(fileStoreTable))
+ .transform(
+ "List Files",
+ BasicTypeInfo.STRING_TYPE_INFO,
+ new ListFilesInBucketOperator(fileStoreTable.fileIO()));
+ SingleOutputStreamOperator<Tuple4<byte[], Integer, byte[], String>> splits =
+ source.transform(
+ "List Data Splits",
+ new TupleTypeInfo<>(
+ PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO,
+ BasicTypeInfo.INT_TYPE_INFO,
+ PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO),
+ new ScanManifestOperator(fileStoreTable));
+
+ SingleOutputStreamOperator<Committable> comparator =
+ files.keyBy(f -> f)
+ .connect(splits.keyBy(t -> new Path(t.f3).getName()))
+ .transform(
+ "Compare Files",
+ new CommittableTypeInfo(),
+ new CompareOperator(fileStoreTable));
+ DataStream<String> result = comparator.getSideOutput(RESULT_SIDE_OUTPUT);
+ if (dryRun) {
+ return result;
+ }
+
+ comparator
+ .transform(
+ "Global Committer : " + table.name(),
+ new CommittableTypeInfo(),
+ new CommitOperator(fileStoreTable))
+ .forceNonParallel();
+ return result;
+ }
+
+ @Override
+ public void run() throws Exception {
+ build();
+ env.execute("Remove Unexisting Files : " + table.name());
+ }
+
+ private static class ListBucketsInPartitionOperator
+ extends BoundedOneInputOperator<byte[], String> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FileStoreTable table;
+
+ private transient BinaryRow reuse;
+ private transient FileStorePathFactory pathFactory;
+
+ private ListBucketsInPartitionOperator(FileStoreTable table) {
+ this.table = table;
+ }
+
+ @Override
+ public void open() throws Exception {
+ reuse = new BinaryRow(table.schema().partitionKeys().size());
+ pathFactory = table.store().pathFactory();
+ }
+
+ @Override
+ public void processElement(StreamRecord<byte[]> record) throws Exception {
+ byte[] bytes = record.getValue();
+ reuse.pointTo(MemorySegment.wrap(bytes), 0, bytes.length);
+ Path partitionPath = pathFactory.getPartitionPath(reuse);
Review Comment:
We need to consider external path for table...
I think it is better to still use fileio.exists method...
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Add a action/procedure to remove unexisting files from manifests [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #4781:
URL: https://github.com/apache/paimon/pull/4781#discussion_r1904013134
##########
paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.ThreadPoolUtils;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
+
+/** List what data files recorded in manifests are missing from the filesystem. */
+public class ListUnexistingFiles {
+
+ private final FileStoreTable table;
+ private final FileStorePathFactory pathFactory;
+ private final ThreadPoolExecutor executor;
+
+ public ListUnexistingFiles(FileStoreTable table) {
+ this.table = table;
+ this.pathFactory = table.store().pathFactory();
+ this.executor =
+ createCachedThreadPool(
+ table.coreOptions().deleteFileThreadNum(), "LIST_UNEXISTING_FILES");
+ }
+
+ public Map<Integer, Map<String, DataFileMeta>> list(BinaryRow partition) throws Exception {
+ Map<Integer, Map<String, DataFileMeta>> result = new ConcurrentHashMap<>();
+ List<Split> splits =
+ table.newScan()
+ .withPartitionFilter(Collections.singletonList(partition))
+ .plan()
+ .splits();
+ ThreadPoolUtils.randomlyOnlyExecute(
Review Comment:
return a List to collect files, otherwise, using `HashMap` has thread safe problem.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org