You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JingsongLi (via GitHub)" <gi...@apache.org> on 2026/10/08 04:44:57 UTC
[PR] [spark][flink] Add metadata-only external file import procedure [paimon]
JingsongLi opened a new pull request, #10423:
URL: https://github.com/apache/paimon/pull/10423
### Purpose
Register files from an absolute directory in a Paimon partition while their contents are unavailable, for example when the files are archived. Add `CALL sys.import_files(table, location, partition)` for Spark and Flink, backed by a shared metadata-only importer. There is no format argument: the target table's `file.format` selects matching file suffixes.
The importer only lists directory entries and file sizes. It records original locations in `DataFileMeta.externalPath`, writes unavailable row counts (`-1`) and no column statistics, and appends all selected files in one commit. It never opens, copies, renames, or restores source files. UUID manifest identities allow the same source basename in different directories. Hidden files and subdirectories are skipped; an empty selection does not create a snapshot. Paths already referenced by the current snapshot are rejected; concurrent imports of the same directory must be avoided.
Partitioned tables require the complete typed partition specification; unpartitioned tables omit it. Source files must match the target's format and non-partition columns, and must eventually become readable for queries. Registered external files participate in the normal Paimon file lifecycle, including deletion after compaction or partition removal and snapshot expiration.
Supported targets are bucket-unaware append tables without row tracking, data evolution, or Iceberg metadata synchronization, using a filesystem/renaming snapshot commit backend (filesystem/Hive catalogs). Catalog-managed commits, including REST, reject unavailable counts.
Depends on #10422 (https://github.com/apache/paimon/pull/10422); its base is that branch so the diff contains only the importer, engine procedures, tests, and procedure documentation.
### Tests
Core importer tests use invalid source bytes and a FileIO spy that throws if any archived source file is opened. They verify metadata, typed partitions, filtering, duplicate rejection without partial commit, colliding basenames, mixed native/imported files, empty directories, and unsupported tables. Spark and Flink SQL tests register readable files and validate actual query results; Spark covers Parquet, ORC, Avro, cache refresh, nullable schema evolution, LIMIT, and ANALYZE.
Validated with JDK 8, Spark 3, and Flink 1. Final Maven verification runs without `fast-build`, including the foundation regression suites and configured formatting/build checks:
```sh
mvn -pl paimon-core,paimon-flink/paimon-flink-common,paimon-spark/paimon-spark-ut -am \
-Pflink1,spark3 -DfailIfNoTests=false \
-DwildcardSuites=org.apache.paimon.spark.sql.UnknownRowCountTest,org.apache.paimon.spark.sql.PushDownAggregatesTest,org.apache.paimon.spark.procedure.ImportFilesProcedureTest \
'-Dtest=ExternalFileImporterTest,UnknownRowCountTest,DataSplitCompatibleTest,FileIndexEvaluatorTest,PartitionStatisticsReporterTest,RemoveUnexistingManifestsTest,SnapshotTest,CatalogSnapshotCommitTest,RenamingSnapshotCommitTest,SplitWeightUtilsTest,FlinkTableSourceStatisticsTest,ImportFilesProcedureITCase,ProcedureTest,AppendOnlySimpleTableTest#testLimitPushDown' test
```
Result: **81 tests passed** (55 core, 9 Flink, 17 Spark).
--
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] [spark][flink] Add import_files procedure with external paths [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #10423:
URL: https://github.com/apache/paimon/pull/10423#discussion_r4216039018
##########
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:
Per discussion with the PR author, snapshot-expiration behavior and re-registering source paths retained in historical snapshots are outside the requested scope of this change. Treating this as non-blocking and resolving the thread as an explicit scope decision, not as a claim that the reproduced case has been fixed.
##########
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:
Per discussion, scan.mode is intended as a dynamic query parameter, and non-default read scan modes on import targets are outside the expected usage here. Treating this as non-blocking and resolving the thread as an explicit scope decision, not as a code fix.
--
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] [spark][flink] Add import_files procedure with external paths [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark][flink] Add metadata-only external file import procedure [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi closed pull request #10423: [spark][flink] Add metadata-only external file import procedure
URL: https://github.com/apache/paimon/pull/10423
--
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] [spark][flink] Add import_files procedure with external paths [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #10423:
URL: https://github.com/apache/paimon/pull/10423
--
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