You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "wwj6591812 (via GitHub)" <gi...@apache.org> on 2024/04/30 01:09:58 UTC
[PR] support clone latest snapshot [paimon]
wwj6591812 opened a new pull request, #3287:
URL: https://github.com/apache/paimon/pull/3287
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: [close #xxx](https://github.com/apache/paimon/issues/3229)
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598405595
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.clone.CloneFileInfo;
+import org.apache.paimon.clone.SnapshotHintChannelComputer;
+import org.apache.paimon.flink.clone.CloneFileInfoTypeInfo;
+import org.apache.paimon.flink.clone.CloneSourceBuilder;
+import org.apache.paimon.flink.clone.CopyFileOperator;
+import org.apache.paimon.flink.clone.PickFilesForCloneOperator;
+import org.apache.paimon.flink.clone.SnapshotHintOperator;
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.options.CatalogOptions;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/** Snapshot/Tag/Table clone action for Flink. */
+public class CloneAction extends ActionBase {
+
+ private final int parallelism;
+
+ private Map<String, String> sourceCatalogConfig = new HashMap<>();
+ private final String database;
+ private final String tableName;
+
+ private Map<String, String> targetCatalogConfig = new HashMap<>();
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneAction(
+ String warehouse,
+ String database,
+ String tableName,
+ Map<String, String> sourceCatalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTableName,
+ Map<String, String> targetCatalogConfig,
+ String parallelismStr) {
+ super(warehouse, sourceCatalogConfig);
+
+ checkNotNull(warehouse, "warehouse must not be null.");
+ checkNotNull(targetWarehouse, "targetWarehouse must not be null.");
+ checkNotNull(targetDatabase, "targetDatabase must not be null.");
+ checkNotNull(targetTableName, "targetTableName must not be null.");
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFileInfoTypeInfo.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.flink.clone;
+
+import org.apache.paimon.clone.CloneFileInfo;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+
+/** Type information of {@link CloneFileInfo}. */
+public class CloneFileInfoTypeInfo extends TypeInformation<CloneFileInfo> {
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598409315
##########
paimon-core/src/main/java/org/apache/paimon/clone/CopyFileUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.IOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** Utility class for copy file. */
+public class CopyFileUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CopyFileUtils.class);
+
+ public static void copyFile(
+ CloneFileInfo cloneFileInfo,
+ FileIO sourceTableFileIO,
+ FileIO targetTableFileIO,
+ Path sourceTableRootPath,
+ Path targetTableRootPath)
+ throws IOException {
+ Path filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
+ Path sourcePath = new Path(sourceTableRootPath.toString() + filePathExcludeTableRoot);
+ Path targetPath = new Path(targetTableRootPath.toString() + filePathExcludeTableRoot);
+
+ if (targetTableFileIO.exists(targetPath)
+ && targetTableFileIO.getFileSize(targetPath)
+ == sourceTableFileIO.getFileSize(sourcePath)) {
+ LOG.info(
+ "Skipping target file {} because it already exists and has the same size.",
+ targetPath);
+ return;
+ }
+
+ LOG.debug("Begin copy file from {} to {}.", sourcePath, targetPath);
Review Comment:
delete this class
##########
paimon-core/src/main/java/org/apache/paimon/clone/CopyFileUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.IOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** Utility class for copy file. */
+public class CopyFileUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CopyFileUtils.class);
+
+ public static void copyFile(
+ CloneFileInfo cloneFileInfo,
+ FileIO sourceTableFileIO,
+ FileIO targetTableFileIO,
+ Path sourceTableRootPath,
+ Path targetTableRootPath)
+ throws IOException {
+ Path filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
+ Path sourcePath = new Path(sourceTableRootPath.toString() + filePathExcludeTableRoot);
+ Path targetPath = new Path(targetTableRootPath.toString() + filePathExcludeTableRoot);
+
+ if (targetTableFileIO.exists(targetPath)
+ && targetTableFileIO.getFileSize(targetPath)
+ == sourceTableFileIO.getFileSize(sourcePath)) {
+ LOG.info(
+ "Skipping target file {} because it already exists and has the same size.",
+ targetPath);
+ return;
+ }
+
+ LOG.debug("Begin copy file from {} to {}.", sourcePath, targetPath);
+ IOUtils.copyBytes(
+ sourceTableFileIO.newInputStream(sourcePath),
+ targetTableFileIO.newOutputStream(targetPath, true));
+ LOG.debug("End copy file from {} to {}.", sourcePath, targetPath);
Review Comment:
delete this class
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598406881
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** CloneSourceBuilder. */
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.PickFilesUtil;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** PickFilesForCloneOperator. */
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605804121
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/SnapshotHintOperator.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+
+/**
+ * When the files copy finished of a table, then create snapshot hint, it means that this table can
+ * be used now.
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CopyFileOperator.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.AbstractCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.IOUtils;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.Map;
+
+/** A Operator to copy files. */
+public class CopyFileOperator extends AbstractStreamOperator<CloneFileInfo>
+ implements OneInputStreamOperator<CloneFileInfo, CloneFileInfo> {
+
+ private final Map<String, String> sourceCatalogConfig;
+ private final Map<String, String> targetCatalogConfig;
+
+ private AbstractCatalog sourceCatalog;
+ private AbstractCatalog targetCatalog;
+
+ public CopyFileOperator(
+ Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
+ super();
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sourceCatalog =
+ (AbstractCatalog)
+ FlinkCatalogFactory.createPaimonCatalog(
+ Options.fromMap(sourceCatalogConfig));
+ targetCatalog =
+ (AbstractCatalog)
+ FlinkCatalogFactory.createPaimonCatalog(
+ Options.fromMap(targetCatalogConfig));
+ }
+
+ @Override
+ public void processElement(StreamRecord<CloneFileInfo> streamRecord) throws Exception {
+ CloneFileInfo cloneFileInfo = streamRecord.getValue();
+
+ FileIO sourceTableFileIO = sourceCatalog.fileIO();
+ FileIO targetTableFileIO = targetCatalog.fileIO();
+ Path sourceTableRootPath =
+ sourceCatalog.getDataTableLocation(
+ Identifier.fromString(cloneFileInfo.sourceIdentifier()));
+ Path targetTableRootPath =
+ targetCatalog.getDataTableLocation(
+ Identifier.fromString(cloneFileInfo.targetIdentifier()));
+
+ Path filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
+ Path sourcePath = new Path(sourceTableRootPath.toString() + filePathExcludeTableRoot);
+ Path targetPath = new Path(targetTableRootPath.toString() + filePathExcludeTableRoot);
+
+ if (targetTableFileIO.exists(targetPath)
+ && targetTableFileIO.getFileSize(targetPath)
+ == sourceTableFileIO.getFileSize(sourcePath)) {
+ LOG.info(
+ "Skipping target file {} because it already exists and has the same size.",
+ targetPath);
+ return;
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598409674
##########
paimon-core/src/main/java/org/apache/paimon/clone/CloneFileInfo.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.clone;
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/SnapshotHintOperator.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+
+/** Operator. */
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605804301
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "true"));
Review Comment:
done
##########
docs/content/migration/clone-tables.md:
##########
@@ -0,0 +1,116 @@
+---
+title: "Clone Tables"
+weight: 3
+type: docs
+aliases:
+- /migration/clone-tables.html
+---
+<!--
+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.
+-->
+
+# Clone Tables
+
+Paimon supports clone tables of the latest Snapshot for data migration.
+
+{{< hint info >}}
+1、Clone Tables only support batch mode yet. Please use -D execution.runtime-mode=batch or -yD execution.runtime-mode=batch (for the ON-YARN scenario) to run clone job.
+
+2、If you want clone job runs quickly, you can add parameter parallelism.
+
+3、Only support Flink now.
+{{< /hint >}}
+
+## Clone Table
+The target table needs to be a non-existent table, and it will have the exact same schema (only the schema for current snapshot) as the source table.
+
+To run a Flink batch job for clone, follow these instructions.
+
+### LatestSnapshot
+Clone the latest snapshot of the source table, copying all the files required for the snapshot to the new target table.
+
+{{< tabs "clone-tables" >}}
+
+{{< tab "Flink" >}}
+
+Flink SQL currently does not support statements related to clone, so we have to submit the clone job through `flink run`.
+
+Run the following command to submit a clone job for the table's latest Snapshot.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone \
+ --warehouse <source-warehouse-path> \
+ [--database <source-database-name>] \
+ [--table <source-table-name>] \
+ [--catalog_conf <source-paimon-catalog-conf> [--catalog_conf <source-paimon-catalog-conf> ...]] \
+ --target_warehouse <target-warehouse-path> \
+ --target_database <target-database> \
+ --target_table <target-table-name> \
+ [--target_catalog_conf <target-paimon-catalog-conf> [--target_catalog_conf <target-paimon-catalog-conf> ...]]
+ [--parallelism 128 ]
+```
+
+{{< hint info >}}
+1、If the database parameter is not passed, then all tables of all databases will be cloned.
+2、If the table parameter is not passed, then all tables of the database will be cloned.
+{{< /hint >}}
+
+Example: clone table latest Snapshot.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone \
+ --warehouse s3:///path/to/warehouse_source \
+ --database test_db \
+ --table test_table \
+ --catalog_conf s3.endpoint=https://****.com \
+ --catalog_conf s3.access-key=***** \
+ --catalog_conf s3.secret-key=***** \
+ --target_warehouse s3:///path/to/warehouse_target \
+ --target_database test_db \
+ --target_table test_table \
+ --target_catalog_conf s3.endpoint=https://****.com \
+ --target_catalog_conf s3.access-key=***** \
+ --target_catalog_conf s3.secret-key=*****
+```
+
+For more usage of the clone action, see
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone --help
+```
+
+{{< /tab >}}
+
+{{< tab "Flink Procedure" >}}
+
+Run the following command to submit a clone job for the table's latest Snapshot.
+
+```bash
+CALL sys.clone('source_warehouse', 'source_database', 'source_table', '', 'target_warehouse', 'target_database', 'target_table', '', '')
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1607881923
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.procedure;
+
+import org.apache.paimon.flink.action.CloneAction;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.apache.paimon.utils.ParameterUtils.parseCommaSeparatedKeyValues;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/**
+ * Clone snapshot/tag/table procedure. Usage:
+ *
+ * <pre><code>
+ * CALL sys.clone('warehouse', 'database', 'table', 'catalog_config', 'target_warehouse', 'target_database',
+ * 'target_table', 'target_catalog_config', 'parallelism')
+ * </code></pre>
+ */
+public class CloneProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "clone";
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String warehouse,
+ String database,
+ String table,
+ String catalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTable,
+ String targetCatalogConfig,
+ String parallelism)
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java:
##########
@@ -43,6 +44,29 @@
/** IT Case for {@link CompactProcedure}. */
public class CompactProcedureITCase extends CatalogITCaseBase {
+ private Object lock = new Object();
+
+ public void lockAndRun(Function function) throws Exception {
+ synchronized (lock) {
+ function.apply(1);
+ }
+ }
+
+ @Test
+ public void testhhh() throws Exception {
+ lockAndRun(
+ a -> {
+ System.out.println(a + "11111");
+ return a;
+ });
+
+ lockAndRun(
+ a -> {
+ System.out.println(a + "22222");
+ return a;
+ });
+ }
+
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598407499
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.PickFilesUtil;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** PickFilesForCloneOperator. */
+public class PickFilesForCloneOperator extends AbstractStreamOperator<CloneFileInfo>
+ implements OneInputStreamOperator<Tuple2<String, String>, CloneFileInfo> {
+
+ private final Map<String, String> sourceCatalogConfig;
+ private final Map<String, String> targetCatalogConfig;
+
+ private Catalog sourceCatalog;
+ private Catalog targetCatalog;
+
+ public PickFilesForCloneOperator(
+ Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+ targetCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(targetCatalogConfig));
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<String, String>> streamRecord) throws Exception {
+ try {
+ processElementImpl(streamRecord);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to pick files", e);
+ }
+ }
+
+ private void processElementImpl(StreamRecord<Tuple2<String, String>> streamRecord)
+ throws Exception {
+ String sourceIdentifierStr = streamRecord.getValue().f0;
+ Identifier sourceIdentifier = Identifier.fromString(sourceIdentifierStr);
+ String targetIdentifierStr = streamRecord.getValue().f1;
+ Identifier targetIdentifier = Identifier.fromString(targetIdentifierStr);
+
+ FileStoreTable sourceTable = (FileStoreTable) sourceCatalog.getTable(sourceIdentifier);
+ FileStore<?> store = sourceTable.store();
+ SnapshotManager snapshotManager = store.snapshotManager();
+ ManifestList manifestList = store.manifestListFactory().create();
+ IndexFileHandler indexFileHandler = store.newIndexFileHandler();
+ SchemaManager schemaManager =
+ new SchemaManager(sourceTable.fileIO(), sourceTable.location());
+
+ targetCatalog.createDatabase(targetIdentifier.getDatabaseName(), true);
+ targetCatalog.createTable(
+ targetIdentifier, Schema.fromTableSchema(sourceTable.schema()), true);
+
+ List<CloneFileInfo> result =
+ toCloneFileInfos(
+ PickFilesUtil.getUsedFilesForSnapshot(
+ snapshotManager.latestSnapshot(),
+ snapshotManager,
+ sourceTable.store().pathFactory(),
+ sourceTable.store().newScan(),
+ manifestList,
+ schemaManager,
+ indexFileHandler),
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598408129
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.PickFilesUtil;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** PickFilesForCloneOperator. */
+public class PickFilesForCloneOperator extends AbstractStreamOperator<CloneFileInfo>
+ implements OneInputStreamOperator<Tuple2<String, String>, CloneFileInfo> {
+
+ private final Map<String, String> sourceCatalogConfig;
+ private final Map<String, String> targetCatalogConfig;
+
+ private Catalog sourceCatalog;
+ private Catalog targetCatalog;
+
+ public PickFilesForCloneOperator(
+ Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+ targetCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(targetCatalogConfig));
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<String, String>> streamRecord) throws Exception {
+ try {
+ processElementImpl(streamRecord);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to pick files", e);
+ }
+ }
+
+ private void processElementImpl(StreamRecord<Tuple2<String, String>> streamRecord)
+ throws Exception {
+ String sourceIdentifierStr = streamRecord.getValue().f0;
+ Identifier sourceIdentifier = Identifier.fromString(sourceIdentifierStr);
+ String targetIdentifierStr = streamRecord.getValue().f1;
+ Identifier targetIdentifier = Identifier.fromString(targetIdentifierStr);
+
+ FileStoreTable sourceTable = (FileStoreTable) sourceCatalog.getTable(sourceIdentifier);
+ FileStore<?> store = sourceTable.store();
+ SnapshotManager snapshotManager = store.snapshotManager();
+ ManifestList manifestList = store.manifestListFactory().create();
+ IndexFileHandler indexFileHandler = store.newIndexFileHandler();
+ SchemaManager schemaManager =
+ new SchemaManager(sourceTable.fileIO(), sourceTable.location());
+
+ targetCatalog.createDatabase(targetIdentifier.getDatabaseName(), true);
+ targetCatalog.createTable(
+ targetIdentifier, Schema.fromTableSchema(sourceTable.schema()), true);
+
+ List<CloneFileInfo> result =
+ toCloneFileInfos(
+ PickFilesUtil.getUsedFilesForSnapshot(
+ snapshotManager.latestSnapshot(),
+ snapshotManager,
+ sourceTable.store().pathFactory(),
+ sourceTable.store().newScan(),
+ manifestList,
+ schemaManager,
+ indexFileHandler),
+ sourceTable.location(),
+ sourceIdentifierStr,
+ targetIdentifierStr);
+
+ for (CloneFileInfo info : result) {
+ if (info.getFilePathExcludeTableRoot().toString().endsWith("snapshot/EARLIEST")
+ || info.getFilePathExcludeTableRoot().toString().endsWith("snapshot/LATEST")) {
Review Comment:
cut this code.
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(
+ Snapshot snapshot,
+ SnapshotManager snapshotManager,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ SchemaManager schemaManager,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot, pathFactory, scan, manifestList, indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles =
+ scan.withSnapshot(snapshot).readSimpleEntries().stream()
+ .map(
+ f ->
+ pathFactory
+ .createDataFilePathFactory(
+ f.partition(), f.bucket())
+ .toPath(f.fileName()))
+ .collect(Collectors.toList());
Review Comment:
done
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(
+ Snapshot snapshot,
+ SnapshotManager snapshotManager,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ SchemaManager schemaManager,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot, pathFactory, scan, manifestList, indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles =
+ scan.withSnapshot(snapshot).readSimpleEntries().stream()
+ .map(
+ f ->
+ pathFactory
+ .createDataFilePathFactory(
+ f.partition(), f.bucket())
+ .toPath(f.fileName()))
+ .collect(Collectors.toList());
+ Collections.reverse(dataFiles);
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605804024
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesUtil.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesUtil.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(FileStoreTable table) {
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598408822
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(
+ Snapshot snapshot,
+ SnapshotManager snapshotManager,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ SchemaManager schemaManager,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot, pathFactory, scan, manifestList, indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles =
+ scan.withSnapshot(snapshot).readSimpleEntries().stream()
+ .map(
+ f ->
+ pathFactory
+ .createDataFilePathFactory(
+ f.partition(), f.bucket())
+ .toPath(f.fileName()))
+ .collect(Collectors.toList());
+ Collections.reverse(dataFiles);
+ files.addAll(dataFiles);
+
+ // try to read index files
+ String indexManifest = snapshot.indexManifest();
+ if (indexManifest != null && indexFileHandler.existsManifest(indexManifest)) {
+ files.add(pathFactory.indexManifestFileFactory().toPath(indexManifest));
+
+ List<IndexManifestEntry> indexManifestEntries =
+ retryReadingFiles(
+ () -> indexFileHandler.readManifestWithIOException(indexManifest));
+ if (indexManifestEntries == null) {
+ return Collections.emptyList();
+ }
+
+ indexManifestEntries.stream()
+ .map(IndexManifestEntry::indexFile)
+ .map(indexFileHandler::filePath)
+ .forEach(files::add);
+ }
+
+ // add statistic file
+ if (snapshot.statistics() != null) {
+ files.add(pathFactory.statsFileFactory().toPath(snapshot.statistics()));
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ return files;
+ }
+
+ private static void addManifestList(
+ List<Path> used, Snapshot snapshot, FileStorePathFactory pathFactory) {
+ used.add(pathFactory.toManifestListPath(snapshot.baseManifestList()));
+ used.add(pathFactory.toManifestListPath(snapshot.deltaManifestList()));
+ String changelogManifestList = snapshot.changelogManifestList();
+ if (changelogManifestList != null) {
+ used.add(pathFactory.toManifestListPath(changelogManifestList));
+ }
+ }
+
+ private static List<ManifestFileMeta> readAllManifestsWithIOException(
+ Snapshot snapshot, ManifestList manifestList) throws IOException {
+ List<ManifestFileMeta> result = new ArrayList<>();
+
+ result.addAll(manifestList.readWithIOException(snapshot.baseManifestList()));
+ result.addAll(manifestList.readWithIOException(snapshot.deltaManifestList()));
+
+ String changelogManifestList = snapshot.changelogManifestList();
+ if (changelogManifestList != null) {
+ result.addAll(manifestList.readWithIOException(changelogManifestList));
+ }
+
+ return result;
+ }
+
+ @Nullable
+ private static <T> T retryReadingFiles(OrphanFilesClean.ReaderWithIOException<T> reader)
Review Comment:
done
##########
paimon-core/src/main/java/org/apache/paimon/clone/CopyFileUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.IOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** Utility class for copy file. */
+public class CopyFileUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CopyFileUtils.class);
+
+ public static void copyFile(
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1598406111
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** CloneSourceBuilder. */
+public class CloneSourceBuilder {
+
+ private final StreamExecutionEnvironment env;
+ private final Map<String, String> sourceCatalogConfig;
+ private final String database;
+ private final String tableName;
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneSourceBuilder(
+ StreamExecutionEnvironment env,
+ Map<String, String> sourceCatalogConfig,
+ String database,
+ String tableName,
+ String targetDatabase,
+ String targetTableName) {
+ this.env = env;
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.database = database;
+ this.tableName = tableName;
+ this.targetDatabase = targetDatabase;
+ this.targetTableName = targetTableName;
+ }
+
+ public DataStream<Tuple2<String, String>> build() {
+ Catalog sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+
+ List<Tuple2<String, String>> result = new ArrayList<>();
+
+ if (database == null) {
+ checkArgument(tableName == null, "tableName must be null when database is null.");
+ for (String db : sourceCatalog.listDatabases()) {
+ try {
+ for (String table : sourceCatalog.listTables(db)) {
+ String s = db + "." + table;
+ result.add(new Tuple2<>(s, s));
+ }
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ } else if (tableName == null) {
+ try {
+ for (String table : sourceCatalog.listTables(database)) {
+ result.add(new Tuple2<>(database + "." + table, targetDatabase + "." + table));
+ }
+ } catch (Exception e) {
+ // ignore
+ }
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605804055
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesUtil.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(FileStoreTable table) {
+ FileStore<?> store = table.store();
+ SnapshotManager snapshotManager = store.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ ManifestList manifestList = store.manifestListFactory().create();
+ SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location());
+ IndexFileHandler indexFileHandler = store.newIndexFileHandler();
+
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot,
+ store.pathFactory(),
+ store.newScan(),
+ manifestList,
+ indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles = new ArrayList<>();
+ List<SimpleFileEntry> simpleFileEntries =
+ scan.withSnapshot(snapshot).readSimpleEntries();
+ for (SimpleFileEntry simpleFileEntry : simpleFileEntries) {
+ Path dataFilePath =
+ pathFactory
+ .createDataFilePathFactory(
+ simpleFileEntry.partition(), simpleFileEntry.bucket())
+ .toPath(simpleFileEntry.fileName());
+ dataFiles.add(dataFilePath);
+ }
+ // In order to make files created earlier perform copying earlier.
+ Collections.reverse(dataFiles);
Review Comment:
done
##########
paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java:
##########
@@ -499,8 +499,7 @@ private void deleteFileOrDirQuietly(Path path) {
/** A helper functional interface for method {@link #retryReadingFiles}. */
@FunctionalInterface
- private interface ReaderWithIOException<T> {
-
+ interface ReaderWithIOException<T> {
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/SnapshotHintChannelComputer.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.table.sink.ChannelComputer;
+
+/** SnapshotHintChannelComputer. */
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1597921571
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.clone.CloneFileInfo;
+import org.apache.paimon.clone.SnapshotHintChannelComputer;
+import org.apache.paimon.flink.clone.CloneFileInfoTypeInfo;
+import org.apache.paimon.flink.clone.CloneSourceBuilder;
+import org.apache.paimon.flink.clone.CopyFileOperator;
+import org.apache.paimon.flink.clone.PickFilesForCloneOperator;
+import org.apache.paimon.flink.clone.SnapshotHintOperator;
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.options.CatalogOptions;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/** Snapshot/Tag/Table clone action for Flink. */
+public class CloneAction extends ActionBase {
+
+ private final int parallelism;
+
+ private Map<String, String> sourceCatalogConfig = new HashMap<>();
+ private final String database;
+ private final String tableName;
+
+ private Map<String, String> targetCatalogConfig = new HashMap<>();
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneAction(
+ String warehouse,
+ String database,
+ String tableName,
+ Map<String, String> sourceCatalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTableName,
+ Map<String, String> targetCatalogConfig,
+ String parallelismStr) {
+ super(warehouse, sourceCatalogConfig);
+
+ checkNotNull(warehouse, "warehouse must not be null.");
+ checkNotNull(targetWarehouse, "targetWarehouse must not be null.");
+ checkNotNull(targetDatabase, "targetDatabase must not be null.");
+ checkNotNull(targetTableName, "targetTableName must not be null.");
Review Comment:
Why can't users just clone the whole warehouse or the whole database? Why must he/she specify target table name?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** CloneSourceBuilder. */
+public class CloneSourceBuilder {
+
+ private final StreamExecutionEnvironment env;
+ private final Map<String, String> sourceCatalogConfig;
+ private final String database;
+ private final String tableName;
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneSourceBuilder(
+ StreamExecutionEnvironment env,
+ Map<String, String> sourceCatalogConfig,
+ String database,
+ String tableName,
+ String targetDatabase,
+ String targetTableName) {
+ this.env = env;
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.database = database;
+ this.tableName = tableName;
+ this.targetDatabase = targetDatabase;
+ this.targetTableName = targetTableName;
+ }
+
+ public DataStream<Tuple2<String, String>> build() {
+ Catalog sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+
+ List<Tuple2<String, String>> result = new ArrayList<>();
+
+ if (database == null) {
+ checkArgument(tableName == null, "tableName must be null when database is null.");
+ for (String db : sourceCatalog.listDatabases()) {
+ try {
+ for (String table : sourceCatalog.listTables(db)) {
+ String s = db + "." + table;
+ result.add(new Tuple2<>(s, s));
+ }
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ } else if (tableName == null) {
+ try {
+ for (String table : sourceCatalog.listTables(database)) {
+ result.add(new Tuple2<>(database + "." + table, targetDatabase + "." + table));
+ }
+ } catch (Exception e) {
+ // ignore
+ }
Review Comment:
Why ignore this exception? What if the user mis-typed the database and the database does not exist?
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(
+ Snapshot snapshot,
+ SnapshotManager snapshotManager,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ SchemaManager schemaManager,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot, pathFactory, scan, manifestList, indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles =
+ scan.withSnapshot(snapshot).readSimpleEntries().stream()
+ .map(
+ f ->
+ pathFactory
+ .createDataFilePathFactory(
+ f.partition(), f.bucket())
+ .toPath(f.fileName()))
+ .collect(Collectors.toList());
+ Collections.reverse(dataFiles);
+ files.addAll(dataFiles);
+
+ // try to read index files
+ String indexManifest = snapshot.indexManifest();
+ if (indexManifest != null && indexFileHandler.existsManifest(indexManifest)) {
+ files.add(pathFactory.indexManifestFileFactory().toPath(indexManifest));
+
+ List<IndexManifestEntry> indexManifestEntries =
+ retryReadingFiles(
+ () -> indexFileHandler.readManifestWithIOException(indexManifest));
+ if (indexManifestEntries == null) {
+ return Collections.emptyList();
+ }
+
+ indexManifestEntries.stream()
+ .map(IndexManifestEntry::indexFile)
+ .map(indexFileHandler::filePath)
+ .forEach(files::add);
+ }
+
+ // add statistic file
+ if (snapshot.statistics() != null) {
+ files.add(pathFactory.statsFileFactory().toPath(snapshot.statistics()));
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ return files;
+ }
+
+ private static void addManifestList(
+ List<Path> used, Snapshot snapshot, FileStorePathFactory pathFactory) {
+ used.add(pathFactory.toManifestListPath(snapshot.baseManifestList()));
+ used.add(pathFactory.toManifestListPath(snapshot.deltaManifestList()));
+ String changelogManifestList = snapshot.changelogManifestList();
+ if (changelogManifestList != null) {
+ used.add(pathFactory.toManifestListPath(changelogManifestList));
+ }
+ }
+
+ private static List<ManifestFileMeta> readAllManifestsWithIOException(
+ Snapshot snapshot, ManifestList manifestList) throws IOException {
+ List<ManifestFileMeta> result = new ArrayList<>();
+
+ result.addAll(manifestList.readWithIOException(snapshot.baseManifestList()));
+ result.addAll(manifestList.readWithIOException(snapshot.deltaManifestList()));
+
+ String changelogManifestList = snapshot.changelogManifestList();
+ if (changelogManifestList != null) {
+ result.addAll(manifestList.readWithIOException(changelogManifestList));
+ }
+
+ return result;
+ }
+
+ @Nullable
+ private static <T> T retryReadingFiles(OrphanFilesClean.ReaderWithIOException<T> reader)
Review Comment:
No need to use interface from `OrphanFilesClean`. This interface is simple, just create one in this file.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** CloneSourceBuilder. */
Review Comment:
Please write useful comments. What does this class do? What kind of source it will build? What kind of record it will produce? What's the meaning of each field in the record?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFileInfoTypeInfo.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.flink.clone;
+
+import org.apache.paimon.clone.CloneFileInfo;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+
+/** Type information of {@link CloneFileInfo}. */
+public class CloneFileInfoTypeInfo extends TypeInformation<CloneFileInfo> {
Review Comment:
Why do we need this class? `CloneFileInfo` is a POJO, which is supported by Flink.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.PickFilesUtil;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** PickFilesForCloneOperator. */
Review Comment:
Same as above. Please write useful comments.
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(
+ Snapshot snapshot,
+ SnapshotManager snapshotManager,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ SchemaManager schemaManager,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot, pathFactory, scan, manifestList, indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles =
+ scan.withSnapshot(snapshot).readSimpleEntries().stream()
+ .map(
+ f ->
+ pathFactory
+ .createDataFilePathFactory(
+ f.partition(), f.bucket())
+ .toPath(f.fileName()))
+ .collect(Collectors.toList());
+ Collections.reverse(dataFiles);
Review Comment:
Why reverse? Add comments.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.PickFilesUtil;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** PickFilesForCloneOperator. */
+public class PickFilesForCloneOperator extends AbstractStreamOperator<CloneFileInfo>
+ implements OneInputStreamOperator<Tuple2<String, String>, CloneFileInfo> {
+
+ private final Map<String, String> sourceCatalogConfig;
+ private final Map<String, String> targetCatalogConfig;
+
+ private Catalog sourceCatalog;
+ private Catalog targetCatalog;
+
+ public PickFilesForCloneOperator(
+ Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+ targetCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(targetCatalogConfig));
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<String, String>> streamRecord) throws Exception {
+ try {
+ processElementImpl(streamRecord);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to pick files", e);
+ }
+ }
+
+ private void processElementImpl(StreamRecord<Tuple2<String, String>> streamRecord)
+ throws Exception {
+ String sourceIdentifierStr = streamRecord.getValue().f0;
+ Identifier sourceIdentifier = Identifier.fromString(sourceIdentifierStr);
+ String targetIdentifierStr = streamRecord.getValue().f1;
+ Identifier targetIdentifier = Identifier.fromString(targetIdentifierStr);
+
+ FileStoreTable sourceTable = (FileStoreTable) sourceCatalog.getTable(sourceIdentifier);
+ FileStore<?> store = sourceTable.store();
+ SnapshotManager snapshotManager = store.snapshotManager();
+ ManifestList manifestList = store.manifestListFactory().create();
+ IndexFileHandler indexFileHandler = store.newIndexFileHandler();
+ SchemaManager schemaManager =
+ new SchemaManager(sourceTable.fileIO(), sourceTable.location());
+
+ targetCatalog.createDatabase(targetIdentifier.getDatabaseName(), true);
+ targetCatalog.createTable(
+ targetIdentifier, Schema.fromTableSchema(sourceTable.schema()), true);
+
+ List<CloneFileInfo> result =
+ toCloneFileInfos(
+ PickFilesUtil.getUsedFilesForSnapshot(
+ snapshotManager.latestSnapshot(),
+ snapshotManager,
+ sourceTable.store().pathFactory(),
+ sourceTable.store().newScan(),
+ manifestList,
+ schemaManager,
+ indexFileHandler),
Review Comment:
No need for these many arguments. Just use `sourceTable` as argument and extract the fields inside the function.
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(
+ Snapshot snapshot,
+ SnapshotManager snapshotManager,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ SchemaManager schemaManager,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot, pathFactory, scan, manifestList, indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles =
+ scan.withSnapshot(snapshot).readSimpleEntries().stream()
+ .map(
+ f ->
+ pathFactory
+ .createDataFilePathFactory(
+ f.partition(), f.bucket())
+ .toPath(f.fileName()))
+ .collect(Collectors.toList());
Review Comment:
Indentation is too deep. Use for loop instead.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesForCloneOperator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.PickFilesUtil;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** PickFilesForCloneOperator. */
+public class PickFilesForCloneOperator extends AbstractStreamOperator<CloneFileInfo>
+ implements OneInputStreamOperator<Tuple2<String, String>, CloneFileInfo> {
+
+ private final Map<String, String> sourceCatalogConfig;
+ private final Map<String, String> targetCatalogConfig;
+
+ private Catalog sourceCatalog;
+ private Catalog targetCatalog;
+
+ public PickFilesForCloneOperator(
+ Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+ targetCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(targetCatalogConfig));
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<String, String>> streamRecord) throws Exception {
+ try {
+ processElementImpl(streamRecord);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to pick files", e);
+ }
+ }
+
+ private void processElementImpl(StreamRecord<Tuple2<String, String>> streamRecord)
+ throws Exception {
+ String sourceIdentifierStr = streamRecord.getValue().f0;
+ Identifier sourceIdentifier = Identifier.fromString(sourceIdentifierStr);
+ String targetIdentifierStr = streamRecord.getValue().f1;
+ Identifier targetIdentifier = Identifier.fromString(targetIdentifierStr);
+
+ FileStoreTable sourceTable = (FileStoreTable) sourceCatalog.getTable(sourceIdentifier);
+ FileStore<?> store = sourceTable.store();
+ SnapshotManager snapshotManager = store.snapshotManager();
+ ManifestList manifestList = store.manifestListFactory().create();
+ IndexFileHandler indexFileHandler = store.newIndexFileHandler();
+ SchemaManager schemaManager =
+ new SchemaManager(sourceTable.fileIO(), sourceTable.location());
+
+ targetCatalog.createDatabase(targetIdentifier.getDatabaseName(), true);
+ targetCatalog.createTable(
+ targetIdentifier, Schema.fromTableSchema(sourceTable.schema()), true);
+
+ List<CloneFileInfo> result =
+ toCloneFileInfos(
+ PickFilesUtil.getUsedFilesForSnapshot(
+ snapshotManager.latestSnapshot(),
+ snapshotManager,
+ sourceTable.store().pathFactory(),
+ sourceTable.store().newScan(),
+ manifestList,
+ schemaManager,
+ indexFileHandler),
+ sourceTable.location(),
+ sourceIdentifierStr,
+ targetIdentifierStr);
+
+ for (CloneFileInfo info : result) {
+ if (info.getFilePathExcludeTableRoot().toString().endsWith("snapshot/EARLIEST")
+ || info.getFilePathExcludeTableRoot().toString().endsWith("snapshot/LATEST")) {
Review Comment:
```suggestion
if (info.getFilePathExcludeTableRoot().toString().endsWith("snapshot/" + SnapshotManager.EARLIEST)
|| info.getFilePathExcludeTableRoot().toString().endsWith("snapshot/" + SnapshotManager.LATEST)) {
```
Why skipping these two files? Add comments.
##########
paimon-core/src/main/java/org/apache/paimon/operation/PickFilesUtil.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
Review Comment:
Move this class to `org.apache.paimon.flink.clone`. It is only used in clone action.
##########
paimon-core/src/main/java/org/apache/paimon/clone/CopyFileUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.IOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** Utility class for copy file. */
+public class CopyFileUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CopyFileUtils.class);
+
+ public static void copyFile(
+ CloneFileInfo cloneFileInfo,
+ FileIO sourceTableFileIO,
+ FileIO targetTableFileIO,
+ Path sourceTableRootPath,
+ Path targetTableRootPath)
+ throws IOException {
+ Path filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
+ Path sourcePath = new Path(sourceTableRootPath.toString() + filePathExcludeTableRoot);
+ Path targetPath = new Path(targetTableRootPath.toString() + filePathExcludeTableRoot);
+
+ if (targetTableFileIO.exists(targetPath)
+ && targetTableFileIO.getFileSize(targetPath)
+ == sourceTableFileIO.getFileSize(sourcePath)) {
+ LOG.info(
+ "Skipping target file {} because it already exists and has the same size.",
+ targetPath);
+ return;
+ }
+
+ LOG.debug("Begin copy file from {} to {}.", sourcePath, targetPath);
Review Comment:
```suggestion
if (LOG.isDebugEnabled()) {
LOG.debug("Begin copy file from {} to {}.", sourcePath, targetPath);
}
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/SnapshotHintOperator.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.clone.CloneFileInfo;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+
+/** Operator. */
Review Comment:
What is this? Write meaningful comments.
##########
paimon-core/src/main/java/org/apache/paimon/clone/CopyFileUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.IOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** Utility class for copy file. */
+public class CopyFileUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CopyFileUtils.class);
+
+ public static void copyFile(
Review Comment:
No need for a util class. There is only one place using this method.
##########
paimon-core/src/main/java/org/apache/paimon/clone/CloneFileInfo.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.clone;
Review Comment:
Move everything in this package to `paimon-flink-common` module.
##########
paimon-core/src/main/java/org/apache/paimon/clone/CopyFileUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.IOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/** Utility class for copy file. */
+public class CopyFileUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CopyFileUtils.class);
+
+ public static void copyFile(
+ CloneFileInfo cloneFileInfo,
+ FileIO sourceTableFileIO,
+ FileIO targetTableFileIO,
+ Path sourceTableRootPath,
+ Path targetTableRootPath)
+ throws IOException {
+ Path filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
+ Path sourcePath = new Path(sourceTableRootPath.toString() + filePathExcludeTableRoot);
+ Path targetPath = new Path(targetTableRootPath.toString() + filePathExcludeTableRoot);
+
+ if (targetTableFileIO.exists(targetPath)
+ && targetTableFileIO.getFileSize(targetPath)
+ == sourceTableFileIO.getFileSize(sourcePath)) {
+ LOG.info(
+ "Skipping target file {} because it already exists and has the same size.",
+ targetPath);
+ return;
+ }
+
+ LOG.debug("Begin copy file from {} to {}.", sourcePath, targetPath);
+ IOUtils.copyBytes(
+ sourceTableFileIO.newInputStream(sourcePath),
+ targetTableFileIO.newOutputStream(targetPath, true));
+ LOG.debug("End copy file from {} to {}.", sourcePath, targetPath);
Review Comment:
Same as above.
--
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] support clone latest snapshot [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1602534247
##########
docs/content/migration/clone-tables.md:
##########
@@ -0,0 +1,116 @@
+---
+title: "Clone Tables"
+weight: 3
+type: docs
+aliases:
+- /migration/clone-tables.html
+---
+<!--
+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.
+-->
+
+# Clone Tables
+
+Paimon supports clone tables of the latest Snapshot for data migration.
+
+{{< hint info >}}
+1、Clone Tables only support batch mode yet. Please use -D execution.runtime-mode=batch or -yD execution.runtime-mode=batch (for the ON-YARN scenario) to run clone job.
+
+2、If you want clone job runs quickly, you can add parameter parallelism.
+
+3、Only support Flink now.
+{{< /hint >}}
+
+## Clone Table
+The target table needs to be a non-existent table, and it will have the exact same schema (only the schema for current snapshot) as the source table.
+
+To run a Flink batch job for clone, follow these instructions.
+
+### LatestSnapshot
+Clone the latest snapshot of the source table, copying all the files required for the snapshot to the new target table.
+
+{{< tabs "clone-tables" >}}
+
+{{< tab "Flink" >}}
+
+Flink SQL currently does not support statements related to clone, so we have to submit the clone job through `flink run`.
+
+Run the following command to submit a clone job for the table's latest Snapshot.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone \
+ --warehouse <source-warehouse-path> \
+ [--database <source-database-name>] \
+ [--table <source-table-name>] \
+ [--catalog_conf <source-paimon-catalog-conf> [--catalog_conf <source-paimon-catalog-conf> ...]] \
+ --target_warehouse <target-warehouse-path> \
+ --target_database <target-database> \
+ --target_table <target-table-name> \
+ [--target_catalog_conf <target-paimon-catalog-conf> [--target_catalog_conf <target-paimon-catalog-conf> ...]]
+ [--parallelism 128 ]
+```
+
+{{< hint info >}}
+1、If the database parameter is not passed, then all tables of all databases will be cloned.
+2、If the table parameter is not passed, then all tables of the database will be cloned.
+{{< /hint >}}
+
+Example: clone table latest Snapshot.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone \
+ --warehouse s3:///path/to/warehouse_source \
+ --database test_db \
+ --table test_table \
+ --catalog_conf s3.endpoint=https://****.com \
+ --catalog_conf s3.access-key=***** \
+ --catalog_conf s3.secret-key=***** \
+ --target_warehouse s3:///path/to/warehouse_target \
+ --target_database test_db \
+ --target_table test_table \
+ --target_catalog_conf s3.endpoint=https://****.com \
+ --target_catalog_conf s3.access-key=***** \
+ --target_catalog_conf s3.secret-key=*****
+```
+
+For more usage of the clone action, see
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone --help
+```
+
+{{< /tab >}}
+
+{{< tab "Flink Procedure" >}}
+
+Run the following command to submit a clone job for the table's latest Snapshot.
+
+```bash
+CALL sys.clone('source_warehouse', 'source_database', 'source_table', '', 'target_warehouse', 'target_database', 'target_table', '', '')
Review Comment:
Just support Flink 1.19. Using named argurment.
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605804308
##########
docs/content/migration/clone-tables.md:
##########
@@ -0,0 +1,116 @@
+---
+title: "Clone Tables"
+weight: 3
+type: docs
+aliases:
+- /migration/clone-tables.html
+---
+<!--
+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.
+-->
+
+# Clone Tables
+
+Paimon supports clone tables of the latest Snapshot for data migration.
+
+{{< hint info >}}
+1、Clone Tables only support batch mode yet. Please use -D execution.runtime-mode=batch or -yD execution.runtime-mode=batch (for the ON-YARN scenario) to run clone job.
+
+2、If you want clone job runs quickly, you can add parameter parallelism.
+
+3、Only support Flink now.
+{{< /hint >}}
+
+## Clone Table
+The target table needs to be a non-existent table, and it will have the exact same schema (only the schema for current snapshot) as the source table.
+
+To run a Flink batch job for clone, follow these instructions.
+
+### LatestSnapshot
+Clone the latest snapshot of the source table, copying all the files required for the snapshot to the new target table.
+
+{{< tabs "clone-tables" >}}
+
+{{< tab "Flink" >}}
+
+Flink SQL currently does not support statements related to clone, so we have to submit the clone job through `flink run`.
+
+Run the following command to submit a clone job for the table's latest Snapshot.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone \
+ --warehouse <source-warehouse-path> \
+ [--database <source-database-name>] \
+ [--table <source-table-name>] \
+ [--catalog_conf <source-paimon-catalog-conf> [--catalog_conf <source-paimon-catalog-conf> ...]] \
+ --target_warehouse <target-warehouse-path> \
+ --target_database <target-database> \
+ --target_table <target-table-name> \
+ [--target_catalog_conf <target-paimon-catalog-conf> [--target_catalog_conf <target-paimon-catalog-conf> ...]]
+ [--parallelism 128 ]
+```
+
+{{< hint info >}}
+1、If the database parameter is not passed, then all tables of all databases will be cloned.
+2、If the table parameter is not passed, then all tables of the database will be cloned.
+{{< /hint >}}
+
+Example: clone table latest Snapshot.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone \
+ --warehouse s3:///path/to/warehouse_source \
+ --database test_db \
+ --table test_table \
+ --catalog_conf s3.endpoint=https://****.com \
+ --catalog_conf s3.access-key=***** \
+ --catalog_conf s3.secret-key=***** \
+ --target_warehouse s3:///path/to/warehouse_target \
+ --target_database test_db \
+ --target_table test_table \
+ --target_catalog_conf s3.endpoint=https://****.com \
+ --target_catalog_conf s3.access-key=***** \
+ --target_catalog_conf s3.secret-key=*****
+```
+
+For more usage of the clone action, see
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ clone --help
+```
+
+{{< /tab >}}
+
+{{< tab "Flink Procedure" >}}
+
+Run the following command to submit a clone job for the table's latest Snapshot.
+
+```bash
+CALL sys.clone('source_warehouse', 'source_database', 'source_table', '', 'target_warehouse', 'target_database', 'target_table', '', '')
Review Comment:
In CloneProcedure I use Overload functions，If I use named arguments, ut case will throw:
Caused by: org.apache.flink.table.api.ValidationException: Could not find the argument names. Currently named arguments are not supported for varArgs and multi different argument names with overload function
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605804159
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CopyFileOperator.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.AbstractCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.IOUtils;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.Map;
+
+/** A Operator to copy files. */
+public class CopyFileOperator extends AbstractStreamOperator<CloneFileInfo>
+ implements OneInputStreamOperator<CloneFileInfo, CloneFileInfo> {
+
+ private final Map<String, String> sourceCatalogConfig;
+ private final Map<String, String> targetCatalogConfig;
+
+ private AbstractCatalog sourceCatalog;
+ private AbstractCatalog targetCatalog;
+
+ public CopyFileOperator(
+ Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
+ super();
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sourceCatalog =
+ (AbstractCatalog)
+ FlinkCatalogFactory.createPaimonCatalog(
+ Options.fromMap(sourceCatalogConfig));
+ targetCatalog =
+ (AbstractCatalog)
+ FlinkCatalogFactory.createPaimonCatalog(
+ Options.fromMap(targetCatalogConfig));
+ }
+
+ @Override
+ public void processElement(StreamRecord<CloneFileInfo> streamRecord) throws Exception {
+ CloneFileInfo cloneFileInfo = streamRecord.getValue();
+
+ FileIO sourceTableFileIO = sourceCatalog.fileIO();
+ FileIO targetTableFileIO = targetCatalog.fileIO();
+ Path sourceTableRootPath =
+ sourceCatalog.getDataTableLocation(
+ Identifier.fromString(cloneFileInfo.sourceIdentifier()));
+ Path targetTableRootPath =
+ targetCatalog.getDataTableLocation(
+ Identifier.fromString(cloneFileInfo.targetIdentifier()));
+
+ Path filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
+ Path sourcePath = new Path(sourceTableRootPath.toString() + filePathExcludeTableRoot);
+ Path targetPath = new Path(targetTableRootPath.toString() + filePathExcludeTableRoot);
+
+ if (targetTableFileIO.exists(targetPath)
+ && targetTableFileIO.getFileSize(targetPath)
+ == sourceTableFileIO.getFileSize(sourcePath)) {
+ LOG.info(
+ "Skipping target file {} because it already exists and has the same size.",
+ targetPath);
+ return;
Review Comment:
done, add output.collect(streamRecord);
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CloneProcedureITCase.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.flink.procedure;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT Case for {@link CloneProcedure}. */
+public class CloneProcedureITCase extends CatalogITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ @Test
+ public void testCloneLatestSnapshot() throws Exception {
+ sql(
+ "CREATE TABLE T ("
+ + " k INT,"
+ + " v INT,"
+ + " hh INT,"
+ + " dt STRING,"
+ + " PRIMARY KEY (k, dt, hh) NOT ENFORCED"
+ + ") PARTITIONED BY (dt, hh) WITH ("
+ + " 'write-only' = 'true',"
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605803915
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 CloneAction}. */
+public class CloneActionFactory implements ActionFactory {
+
+ private static final String IDENTIFIER = "clone";
+ private static final String PARALLELISM = "parallelism";
+ private static final String TARGET_WAREHOUSE = "target_warehouse";
+ private static final String TARGET_DATABASE = "target_database";
+ private static final String TARGET_TABLE = "target_table";
+ private static final String TARGET_CATALOG_CONF = "target_catalog_conf";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterToolAdapter params) {
+ CloneAction cloneAction =
+ new CloneAction(
+ params.get(WAREHOUSE),
+ params.get(DATABASE),
+ params.get(TABLE),
+ optionalConfigMap(params, CATALOG_CONF),
+ params.get(TARGET_WAREHOUSE),
+ params.get(TARGET_DATABASE),
+ params.get(TARGET_TABLE),
+ optionalConfigMap(params, TARGET_CATALOG_CONF),
+ params.get(PARALLELISM));
+
+ return Optional.of(cloneAction);
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println("Action \"clone\" runs a batch job for clone the latest Snapshot.");
+ System.out.println();
+
+ System.out.println("Syntax:");
+ System.out.println(
+ " clone --warehouse <warehouse_path>"
+ + "[--database <database_name>]"
+ + "[--table <table_name>]"
+ + "[--catalog_conf <source-paimon-catalog-conf> [--catalog_conf <source-paimon-catalog-conf> ...]]"
+ + "--target_warehouse <target_warehouse_path>"
+ + "--target_database <target_database_name> "
+ + "--target_table <target_table_name> "
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.clone.CloneFileInfo;
+import org.apache.paimon.flink.clone.CloneSourceBuilder;
+import org.apache.paimon.flink.clone.CopyFileOperator;
+import org.apache.paimon.flink.clone.PickFilesForCloneOperator;
+import org.apache.paimon.flink.clone.SnapshotHintChannelComputer;
+import org.apache.paimon.flink.clone.SnapshotHintOperator;
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.options.CatalogOptions;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/** Snapshot/Tag/Table clone action for Flink. */
+public class CloneAction extends ActionBase {
+
+ private final int parallelism;
+
+ private Map<String, String> sourceCatalogConfig = new HashMap<>();
+ private final String database;
+ private final String tableName;
+
+ private Map<String, String> targetCatalogConfig = new HashMap<>();
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneAction(
+ String warehouse,
+ String database,
+ String tableName,
+ Map<String, String> sourceCatalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTableName,
+ Map<String, String> targetCatalogConfig,
+ String parallelismStr) {
+ super(warehouse, sourceCatalogConfig);
+
+ checkNotNull(warehouse, "warehouse must not be null.");
+ checkNotNull(targetWarehouse, "targetWarehouse must not be null.");
+
+ this.parallelism =
+ isBlank(parallelismStr) ? env.getParallelism() : Integer.parseInt(parallelismStr);
+
+ if (!sourceCatalogConfig.isEmpty()) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ }
Review Comment:
No, If sourceCatalogConfig is Collections.emptyMap() from CloneActionFactory, then sourceCatalogConfig.put() will cause java.lang.UnsupportedOperationException.
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1607883466
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,400 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.flink.clone.PickFilesUtil;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.emptyMap());
+
+ /** write data to source table */
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString("20221208")),
+ rowData(2, 100, 16, BinaryString.fromString("20221208")),
+ rowData(2, 100, 15, BinaryString.fromString("20221209")));
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(3);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(3);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]",
+ "+I[2, 100, 15, 20221208]",
+ "+I[2, 100, 15, 20221209]",
+ "+I[2, 100, 16, 20221208]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testCopyFilesNameAndSize() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.INPUT.toString());
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ });
+
+ /** write data to source table */
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(1, 100, 15, BinaryString.fromString("20221201")),
+ rowData(2, 40000, 16, BinaryString.fromString("20221208")),
+ rowData(4, 100, 19, BinaryString.fromString("20221209"))),
+ 1);
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(24, 898, 1, BinaryString.fromString("20221209")),
+ rowData(12, 312, 90, BinaryString.fromString("20221208")),
+ rowData(29, 434, 15, BinaryString.fromString("20221209"))),
+ 2);
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(6);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(6);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221201]",
+ "+I[12, 312, 90, 20221208]",
+ "+I[2, 40000, 16, 20221208]",
+ "+I[24, 898, 1, 20221209]",
+ "+I[29, 434, 15, 20221209]",
+ "+I[4, 100, 19, 20221209]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+
+ /**
+ * check file name and file size of manifest && index && data && schema && changelog &&
+ * snapshot files
+ */
+ List<Path> targetTableFiles = PickFilesUtil.getUsedFilesForLatestSnapshot(targetTable);
+ List<Pair<Path, Path>> filesPathInfoList =
+ targetTableFiles.stream()
+ .map(
+ absolutePath ->
+ Pair.of(
+ absolutePath,
+ getPathExcludeTableRoot(
+ absolutePath, targetTable.location())))
+ .collect(Collectors.toList());
+
+ Path tableLocation = table.location();
+ for (Pair<Path, Path> filesPathInfo : filesPathInfoList) {
+ Path sourceTableFile = new Path(tableLocation.toString() + filesPathInfo.getRight());
+ Assertions.assertTrue(table.fileIO().exists(sourceTableFile));
+ Assertions.assertEquals(
+ table.fileIO().getFileSize(sourceTableFile),
+ targetTable.fileIO().getFileSize(filesPathInfo.getLeft()));
+ }
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithSlowExpire() throws Exception {
+ /** prepare source table */
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+ }
+ });
+
+ /** write data into source table and commit them */
+ Thread writeDataToSourceTablethread =
+ new Thread(
+ () -> {
+ try {
+ /** write and commit data to source table. */
+ while (true) {
+ writeData(
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
Review Comment:
OK, these code is not duplicated after I delete testSnapshotWithSlowExpire case.
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1605803994
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.clone.CloneFileInfo;
+import org.apache.paimon.flink.clone.CloneSourceBuilder;
+import org.apache.paimon.flink.clone.CopyFileOperator;
+import org.apache.paimon.flink.clone.PickFilesForCloneOperator;
+import org.apache.paimon.flink.clone.SnapshotHintChannelComputer;
+import org.apache.paimon.flink.clone.SnapshotHintOperator;
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.options.CatalogOptions;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/** Snapshot/Tag/Table clone action for Flink. */
+public class CloneAction extends ActionBase {
+
+ private final int parallelism;
+
+ private Map<String, String> sourceCatalogConfig = new HashMap<>();
+ private final String database;
+ private final String tableName;
+
+ private Map<String, String> targetCatalogConfig = new HashMap<>();
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneAction(
+ String warehouse,
+ String database,
+ String tableName,
+ Map<String, String> sourceCatalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTableName,
+ Map<String, String> targetCatalogConfig,
+ String parallelismStr) {
+ super(warehouse, sourceCatalogConfig);
+
+ checkNotNull(warehouse, "warehouse must not be null.");
+ checkNotNull(targetWarehouse, "targetWarehouse must not be null.");
+
+ this.parallelism =
+ isBlank(parallelismStr) ? env.getParallelism() : Integer.parseInt(parallelismStr);
+
+ if (!sourceCatalogConfig.isEmpty()) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ }
+ this.sourceCatalogConfig.put(CatalogOptions.WAREHOUSE.key(), warehouse);
+ this.database = database;
+ this.tableName = tableName;
+
+ if (!targetCatalogConfig.isEmpty()) {
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
Review Comment:
No, If targetCatalogConfig is Collections.emptyMap() from CloneActionFactory, then targetCatalogConfig.put() will cause java.lang.UnsupportedOperationException.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Pick the tables to be cloned based on the user input parameters. The record type of the build
+ * DataStream is Tuple2. The left element is the identifier of source table and the right element is
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Pick the tables to be cloned based on the user input parameters. The record type of the build
+ * DataStream is Tuple2. The left element is the identifier of source table and the right element is
+ * the identifier of target table.
+ */
+public class CloneSourceBuilder {
+
+ private final StreamExecutionEnvironment env;
+ private final Map<String, String> sourceCatalogConfig;
+ private final String database;
+ private final String tableName;
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneSourceBuilder(
+ StreamExecutionEnvironment env,
+ Map<String, String> sourceCatalogConfig,
+ String database,
+ String tableName,
+ String targetDatabase,
+ String targetTableName) {
+ this.env = env;
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.database = database;
+ this.tableName = tableName;
+ this.targetDatabase = targetDatabase;
+ this.targetTableName = targetTableName;
+ }
+
+ public DataStream<Tuple2<String, String>> build() {
+ Catalog sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+
+ List<Tuple2<String, String>> result = new ArrayList<>();
+
+ if (database == null) {
+ checkArgument(tableName == null, "tableName must be null when database is null.");
+ for (String db : sourceCatalog.listDatabases()) {
+ try {
+ for (String table : sourceCatalog.listTables(db)) {
+ String s = db + "." + table;
+ result.add(new Tuple2<>(s, s));
+ }
+ } catch (Exception e) {
+ // ignore
+ }
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1606316456
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java:
##########
@@ -43,6 +44,29 @@
/** IT Case for {@link CompactProcedure}. */
public class CompactProcedureITCase extends CatalogITCaseBase {
+ private Object lock = new Object();
+
+ public void lockAndRun(Function function) throws Exception {
+ synchronized (lock) {
+ function.apply(1);
+ }
+ }
+
+ @Test
+ public void testhhh() throws Exception {
+ lockAndRun(
+ a -> {
+ System.out.println(a + "11111");
+ return a;
+ });
+
+ lockAndRun(
+ a -> {
+ System.out.println(a + "22222");
+ return a;
+ });
+ }
+
Review Comment:
What is this?
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,400 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.flink.clone.PickFilesUtil;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.emptyMap());
+
+ /** write data to source table */
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString("20221208")),
+ rowData(2, 100, 16, BinaryString.fromString("20221208")),
+ rowData(2, 100, 15, BinaryString.fromString("20221209")));
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(3);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(3);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]",
+ "+I[2, 100, 15, 20221208]",
+ "+I[2, 100, 15, 20221209]",
+ "+I[2, 100, 16, 20221208]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testCopyFilesNameAndSize() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.INPUT.toString());
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ });
+
+ /** write data to source table */
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(1, 100, 15, BinaryString.fromString("20221201")),
+ rowData(2, 40000, 16, BinaryString.fromString("20221208")),
+ rowData(4, 100, 19, BinaryString.fromString("20221209"))),
+ 1);
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(24, 898, 1, BinaryString.fromString("20221209")),
+ rowData(12, 312, 90, BinaryString.fromString("20221208")),
+ rowData(29, 434, 15, BinaryString.fromString("20221209"))),
+ 2);
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(6);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(6);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221201]",
+ "+I[12, 312, 90, 20221208]",
+ "+I[2, 40000, 16, 20221208]",
+ "+I[24, 898, 1, 20221209]",
+ "+I[29, 434, 15, 20221209]",
+ "+I[4, 100, 19, 20221209]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+
+ /**
+ * check file name and file size of manifest && index && data && schema && changelog &&
+ * snapshot files
+ */
+ List<Path> targetTableFiles = PickFilesUtil.getUsedFilesForLatestSnapshot(targetTable);
+ List<Pair<Path, Path>> filesPathInfoList =
+ targetTableFiles.stream()
+ .map(
+ absolutePath ->
+ Pair.of(
+ absolutePath,
+ getPathExcludeTableRoot(
+ absolutePath, targetTable.location())))
+ .collect(Collectors.toList());
+
+ Path tableLocation = table.location();
+ for (Pair<Path, Path> filesPathInfo : filesPathInfoList) {
+ Path sourceTableFile = new Path(tableLocation.toString() + filesPathInfo.getRight());
+ Assertions.assertTrue(table.fileIO().exists(sourceTableFile));
+ Assertions.assertEquals(
+ table.fileIO().getFileSize(sourceTableFile),
+ targetTable.fileIO().getFileSize(filesPathInfo.getLeft()));
+ }
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithSlowExpire() throws Exception {
+ /** prepare source table */
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+ }
+ });
+
+ /** write data into source table and commit them */
+ Thread writeDataToSourceTablethread =
+ new Thread(
+ () -> {
+ try {
+ /** write and commit data to source table. */
+ while (true) {
+ writeData(
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
Review Comment:
Too many indention. Extract a method.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,400 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.flink.clone.PickFilesUtil;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.emptyMap());
+
+ /** write data to source table */
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString("20221208")),
+ rowData(2, 100, 16, BinaryString.fromString("20221208")),
+ rowData(2, 100, 15, BinaryString.fromString("20221209")));
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(3);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(3);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]",
+ "+I[2, 100, 15, 20221208]",
+ "+I[2, 100, 15, 20221209]",
+ "+I[2, 100, 16, 20221208]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testCopyFilesNameAndSize() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.INPUT.toString());
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ });
+
+ /** write data to source table */
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(1, 100, 15, BinaryString.fromString("20221201")),
+ rowData(2, 40000, 16, BinaryString.fromString("20221208")),
+ rowData(4, 100, 19, BinaryString.fromString("20221209"))),
+ 1);
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(24, 898, 1, BinaryString.fromString("20221209")),
+ rowData(12, 312, 90, BinaryString.fromString("20221208")),
+ rowData(29, 434, 15, BinaryString.fromString("20221209"))),
+ 2);
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(6);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(6);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221201]",
+ "+I[12, 312, 90, 20221208]",
+ "+I[2, 40000, 16, 20221208]",
+ "+I[24, 898, 1, 20221209]",
+ "+I[29, 434, 15, 20221209]",
+ "+I[4, 100, 19, 20221209]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+
+ /**
+ * check file name and file size of manifest && index && data && schema && changelog &&
+ * snapshot files
+ */
+ List<Path> targetTableFiles = PickFilesUtil.getUsedFilesForLatestSnapshot(targetTable);
+ List<Pair<Path, Path>> filesPathInfoList =
+ targetTableFiles.stream()
+ .map(
+ absolutePath ->
+ Pair.of(
+ absolutePath,
+ getPathExcludeTableRoot(
+ absolutePath, targetTable.location())))
+ .collect(Collectors.toList());
+
+ Path tableLocation = table.location();
+ for (Pair<Path, Path> filesPathInfo : filesPathInfoList) {
+ Path sourceTableFile = new Path(tableLocation.toString() + filesPathInfo.getRight());
+ Assertions.assertTrue(table.fileIO().exists(sourceTableFile));
+ Assertions.assertEquals(
+ table.fileIO().getFileSize(sourceTableFile),
+ targetTable.fileIO().getFileSize(filesPathInfo.getLeft()));
+ }
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithSlowExpire() throws Exception {
+ /** prepare source table */
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+ }
+ });
+
+ /** write data into source table and commit them */
+ Thread writeDataToSourceTablethread =
+ new Thread(
+ () -> {
+ try {
+ /** write and commit data to source table. */
+ while (true) {
+ writeData(
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")));
+ Thread.sleep(30000);
+ }
+ } catch (Exception ignored) {
+ }
+ });
+ writeDataToSourceTablethread.start();
+
+ /** do clone */
+ int runCloneJobCount = 0;
+ boolean cloneJobHasException = false;
+ do {
+ try {
+ ++runCloneJobCount;
+ runCloneLatestSnapshotAction();
+ cloneJobHasException = false;
+ } catch (Exception e) {
+ cloneJobHasException = true;
+ }
+ } while (cloneJobHasException);
+
+ Assertions.assertEquals(1, runCloneJobCount);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithFastExpire() throws Exception {
+ /** prepare source table */
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+ }
+ });
+
+ /** write data into source table and commit them */
+ Thread writeDataToSourceTablethread =
+ new Thread(
+ () -> {
+ try {
+ /** write and commit data to source table every 100 ms. */
+ while (true) {
+ writeData(
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")));
+ }
+ } catch (Exception ignored) {
+ }
+ });
+ writeDataToSourceTablethread.start();
+
+ /** do clone */
+ int runCloneJobCount = 0;
+ boolean cloneJobHasException = false;
+ do {
+ try {
+ ++runCloneJobCount;
+ runCloneLatestSnapshotAction();
+ cloneJobHasException = false;
+ } catch (Exception e) {
+ cloneJobHasException = true;
+ }
+ } while (cloneJobHasException);
+
+ Assertions.assertTrue(runCloneJobCount > 1);
Review Comment:
Why do you check this? Why don't you check clone result?
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionITCaseBase.java:
##########
@@ -136,14 +130,12 @@ protected void writeData(GenericRow... data) throws Exception {
incrementalIdentifier++;
}
- protected List<String> getResult(TableRead read, List<Split> splits, RowType rowType)
- throws Exception {
- try (RecordReader<InternalRow> recordReader = read.createReader(splits)) {
- List<String> result = new ArrayList<>();
- recordReader.forEachRemaining(
- row -> result.add(DataFormatTestUtil.internalRowToString(row, rowType)));
- return result;
+ protected void writeDataToDynamicBucket(List<GenericRow> data, int bucket) throws Exception {
Review Comment:
```suggestion
protected void writeDataToBucket(List<GenericRow> data, int bucket) throws Exception {
```
This is only used in `CloneProcedureITCase`. Move it into that specific class.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,400 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.flink.clone.PickFilesUtil;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.emptyMap());
+
+ /** write data to source table */
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString("20221208")),
+ rowData(2, 100, 16, BinaryString.fromString("20221208")),
+ rowData(2, 100, 15, BinaryString.fromString("20221209")));
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(3);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(3);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]",
+ "+I[2, 100, 15, 20221208]",
+ "+I[2, 100, 15, 20221209]",
+ "+I[2, 100, 16, 20221208]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testCopyFilesNameAndSize() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.INPUT.toString());
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ });
+
+ /** write data to source table */
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(1, 100, 15, BinaryString.fromString("20221201")),
+ rowData(2, 40000, 16, BinaryString.fromString("20221208")),
+ rowData(4, 100, 19, BinaryString.fromString("20221209"))),
+ 1);
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(24, 898, 1, BinaryString.fromString("20221209")),
+ rowData(12, 312, 90, BinaryString.fromString("20221208")),
+ rowData(29, 434, 15, BinaryString.fromString("20221209"))),
+ 2);
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(6);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(6);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221201]",
+ "+I[12, 312, 90, 20221208]",
+ "+I[2, 40000, 16, 20221208]",
+ "+I[24, 898, 1, 20221209]",
+ "+I[29, 434, 15, 20221209]",
+ "+I[4, 100, 19, 20221209]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+
+ /**
+ * check file name and file size of manifest && index && data && schema && changelog &&
+ * snapshot files
+ */
+ List<Path> targetTableFiles = PickFilesUtil.getUsedFilesForLatestSnapshot(targetTable);
+ List<Pair<Path, Path>> filesPathInfoList =
+ targetTableFiles.stream()
+ .map(
+ absolutePath ->
+ Pair.of(
+ absolutePath,
+ getPathExcludeTableRoot(
+ absolutePath, targetTable.location())))
+ .collect(Collectors.toList());
+
+ Path tableLocation = table.location();
+ for (Pair<Path, Path> filesPathInfo : filesPathInfoList) {
+ Path sourceTableFile = new Path(tableLocation.toString() + filesPathInfo.getRight());
+ Assertions.assertTrue(table.fileIO().exists(sourceTableFile));
+ Assertions.assertEquals(
+ table.fileIO().getFileSize(sourceTableFile),
+ targetTable.fileIO().getFileSize(filesPathInfo.getLeft()));
+ }
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithSlowExpire() throws Exception {
Review Comment:
Slow expire is safe. No need for such case.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.procedure;
+
+import org.apache.paimon.flink.action.CloneAction;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.apache.paimon.utils.ParameterUtils.parseCommaSeparatedKeyValues;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/**
+ * Clone snapshot/tag/table procedure. Usage:
+ *
+ * <pre><code>
+ * CALL sys.clone('warehouse', 'database', 'table', 'catalog_config', 'target_warehouse', 'target_database',
+ * 'target_table', 'target_catalog_config', 'parallelism')
+ * </code></pre>
+ */
+public class CloneProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "clone";
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String warehouse,
+ String database,
+ String table,
+ String catalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTable,
+ String targetCatalogConfig,
+ String parallelism)
Review Comment:
How can user clone the whole warehouse or the whole database? Consider creating multiple `call` methods with different parameters.
--
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] support clone latest snapshot [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper merged PR #3287:
URL: https://github.com/apache/paimon/pull/3287
--
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] support clone latest snapshot [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1607883047
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionITCaseBase.java:
##########
@@ -136,14 +130,12 @@ protected void writeData(GenericRow... data) throws Exception {
incrementalIdentifier++;
}
- protected List<String> getResult(TableRead read, List<Split> splits, RowType rowType)
- throws Exception {
- try (RecordReader<InternalRow> recordReader = read.createReader(splits)) {
- List<String> result = new ArrayList<>();
- recordReader.forEachRemaining(
- row -> result.add(DataFormatTestUtil.internalRowToString(row, rowType)));
- return result;
+ protected void writeDataToDynamicBucket(List<GenericRow> data, int bucket) throws Exception {
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,400 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.flink.clone.PickFilesUtil;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.emptyMap());
+
+ /** write data to source table */
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString("20221208")),
+ rowData(2, 100, 16, BinaryString.fromString("20221208")),
+ rowData(2, 100, 15, BinaryString.fromString("20221209")));
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(3);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(3);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]",
+ "+I[2, 100, 15, 20221208]",
+ "+I[2, 100, 15, 20221209]",
+ "+I[2, 100, 16, 20221208]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testCopyFilesNameAndSize() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.INPUT.toString());
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ });
+
+ /** write data to source table */
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(1, 100, 15, BinaryString.fromString("20221201")),
+ rowData(2, 40000, 16, BinaryString.fromString("20221208")),
+ rowData(4, 100, 19, BinaryString.fromString("20221209"))),
+ 1);
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(24, 898, 1, BinaryString.fromString("20221209")),
+ rowData(12, 312, 90, BinaryString.fromString("20221208")),
+ rowData(29, 434, 15, BinaryString.fromString("20221209"))),
+ 2);
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(6);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(6);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221201]",
+ "+I[12, 312, 90, 20221208]",
+ "+I[2, 40000, 16, 20221208]",
+ "+I[24, 898, 1, 20221209]",
+ "+I[29, 434, 15, 20221209]",
+ "+I[4, 100, 19, 20221209]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+
+ /**
+ * check file name and file size of manifest && index && data && schema && changelog &&
+ * snapshot files
+ */
+ List<Path> targetTableFiles = PickFilesUtil.getUsedFilesForLatestSnapshot(targetTable);
+ List<Pair<Path, Path>> filesPathInfoList =
+ targetTableFiles.stream()
+ .map(
+ absolutePath ->
+ Pair.of(
+ absolutePath,
+ getPathExcludeTableRoot(
+ absolutePath, targetTable.location())))
+ .collect(Collectors.toList());
+
+ Path tableLocation = table.location();
+ for (Pair<Path, Path> filesPathInfo : filesPathInfoList) {
+ Path sourceTableFile = new Path(tableLocation.toString() + filesPathInfo.getRight());
+ Assertions.assertTrue(table.fileIO().exists(sourceTableFile));
+ Assertions.assertEquals(
+ table.fileIO().getFileSize(sourceTableFile),
+ targetTable.fileIO().getFileSize(filesPathInfo.getLeft()));
+ }
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithSlowExpire() throws Exception {
Review Comment:
done
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,400 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.flink.clone.PickFilesUtil;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.emptyMap());
+
+ /** write data to source table */
+ writeData(
+ rowData(1, 100, 15, BinaryString.fromString("20221208")),
+ rowData(1, 100, 16, BinaryString.fromString("20221208")),
+ rowData(1, 100, 15, BinaryString.fromString("20221209")));
+ writeData(
+ rowData(2, 100, 15, BinaryString.fromString("20221208")),
+ rowData(2, 100, 16, BinaryString.fromString("20221208")),
+ rowData(2, 100, 15, BinaryString.fromString("20221209")));
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(3);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(3);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221208]",
+ "+I[1, 100, 15, 20221209]",
+ "+I[1, 100, 16, 20221208]",
+ "+I[2, 100, 15, 20221208]",
+ "+I[2, 100, 15, 20221209]",
+ "+I[2, 100, 16, 20221208]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testCopyFilesNameAndSize() throws Exception {
+ /** prepare source table */
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.INPUT.toString());
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ });
+
+ /** write data to source table */
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(1, 100, 15, BinaryString.fromString("20221201")),
+ rowData(2, 40000, 16, BinaryString.fromString("20221208")),
+ rowData(4, 100, 19, BinaryString.fromString("20221209"))),
+ 1);
+ writeDataToDynamicBucket(
+ Arrays.asList(
+ rowData(24, 898, 1, BinaryString.fromString("20221209")),
+ rowData(12, 312, 90, BinaryString.fromString("20221208")),
+ rowData(29, 434, 15, BinaryString.fromString("20221209"))),
+ 2);
+ checkLatestSnapshot(table, 2, Snapshot.CommitKind.APPEND);
+
+ /** do clone */
+ runCloneLatestSnapshotAction();
+ FileStoreTable targetTable = getFileStoreTable(targetTableName);
+ checkLatestSnapshot(targetTable, 2, Snapshot.CommitKind.APPEND);
+
+ /** check scan result */
+ List<DataSplit> splits1 = table.newSnapshotReader().read().dataSplits();
+ assertThat(splits1.size()).isEqualTo(6);
+ List<DataSplit> splits2 = targetTable.newSnapshotReader().read().dataSplits();
+ assertThat(splits2.size()).isEqualTo(6);
+
+ TableScan sourceTableScan = table.newReadBuilder().newScan();
+ TableScan targetTableScan = targetTable.newReadBuilder().newScan();
+
+ List<String> scanResult =
+ Arrays.asList(
+ "+I[1, 100, 15, 20221201]",
+ "+I[12, 312, 90, 20221208]",
+ "+I[2, 40000, 16, 20221208]",
+ "+I[24, 898, 1, 20221209]",
+ "+I[29, 434, 15, 20221209]",
+ "+I[4, 100, 19, 20221209]");
+ validateResult(table, ROW_TYPE, sourceTableScan, scanResult, 60_000);
+ validateResult(targetTable, ROW_TYPE, targetTableScan, scanResult, 60_000);
+
+ /**
+ * check file name and file size of manifest && index && data && schema && changelog &&
+ * snapshot files
+ */
+ List<Path> targetTableFiles = PickFilesUtil.getUsedFilesForLatestSnapshot(targetTable);
+ List<Pair<Path, Path>> filesPathInfoList =
+ targetTableFiles.stream()
+ .map(
+ absolutePath ->
+ Pair.of(
+ absolutePath,
+ getPathExcludeTableRoot(
+ absolutePath, targetTable.location())))
+ .collect(Collectors.toList());
+
+ Path tableLocation = table.location();
+ for (Pair<Path, Path> filesPathInfo : filesPathInfoList) {
+ Path sourceTableFile = new Path(tableLocation.toString() + filesPathInfo.getRight());
+ Assertions.assertTrue(table.fileIO().exists(sourceTableFile));
+ Assertions.assertEquals(
+ table.fileIO().getFileSize(sourceTableFile),
+ targetTable.fileIO().getFileSize(filesPathInfo.getLeft()));
+ }
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithSlowExpire() throws Exception {
+ /** prepare source table */
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+ }
+ });
+
+ /** write data into source table and commit them */
+ Thread writeDataToSourceTablethread =
+ new Thread(
+ () -> {
+ try {
+ /** write and commit data to source table. */
+ while (true) {
+ writeData(
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")));
+ Thread.sleep(30000);
+ }
+ } catch (Exception ignored) {
+ }
+ });
+ writeDataToSourceTablethread.start();
+
+ /** do clone */
+ int runCloneJobCount = 0;
+ boolean cloneJobHasException = false;
+ do {
+ try {
+ ++runCloneJobCount;
+ runCloneLatestSnapshotAction();
+ cloneJobHasException = false;
+ } catch (Exception e) {
+ cloneJobHasException = true;
+ }
+ } while (cloneJobHasException);
+
+ Assertions.assertEquals(1, runCloneJobCount);
+ }
+
+ @Test
+ @Timeout(60_000)
+ public void testSnapshotWithFastExpire() throws Exception {
+ /** prepare source table */
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+ put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+ }
+ });
+
+ /** write data into source table and commit them */
+ Thread writeDataToSourceTablethread =
+ new Thread(
+ () -> {
+ try {
+ /** write and commit data to source table every 100 ms. */
+ while (true) {
+ writeData(
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 1,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 16,
+ BinaryString.fromString("20221208")),
+ rowData(
+ 2,
+ 100,
+ 15,
+ BinaryString.fromString("20221209")));
+ }
+ } catch (Exception ignored) {
+ }
+ });
+ writeDataToSourceTablethread.start();
+
+ /** do clone */
+ int runCloneJobCount = 0;
+ boolean cloneJobHasException = false;
+ do {
+ try {
+ ++runCloneJobCount;
+ runCloneLatestSnapshotAction();
+ cloneJobHasException = false;
+ } catch (Exception e) {
+ cloneJobHasException = true;
+ }
+ } while (cloneJobHasException);
+
+ Assertions.assertTrue(runCloneJobCount > 1);
Review Comment:
done
--
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] support clone latest snapshot [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #3287:
URL: https://github.com/apache/paimon/pull/3287#discussion_r1601004479
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.clone.CloneFileInfo;
+import org.apache.paimon.flink.clone.CloneSourceBuilder;
+import org.apache.paimon.flink.clone.CopyFileOperator;
+import org.apache.paimon.flink.clone.PickFilesForCloneOperator;
+import org.apache.paimon.flink.clone.SnapshotHintChannelComputer;
+import org.apache.paimon.flink.clone.SnapshotHintOperator;
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.options.CatalogOptions;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/** Snapshot/Tag/Table clone action for Flink. */
+public class CloneAction extends ActionBase {
+
+ private final int parallelism;
+
+ private Map<String, String> sourceCatalogConfig = new HashMap<>();
+ private final String database;
+ private final String tableName;
+
+ private Map<String, String> targetCatalogConfig = new HashMap<>();
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneAction(
+ String warehouse,
+ String database,
+ String tableName,
+ Map<String, String> sourceCatalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTableName,
+ Map<String, String> targetCatalogConfig,
+ String parallelismStr) {
+ super(warehouse, sourceCatalogConfig);
+
+ checkNotNull(warehouse, "warehouse must not be null.");
+ checkNotNull(targetWarehouse, "targetWarehouse must not be null.");
+
+ this.parallelism =
+ isBlank(parallelismStr) ? env.getParallelism() : Integer.parseInt(parallelismStr);
+
+ if (!sourceCatalogConfig.isEmpty()) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ }
Review Comment:
```suggestion
this.sourceCatalogConfig = sourceCatalogConfig;
```
##########
paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java:
##########
@@ -499,8 +499,7 @@ private void deleteFileOrDirQuietly(Path path) {
/** A helper functional interface for method {@link #retryReadingFiles}. */
@FunctionalInterface
- private interface ReaderWithIOException<T> {
-
+ interface ReaderWithIOException<T> {
Review Comment:
Remove useless changes.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesUtil.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
Review Comment:
Write meaningful comments please.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.clone.CloneFileInfo;
+import org.apache.paimon.flink.clone.CloneSourceBuilder;
+import org.apache.paimon.flink.clone.CopyFileOperator;
+import org.apache.paimon.flink.clone.PickFilesForCloneOperator;
+import org.apache.paimon.flink.clone.SnapshotHintChannelComputer;
+import org.apache.paimon.flink.clone.SnapshotHintOperator;
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.options.CatalogOptions;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.StringUtils.isBlank;
+
+/** Snapshot/Tag/Table clone action for Flink. */
+public class CloneAction extends ActionBase {
+
+ private final int parallelism;
+
+ private Map<String, String> sourceCatalogConfig = new HashMap<>();
+ private final String database;
+ private final String tableName;
+
+ private Map<String, String> targetCatalogConfig = new HashMap<>();
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneAction(
+ String warehouse,
+ String database,
+ String tableName,
+ Map<String, String> sourceCatalogConfig,
+ String targetWarehouse,
+ String targetDatabase,
+ String targetTableName,
+ Map<String, String> targetCatalogConfig,
+ String parallelismStr) {
+ super(warehouse, sourceCatalogConfig);
+
+ checkNotNull(warehouse, "warehouse must not be null.");
+ checkNotNull(targetWarehouse, "targetWarehouse must not be null.");
+
+ this.parallelism =
+ isBlank(parallelismStr) ? env.getParallelism() : Integer.parseInt(parallelismStr);
+
+ if (!sourceCatalogConfig.isEmpty()) {
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ }
+ this.sourceCatalogConfig.put(CatalogOptions.WAREHOUSE.key(), warehouse);
+ this.database = database;
+ this.tableName = tableName;
+
+ if (!targetCatalogConfig.isEmpty()) {
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
Review Comment:
```suggestion
this.targetCatalogConfig = targetCatalogConfig;
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/SnapshotHintChannelComputer.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.table.sink.ChannelComputer;
+
+/** SnapshotHintChannelComputer. */
Review Comment:
Write meaningful comments please.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Pick the tables to be cloned based on the user input parameters. The record type of the build
+ * DataStream is Tuple2. The left element is the identifier of source table and the right element is
Review Comment:
```suggestion
* DataStream is {@link Tuple2}. The left element is the identifier of source table and the right element is
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/SnapshotHintOperator.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+
+/**
+ * When the files copy finished of a table, then create snapshot hint, it means that this table can
+ * be used now.
Review Comment:
```suggestion
* Creates snapshot hint files after copying a table.
```
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CloneProcedureITCase.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.flink.procedure;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT Case for {@link CloneProcedure}. */
+public class CloneProcedureITCase extends CatalogITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ @Test
+ public void testCloneLatestSnapshot() throws Exception {
+ sql(
+ "CREATE TABLE T ("
+ + " k INT,"
+ + " v INT,"
+ + " hh INT,"
+ + " dt STRING,"
+ + " PRIMARY KEY (k, dt, hh) NOT ENFORCED"
+ + ") PARTITIONED BY (dt, hh) WITH ("
+ + " 'write-only' = 'true',"
Review Comment:
Why write only?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CopyFileOperator.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.AbstractCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.IOUtils;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.Map;
+
+/** A Operator to copy files. */
+public class CopyFileOperator extends AbstractStreamOperator<CloneFileInfo>
+ implements OneInputStreamOperator<CloneFileInfo, CloneFileInfo> {
+
+ private final Map<String, String> sourceCatalogConfig;
+ private final Map<String, String> targetCatalogConfig;
+
+ private AbstractCatalog sourceCatalog;
+ private AbstractCatalog targetCatalog;
+
+ public CopyFileOperator(
+ Map<String, String> sourceCatalogConfig, Map<String, String> targetCatalogConfig) {
+ super();
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.targetCatalogConfig = targetCatalogConfig;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ sourceCatalog =
+ (AbstractCatalog)
+ FlinkCatalogFactory.createPaimonCatalog(
+ Options.fromMap(sourceCatalogConfig));
+ targetCatalog =
+ (AbstractCatalog)
+ FlinkCatalogFactory.createPaimonCatalog(
+ Options.fromMap(targetCatalogConfig));
+ }
+
+ @Override
+ public void processElement(StreamRecord<CloneFileInfo> streamRecord) throws Exception {
+ CloneFileInfo cloneFileInfo = streamRecord.getValue();
+
+ FileIO sourceTableFileIO = sourceCatalog.fileIO();
+ FileIO targetTableFileIO = targetCatalog.fileIO();
+ Path sourceTableRootPath =
+ sourceCatalog.getDataTableLocation(
+ Identifier.fromString(cloneFileInfo.sourceIdentifier()));
+ Path targetTableRootPath =
+ targetCatalog.getDataTableLocation(
+ Identifier.fromString(cloneFileInfo.targetIdentifier()));
+
+ Path filePathExcludeTableRoot = cloneFileInfo.getFilePathExcludeTableRoot();
+ Path sourcePath = new Path(sourceTableRootPath.toString() + filePathExcludeTableRoot);
+ Path targetPath = new Path(targetTableRootPath.toString() + filePathExcludeTableRoot);
+
+ if (targetTableFileIO.exists(targetPath)
+ && targetTableFileIO.getFileSize(targetPath)
+ == sourceTableFileIO.getFileSize(sourcePath)) {
+ LOG.info(
+ "Skipping target file {} because it already exists and has the same size.",
+ targetPath);
+ return;
Review Comment:
Consider copying the whole database. There are two tables in this database. All files of table A has been copied, so no record about table A will be sent to `SnapshotHintOperator`. Snapshot hint files of table A will not be created.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesUtil.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(FileStoreTable table) {
Review Comment:
```suggestion
public static List<Path> getUsedFilesForLatestSnapshot(FileStoreTable table) {
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneSourceBuilder.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Pick the tables to be cloned based on the user input parameters. The record type of the build
+ * DataStream is Tuple2. The left element is the identifier of source table and the right element is
+ * the identifier of target table.
+ */
+public class CloneSourceBuilder {
+
+ private final StreamExecutionEnvironment env;
+ private final Map<String, String> sourceCatalogConfig;
+ private final String database;
+ private final String tableName;
+ private final String targetDatabase;
+ private final String targetTableName;
+
+ public CloneSourceBuilder(
+ StreamExecutionEnvironment env,
+ Map<String, String> sourceCatalogConfig,
+ String database,
+ String tableName,
+ String targetDatabase,
+ String targetTableName) {
+ this.env = env;
+ this.sourceCatalogConfig = sourceCatalogConfig;
+ this.database = database;
+ this.tableName = tableName;
+ this.targetDatabase = targetDatabase;
+ this.targetTableName = targetTableName;
+ }
+
+ public DataStream<Tuple2<String, String>> build() {
+ Catalog sourceCatalog =
+ FlinkCatalogFactory.createPaimonCatalog(Options.fromMap(sourceCatalogConfig));
+
+ List<Tuple2<String, String>> result = new ArrayList<>();
+
+ if (database == null) {
+ checkArgument(tableName == null, "tableName must be null when database is null.");
+ for (String db : sourceCatalog.listDatabases()) {
+ try {
+ for (String table : sourceCatalog.listTables(db)) {
+ String s = db + "." + table;
+ result.add(new Tuple2<>(s, s));
+ }
+ } catch (Exception e) {
+ // ignore
+ }
Review Comment:
Don't igonre exceptions. Throw it out.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CloneActionITCase.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CloneAction}. */
+public class CloneActionITCase extends ActionITCaseBase {
+
+ private static final DataType[] FIELD_TYPES =
+ new DataType[] {DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING()};
+
+ private static final RowType ROW_TYPE =
+ RowType.of(FIELD_TYPES, new String[] {"k", "v", "hh", "dt"});
+
+ private final String targetTableName = "copy_table";
+
+ @Test
+ @Timeout(60_000)
+ public void testCloneLatestSnapshot() throws Exception {
+ FileStoreTable table =
+ prepareTable(
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "true"));
Review Comment:
Why write only?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/PickFilesUtil.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.clone;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** PickFilesUtil. */
+public class PickFilesUtil {
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ public static List<Path> getUsedFilesForSnapshot(FileStoreTable table) {
+ FileStore<?> store = table.store();
+ SnapshotManager snapshotManager = store.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ ManifestList manifestList = store.manifestListFactory().create();
+ SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location());
+ IndexFileHandler indexFileHandler = store.newIndexFileHandler();
+
+ List<Path> files = new ArrayList<>();
+ if (snapshot != null) {
+ files.add(snapshotManager.snapshotPath(snapshot.id()));
+ files.addAll(
+ getUsedFilesInternal(
+ snapshot,
+ store.pathFactory(),
+ store.newScan(),
+ manifestList,
+ indexFileHandler));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ files.add(schemaManager.toSchemaPath(id));
+ }
+ return files;
+ }
+
+ private static List<Path> getUsedFilesInternal(
+ Snapshot snapshot,
+ FileStorePathFactory pathFactory,
+ FileStoreScan scan,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler) {
+ List<Path> files = new ArrayList<>();
+ addManifestList(files, snapshot, pathFactory);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(
+ () -> readAllManifestsWithIOException(snapshot, manifestList));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(
+ manifestFileName.stream()
+ .map(pathFactory::toManifestFilePath)
+ .collect(Collectors.toList()));
+
+ // try to read data files
+ List<Path> dataFiles = new ArrayList<>();
+ List<SimpleFileEntry> simpleFileEntries =
+ scan.withSnapshot(snapshot).readSimpleEntries();
+ for (SimpleFileEntry simpleFileEntry : simpleFileEntries) {
+ Path dataFilePath =
+ pathFactory
+ .createDataFilePathFactory(
+ simpleFileEntry.partition(), simpleFileEntry.bucket())
+ .toPath(simpleFileEntry.fileName());
+ dataFiles.add(dataFilePath);
+ }
+ // In order to make files created earlier perform copying earlier.
+ Collections.reverse(dataFiles);
Review Comment:
When scanning, `dataFiles` are listed from older to newer. By reversing `dataFiles`, newer files will be copied first. Is this better or worse? Did you perform some experiences? What's the result?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 CloneAction}. */
+public class CloneActionFactory implements ActionFactory {
+
+ private static final String IDENTIFIER = "clone";
+ private static final String PARALLELISM = "parallelism";
+ private static final String TARGET_WAREHOUSE = "target_warehouse";
+ private static final String TARGET_DATABASE = "target_database";
+ private static final String TARGET_TABLE = "target_table";
+ private static final String TARGET_CATALOG_CONF = "target_catalog_conf";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterToolAdapter params) {
+ CloneAction cloneAction =
+ new CloneAction(
+ params.get(WAREHOUSE),
+ params.get(DATABASE),
+ params.get(TABLE),
+ optionalConfigMap(params, CATALOG_CONF),
+ params.get(TARGET_WAREHOUSE),
+ params.get(TARGET_DATABASE),
+ params.get(TARGET_TABLE),
+ optionalConfigMap(params, TARGET_CATALOG_CONF),
+ params.get(PARALLELISM));
+
+ return Optional.of(cloneAction);
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println("Action \"clone\" runs a batch job for clone the latest Snapshot.");
+ System.out.println();
+
+ System.out.println("Syntax:");
+ System.out.println(
+ " clone --warehouse <warehouse_path>"
+ + "[--database <database_name>]"
+ + "[--table <table_name>]"
+ + "[--catalog_conf <source-paimon-catalog-conf> [--catalog_conf <source-paimon-catalog-conf> ...]]"
+ + "--target_warehouse <target_warehouse_path>"
+ + "--target_database <target_database_name> "
+ + "--target_table <target_table_name> "
Review Comment:
`target_database` and `target_table` are also optional.
--
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