leaves12138 commented on code in PR #10423: URL: https://github.com/apache/paimon/pull/10423#discussion_r4216009368
########## paimon-core/src/main/java/org/apache/paimon/migrate/ExternalFileImporter.java: ########## @@ -0,0 +1,148 @@ +/* + * 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.migrate; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryWriter; +import org.apache.paimon.format.SimpleStatsExtractor; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.StringUtils; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.apache.paimon.utils.ParameterUtils.parseCommaSeparatedKeyValues; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Imports existing files by recording their external paths in a single append commit. */ +public class ExternalFileImporter { + + private ExternalFileImporter() {} + + public static long importFiles(Table table, String location, @Nullable String partition) + throws Exception { + checkArgument(table instanceof FileStoreTable, "import_files requires a FileStoreTable."); + FileStoreTable fileStoreTable = (FileStoreTable) table; + checkArgument( + table.primaryKeys().isEmpty() && fileStoreTable.coreOptions().bucket() == -1, + "import_files only supports append tables with bucket = -1."); + checkArgument( + !fileStoreTable.coreOptions().rowTrackingEnabled() + && !fileStoreTable.coreOptions().dataEvolutionEnabled(), + "import_files does not support row tracking or data-evolution tables."); + + Map<String, String> partitionSpec = parseCommaSeparatedKeyValues(partition); + checkArgument( + partitionSpec.keySet().equals(new HashSet<>(table.partitionKeys())), + "Partition must specify exactly the table partition keys: %s.", + table.partitionKeys()); + RowType partitionType = fileStoreTable.store().partitionType(); + BinaryRow partitionRow = + FileMetaUtils.writePartitionValue( + partitionType, + table.partitionKeys().stream() + .map(partitionSpec::get) + .collect(Collectors.toList()), + partitionType.getFields().stream() + .map(f -> BinaryWriter.createValueSetter(f.type())) + .collect(Collectors.toList()), + fileStoreTable.coreOptions().partitionDefaultName()); + + String fileFormat = fileStoreTable.coreOptions().fileFormatString(); + checkArgument(!StringUtils.isNullOrWhitespaceOnly(location), "Location must not be empty."); + Path directory = new Path(location); + checkArgument( + directory.toUri().getPath().startsWith("/"), "Location must be an absolute path."); + FileIO fileIO = table.fileIO(); + checkArgument(fileIO.getFileStatus(directory).isDir(), "Location must be a directory."); + + Map<String, FileStatus> files = new LinkedHashMap<>(); + for (FileStatus status : fileIO.listStatus(directory)) { + String name = status.getPath().getName(); + if (!status.isDir() + && !name.startsWith(".") + && !name.startsWith("_") + && name.endsWith("." + fileFormat)) { + files.putIfAbsent(status.getPath().toString(), status); + } + } + if (files.isEmpty()) { + return 0; + } + + Set<String> existingPaths = new HashSet<>(); + for (Split split : fileStoreTable.newScan().plan().splits()) { + for (DataFileMeta file : ((DataSplit) split).dataFiles()) { + file.externalPath().ifPresent(existingPaths::add); + } + } + for (String externalPath : files.keySet()) { + checkArgument( + !existingPaths.contains(externalPath), + "File has already been imported: %s.", + externalPath); Review Comment: [P1] Prevent re-registering a path still owned by retained snapshot history Checking only the current snapshot allows the same physical file to acquire a new manifest identity while its previous identity is still scheduled for deletion. I reproduced this with an ordinary bucket-unaware append table and an unchanged Parquet source: 1. Import `first.parquet` (snapshot 1). 2. Call `BatchTableCommit.truncateTable()` (snapshot 2). The current table is empty, but the source file is still present. 3. Import the same directory again (snapshot 3); this succeeds and records the same external path under a new UUID. 4. Expire the first two snapshots with `snapshotRetainMin(1)`, `snapshotRetainMax(1)`, and `snapshotTimeRetain(Duration.ZERO)`. After expiration, both earliest and latest snapshot IDs are 3 and its file metadata still references `first.parquet`, but `fileIO.exists(imported)` is false. The DELETE entry from the older truncate snapshot deletes the physical path without protecting its new live reference. This is loss of data referenced by the latest snapshot, not merely the documented deletion of obsolete imported files. Please reject paths still owned by retained snapshot/deletion history before registration, or make the lifecycle handling safe for multiple manifest identities sharing a physical path. Add a truncate/re-import/expiration regression. ########## paimon-core/src/main/java/org/apache/paimon/migrate/ExternalFileImporter.java: ########## @@ -0,0 +1,148 @@ +/* + * 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.migrate; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryWriter; +import org.apache.paimon.format.SimpleStatsExtractor; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.StringUtils; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.apache.paimon.utils.ParameterUtils.parseCommaSeparatedKeyValues; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Imports existing files by recording their external paths in a single append commit. */ +public class ExternalFileImporter { + + private ExternalFileImporter() {} + + public static long importFiles(Table table, String location, @Nullable String partition) + throws Exception { + checkArgument(table instanceof FileStoreTable, "import_files requires a FileStoreTable."); + FileStoreTable fileStoreTable = (FileStoreTable) table; + checkArgument( + table.primaryKeys().isEmpty() && fileStoreTable.coreOptions().bucket() == -1, + "import_files only supports append tables with bucket = -1."); + checkArgument( + !fileStoreTable.coreOptions().rowTrackingEnabled() + && !fileStoreTable.coreOptions().dataEvolutionEnabled(), + "import_files does not support row tracking or data-evolution tables."); + + Map<String, String> partitionSpec = parseCommaSeparatedKeyValues(partition); + checkArgument( + partitionSpec.keySet().equals(new HashSet<>(table.partitionKeys())), + "Partition must specify exactly the table partition keys: %s.", + table.partitionKeys()); + RowType partitionType = fileStoreTable.store().partitionType(); + BinaryRow partitionRow = + FileMetaUtils.writePartitionValue( + partitionType, + table.partitionKeys().stream() + .map(partitionSpec::get) + .collect(Collectors.toList()), + partitionType.getFields().stream() + .map(f -> BinaryWriter.createValueSetter(f.type())) + .collect(Collectors.toList()), + fileStoreTable.coreOptions().partitionDefaultName()); + + String fileFormat = fileStoreTable.coreOptions().fileFormatString(); + checkArgument(!StringUtils.isNullOrWhitespaceOnly(location), "Location must not be empty."); + Path directory = new Path(location); + checkArgument( + directory.toUri().getPath().startsWith("/"), "Location must be an absolute path."); + FileIO fileIO = table.fileIO(); + checkArgument(fileIO.getFileStatus(directory).isDir(), "Location must be a directory."); + + Map<String, FileStatus> files = new LinkedHashMap<>(); + for (FileStatus status : fileIO.listStatus(directory)) { + String name = status.getPath().getName(); + if (!status.isDir() + && !name.startsWith(".") + && !name.startsWith("_") + && name.endsWith("." + fileFormat)) { + files.putIfAbsent(status.getPath().toString(), status); + } + } + if (files.isEmpty()) { + return 0; + } + + Set<String> existingPaths = new HashSet<>(); + for (Split split : fileStoreTable.newScan().plan().splits()) { + for (DataFileMeta file : ((DataSplit) split).dataFiles()) { + file.externalPath().ifPresent(existingPaths::add); Review Comment: [P2] Check duplicates against full latest-snapshot metadata, not the configured read scan `newScan().plan()` honors the target table's scan options, so it is not guaranteed to enumerate all current file references. A table created with the valid option `scan.mode=latest-delta` passes every target capability check here. Import a source file, commit one ordinary native append, then import the same directory again: the duplicate check sees only the native append's delta and the second import unexpectedly succeeds, registering the same external file under another UUID. The latest full snapshot still references the original import throughout. I reproduced this by creating the existing test helper's append table with `Collections.singletonMap("scan.mode", "latest-delta")`, importing once, calling `write(table, GenericRow.of(9, fromString("p1"), 1))`, and attempting the identical import; the expected `already been imported` exception is not thrown. No concurrent import or source modification is involved. Please collect existing paths from a full scan of the actual latest snapshot on the target branch, independent of query startup/time-travel options, and add a regression for a configured latest-delta target. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
