This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 012c99d6f7 [spark][flink] Add import_files procedure with external
paths (#10423)
012c99d6f7 is described below
commit 012c99d6f76e69cea076ddacb7209cfd1dc25717
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 15:29:19 2026 +0800
[spark][flink] Add import_files procedure with external paths (#10423)
---
docs/docs/flink/procedures.md | 2 +-
docs/docs/flink/procedures/table-operations.md | 41 +++
docs/docs/spark/procedures.md | 3 +-
docs/docs/spark/procedures/migration.md | 38 ++
.../paimon/migrate/ExternalFileImporter.java | 148 ++++++++
.../paimon/migrate/ExternalFileImporterTest.java | 386 +++++++++++++++++++++
.../flink/procedure/ImportFilesProcedure.java | 50 +++
.../services/org.apache.paimon.factories.Factory | 1 +
.../procedure/ImportFilesProcedureITCase.java | 102 ++++++
.../org/apache/paimon/spark/SparkProcedures.java | 2 +
.../spark/procedure/ImportFilesProcedure.java | 101 ++++++
.../spark/procedure/ImportFilesProcedureTest.scala | 111 ++++++
12 files changed, 983 insertions(+), 2 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index b248d8ab1c..883324e889 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -69,7 +69,7 @@ defaults are specific to the procedure.
| [Compaction and Layout](./procedures/compaction) |
[`compact`](./procedures/compaction#compact),
[`compact_database`](./procedures/compaction#compact_database),
[`compact_chain_table`](./procedures/compaction#compact_chain_table),
[`compact_manifest`](./procedures/compaction#compact_manifest),
[`rescale`](./procedures/compaction#rescale),
[`materialize_deletion_vectors`](./procedures/compaction#materialize_deletion_vectors),
[`reassign_row_id`](./procedures/compaction#reassign_row_id) |
| [Snapshots and Retention](./procedures/snapshots) |
[`expire_snapshots`](./procedures/snapshots#expire_snapshots),
[`expire_changelogs`](./procedures/snapshots#expire_changelogs),
[`expire_partitions`](./procedures/snapshots#expire_partitions),
[`rollback_to`](./procedures/snapshots#rollback_to),
[`rollback_to_as_latest`](./procedures/snapshots#rollback_to_as_latest),
[`rollback_to_timestamp`](./procedures/snapshots#rollback_to_timestamp),
[`rollback_to_watermark`](./procedures/snapsho [...]
| [Tags and Branches](./procedures/tags-and-branches) |
[`create_tag`](./procedures/tags-and-branches#create_tag),
[`create_tag_from_timestamp`](./procedures/tags-and-branches#create_tag_from_timestamp),
[`create_tag_from_watermark`](./procedures/tags-and-branches#create_tag_from_watermark),
[`replace_tag`](./procedures/tags-and-branches#replace_tag),
[`rename_tag`](./procedures/tags-and-branches#rename_tag),
[`delete_tag`](./procedures/tags-and-branches#delete_tag), [`expire_tags`](./pr
[...]
-| [Table Operations](./procedures/table-operations) |
[`merge_into`](./procedures/table-operations#merge_into),
[`data_evolution_merge_into`](./procedures/table-operations#data_evolution_merge_into),
[`migrate_database`](./procedures/table-operations#migrate_database),
[`migrate_table`](./procedures/table-operations#migrate_table),
[`clone`](./procedures/table-operations#clone),
[`copy_files`](./procedures/table-operations#copy_files),
[`alter_column_default_value`](./procedures/table-op [...]
+| [Table Operations](./procedures/table-operations) |
[`merge_into`](./procedures/table-operations#merge_into),
[`data_evolution_merge_into`](./procedures/table-operations#data_evolution_merge_into),
[`migrate_database`](./procedures/table-operations#migrate_database),
[`migrate_table`](./procedures/table-operations#migrate_table),
[`import_files`](./procedures/table-operations#import_files),
[`clone`](./procedures/table-operations#clone),
[`copy_files`](./procedures/table-operations#cop [...]
| [Indexes and Search](./procedures/indexes) |
[`create_global_index`](./procedures/indexes#create_global_index),
[`drop_global_index`](./procedures/indexes#drop_global_index),
[`full_text_search`](./procedures/indexes#full_text_search),
[`vector_search`](./procedures/indexes#vector_search),
[`rewrite_file_index`](./procedures/indexes#rewrite_file_index) |
| [Consumers and Query Service](./procedures/consumers) |
[`reset_consumer`](./procedures/consumers#reset_consumer),
[`clear_consumers`](./procedures/consumers#clear_consumers),
[`query_service`](./procedures/consumers#query_service) |
| [Cleanup and Repair](./procedures/repair) |
[`remove_orphan_files`](./procedures/repair#remove_orphan_files),
[`remove_orphan_blobs`](./procedures/repair#remove_orphan_blobs),
[`remove_unexisting_files`](./procedures/repair#remove_unexisting_files),
[`remove_unexisting_manifests`](./procedures/repair#remove_unexisting_manifests),
[`repair`](./procedures/repair#repair),
[`repair_earliest_snapshot`](./procedures/repair#repair_earliest_snapshot) |
diff --git a/docs/docs/flink/procedures/table-operations.md
b/docs/docs/flink/procedures/table-operations.md
index 6eaff3df24..503796f269 100644
--- a/docs/docs/flink/procedures/table-operations.md
+++ b/docs/docs/flink/procedures/table-operations.md
@@ -28,6 +28,47 @@ Merge records, migrate or clone tables, and manage defaults
and partitions.
See [Procedures](../procedures) for Flink version requirements, argument
conventions, and catalog selection.
+## import_files
+
+Append existing files to one partition by recording their original paths in
+`DataFileMeta.externalPath`. The procedure opens source files to extract their
actual row counts
+and available column statistics, using the same metadata extractors as
`migrate_table`. It does not
+copy, rename, or restore files. Archived files must be restored and readable
before import.
+
+**Arguments**
+
+- `table` (`STRING`, required): target Paimon table identifier.
+- `location` (`STRING`, required): absolute source directory path.
+- `partition` (`STRING`, optional): complete partition specification, such as
`dt=2026-10-08,hour=12`.
+ Required for partitioned tables; omit it for an unpartitioned table.
+
+```sql
+CALL sys.import_files(
+ `table` => 'default.T',
+ location => 'oss://bucket/external-files/',
+ `partition` => 'dt=2026-10-08,hour=12'
+);
+
+-- Unpartitioned table.
+CALL sys.import_files('default.unpartitioned_table',
'oss://bucket/external-files/');
+```
+
+Returns the number of registered files. Only files directly in the directory
with a suffix matching
+the table's `file.format` are imported; hidden files and subdirectories are
skipped. An empty
+directory returns zero without creating a snapshot. Repeated directory entries
are deduplicated by
+their full external paths. Files already referenced by the current table
snapshot are rejected before
+reading any source file. Avoid concurrent imports of the same directory.
+
+The target must be an append table with `bucket = -1`, without row tracking or
data evolution.
+Source files must match the target format and non-partition columns. Partition
values come from
+the supplied specification, and source locations must be accessible through
the table's FileIO.
+
+Metadata extraction respects the table's `metadata.stats-mode`; row counts are
collected even when
+column statistics are disabled. If any selected file cannot be read or its
metadata cannot be
+extracted, the import fails without committing a snapshot. Do not modify
registered source files.
+Imported files participate in the normal Paimon lifecycle, including deletion
after compaction or
+partition removal and snapshot expiration.
+
## merge_into
To perform "MERGE INTO" syntax. See [merge_into
action](../action-jars#merging-into-table) for details of arguments.
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index 441033b332..70f329b3a4 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -50,7 +50,7 @@ Choose a group, then use the page contents to jump to a
procedure:
| --- | --- |
| [Compaction and cleanup](./procedures/maintenance) | Compact files and
manifests, rescale buckets, expire history, and repair metadata. |
| [Tags, branches, and rollback](./procedures/versions) | Create and manage
versions, merge branches, or restore table state. |
-| [Migration and copy](./procedures/migration) | Migrate Hive tables or copy
Paimon files. |
+| [Migration and copy](./procedures/migration) | Migrate Hive tables, import
external files, or copy Paimon files. |
| [Indexes and row IDs](./procedures/indexes) | Build or drop indexes and
reassign row IDs. |
| [Consumers, views, and functions](./procedures/metadata) | Manage reader
progress, partition completion, and catalog objects. |
@@ -98,6 +98,7 @@ Choose a group, then use the page contents to jump to a
procedure:
[`migrate_database`](./procedures/migration#migrate_database),
[`migrate_table`](./procedures/migration#migrate_table),
+[`import_files`](./procedures/migration#import_files),
[`copy`](./procedures/migration#copy)
### Indexes and Row IDs
diff --git a/docs/docs/spark/procedures/migration.md
b/docs/docs/spark/procedures/migration.md
index 6812d8f61c..235dbe70c4 100644
--- a/docs/docs/spark/procedures/migration.md
+++ b/docs/docs/spark/procedures/migration.md
@@ -90,3 +90,41 @@ CALL sys.copy(source_table => "t1", target_table =>
"t1_copy");
CALL sys.copy(source_table => "t1", target_table => "t1_copy", where => "day =
'2025-08-17'");
```
+
+## import_files
+
+Append files from a directory to one partition by storing their original
locations in
+`DataFileMeta.externalPath`. The procedure opens source files to extract their
actual row counts
+and available column statistics, using the same metadata extractors as
`migrate_table`. It does not
+copy, rename, or restore files. Archived files must be restored and readable
before import.
+
+**Arguments**
+
+- `table` (`STRING`, required): target Paimon table identifier.
+- `location` (`STRING`, required): absolute source directory path.
+- `partition` (`STRING`, optional): complete partition specification, such as
`dt=2026-10-08,hour=12`.
+ Required for partitioned tables; omit it for an unpartitioned table.
+
+```sql
+CALL sys.import_files(
+ table => 'default.T',
+ location => 'oss://bucket/external-files/',
+ partition => 'dt=2026-10-08,hour=12'
+);
+```
+
+Returns `imported_files`, the number of registered files. An empty directory
returns zero without
+creating a snapshot. Only files directly in the directory whose suffix matches
the table's
+`file.format` are imported. Hidden files and subdirectories are skipped.
Repeated directory entries
+are deduplicated by their full external paths. Files already referenced by the
current table snapshot
+are rejected before reading any source file. Concurrent imports of the same
directory must be avoided.
+
+The target must be an append table with `bucket = -1`, without row tracking or
data evolution.
+Source files must match the target's format and non-partition columns.
Partition values come from
+the supplied specification. Source locations must be accessible through the
table's FileIO.
+
+Metadata extraction respects the table's `metadata.stats-mode`; row counts are
collected even when
+column statistics are disabled. If any selected file cannot be read or its
metadata cannot be
+extracted, the import fails without committing a snapshot. Do not modify
registered source files.
+Imported files participate in the normal Paimon file lifecycle, including
deletion after compaction
+or partition removal and snapshot expiration.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/migrate/ExternalFileImporter.java
b/paimon-core/src/main/java/org/apache/paimon/migrate/ExternalFileImporter.java
new file mode 100644
index 0000000000..096a031c79
--- /dev/null
+++
b/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);
+ }
+ List<DataFileMeta> fileMetas = new ArrayList<>();
+ SimpleStatsExtractor statsExtractor =
+ FileMetaUtils.createSimpleStatsExtractor(table, fileFormat);
+ for (FileStatus file : files.values()) {
+ String externalPath = file.getPath().toString();
+ // Manifest identity is independent of the source name, which can
collide across dirs.
+ DataFileMeta fileMeta =
+ FileMetaUtils.constructFileMeta(
+ "data-" + UUID.randomUUID() + "." + fileFormat,
+ file.getLen(),
+ file.getPath(),
+ statsExtractor,
+ fileIO,
+ table);
+ checkArgument(
+ fileMeta.rowCount() >= 0,
+ "Cannot determine row count for file: %s.",
+ externalPath);
+ fileMetas.add(fileMeta.newExternalPath(externalPath));
+ }
+
+ // Extract every file's metadata before committing, so a failed
extraction adds no files.
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(
+ Collections.singletonList(
+ FileMetaUtils.createCommitMessage(partitionRow,
-1, fileMetas)));
+ }
+ return fileMetas.size();
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/migrate/ExternalFileImporterTest.java
b/paimon-core/src/test/java/org/apache/paimon/migrate/ExternalFileImporterTest.java
new file mode 100644
index 0000000000..d3dc8992a8
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/migrate/ExternalFileImporterTest.java
@@ -0,0 +1,386 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaUtils;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.nio.file.Files;
+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.data.BinaryString.fromString;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+/** Tests external file registration with extracted counts and unchanged
source files. */
+class ExternalFileImporterTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private FileIO fileIO;
+ private Path source;
+
+ @BeforeEach
+ void before() throws IOException {
+ fileIO = spy(LocalFileIO.create());
+ source = new Path(tempDir.resolve("external").toUri());
+ fileIO.mkdirs(source);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"parquet", "orc", "avro"})
+ void testImportExtractsCountsAndPreservesSources(String format) throws
Exception {
+ FileStoreTable table = createTable(format, Collections.emptyMap());
+ Path first = externalFile(format, "first." + format, 1, 2);
+ Path second = externalFile(format, "second." + format, 3);
+ invalidFile("_hidden." + format);
+ invalidFile(".hidden." + format);
+ invalidFile("unrelated.txt");
+ invalidFile("nested/ignored." + format);
+
+ assertThat(
+ ExternalFileImporter.importFiles(
+ table, source.toString(),
"hour=12,dt=2026-10-08"))
+ .isEqualTo(2);
+
+ DataSplit split = splits(table).get(0);
+
assertThat(split.partition().getString(0)).isEqualTo(fromString("2026-10-08"));
+ assertThat(split.partition().getInt(1)).isEqualTo(12);
+ assertThat(split.bucket()).isZero();
+ assertThat(split.rowCount()).isEqualTo(3);
+ assertThat(split.mergedRowCount()).hasValue(3);
+ List<DataFileMeta> metas = split.dataFiles();
+ assertThat(metas).hasSize(2);
+ assertThat(metas.stream().map(f -> f.externalPath().get()))
+ .containsExactlyInAnyOrder(first.toString(),
second.toString());
+ for (DataFileMeta meta : metas) {
+ Path external = new Path(meta.externalPath().get());
+
assertThat(meta.fileSize()).isEqualTo(fileIO.getFileStatus(external).getLen());
+ assertThat(meta.schemaId()).isEqualTo(table.schema().id());
+ assertThat(meta.fileFormat()).isEqualTo(format);
+ assertThat(fileIO.exists(external)).isTrue();
+ boolean isFirst = external.equals(first);
+ assertThat(meta.rowCount()).isEqualTo(isFirst ? 2 : 1);
+ if (!format.equals("avro")) {
+
assertThat(meta.valueStats().minValues().getInt(0)).isEqualTo(isFirst ? 1 : 3);
+
assertThat(meta.valueStats().maxValues().getInt(0)).isEqualTo(isFirst ? 2 : 3);
+ }
+ }
+
assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()).isEqualTo(3);
+
assertThat(table.snapshotManager().latestSnapshot().deltaRecordCount()).isEqualTo(3);
+
assertThat(table.newScan().listPartitionEntries().get(0).recordCount()).isEqualTo(3);
+ verify(fileIO, atLeastOnce())
+ .newInputStream(argThat(path ->
path.toString().startsWith(source.toString())));
+ verify(fileIO, never())
+ .rename(argThat(path ->
path.toString().startsWith(source.toString())), any());
+ verify(fileIO, never())
+ .copyFile(
+ argThat(path ->
path.toString().startsWith(source.toString())),
+ any(),
+ anyBoolean());
+ }
+
+ @Test
+ void testStatsModeNoneStillExtractsRowCount() throws Exception {
+ FileStoreTable table =
+ createTable("parquet",
Collections.singletonMap("metadata.stats-mode", "none"));
+ externalFile("parquet", "data.parquet", 1, 2);
+ ExternalFileImporter.importFiles(table, source.toString(),
"dt=p1,hour=1");
+ DataFileMeta meta = splits(table).get(0).dataFiles().get(0);
+ assertThat(meta.rowCount()).isEqualTo(2);
+ assertThat(meta.valueStats().minValues().getFieldCount()).isZero();
+ assertThat(meta.valueStats().nullCounts().size()).isZero();
+ }
+
+ @Test
+ void testUnreadableFileDoesNotCommit() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ write(table, GenericRow.of(9, fromString("p1"), 1));
+ long snapshotId = table.snapshotManager().latestSnapshotId();
+ Path first = externalFile("parquet", "first.parquet", 1);
+ Path unreadable = externalFile("parquet", "unreadable.parquet", 2);
+ doThrow(new IOException("Source file is unavailable"))
+ .when(fileIO)
+ .newInputStream(unreadable);
+
+ assertThatThrownBy(
+ () ->
+ ExternalFileImporter.importFiles(
+ table, source.toString(),
"dt=p1,hour=1"))
+ .hasStackTraceContaining("Source file is unavailable");
+
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
+ assertThat(splits(table).get(0).dataFiles()).hasSize(1);
+
assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()).isEqualTo(1);
+ assertThat(fileIO.exists(first)).isTrue();
+ assertThat(fileIO.exists(unreadable)).isTrue();
+ }
+
+ @Test
+ void testCorruptFileDoesNotCommit() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ externalFile("parquet", "first.parquet", 1);
+ Path corrupt = invalidFile("corrupt.parquet");
+ assertThatThrownBy(
+ () ->
+ ExternalFileImporter.importFiles(
+ table, source.toString(),
"dt=p1,hour=1"))
+ .hasStackTraceContaining("corrupt.parquet");
+ assertThat(table.snapshotManager().latestSnapshot()).isNull();
+ assertThat(fileIO.exists(corrupt)).isTrue();
+ }
+
+ @Test
+ void testDuplicateDirectoryEntriesAreImportedOnce() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ Path first = externalFile("parquet", "first.parquet", 1, 2);
+ Path second = externalFile("parquet", "second.parquet", 3);
+ doReturn(
+ new FileStatus[] {
+ fileIO.getFileStatus(first),
+ fileIO.getFileStatus(second),
+ fileIO.getFileStatus(first),
+ fileIO.getFileStatus(second)
+ })
+ .when(fileIO)
+ .listStatus(source);
+
+ assertThat(ExternalFileImporter.importFiles(table, source.toString(),
"dt=p1,hour=1"))
+ .isEqualTo(2);
+ DataSplit split = splits(table).get(0);
+ assertThat(split.dataFiles().stream().map(f -> f.externalPath().get()))
+ .containsExactlyInAnyOrder(first.toString(),
second.toString());
+ assertThat(split.rowCount()).isEqualTo(3);
+
assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()).isEqualTo(3);
+
assertThat(table.newScan().listPartitionEntries().get(0).recordCount()).isEqualTo(3);
+ assertThat(fileIO.exists(first)).isTrue();
+ assertThat(fileIO.exists(second)).isTrue();
+ }
+
+ @Test
+ void testDuplicateImportDoesNotCommit() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ Path imported = externalFile("parquet", "first.parquet", 1);
+ ExternalFileImporter.importFiles(table, source.toString(),
"dt=p1,hour=1");
+ long snapshotId = table.snapshotManager().latestSnapshotId();
+ Path fresh = externalFile("parquet", "second.parquet", 2);
+ doReturn(new FileStatus[] {fileIO.getFileStatus(fresh),
fileIO.getFileStatus(imported)})
+ .when(fileIO)
+ .listStatus(source);
+ clearInvocations(fileIO);
+
+ assertThatThrownBy(
+ () ->
+ ExternalFileImporter.importFiles(
+ table, source.toString(),
"dt=p1,hour=1"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("already been imported");
+
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
+ assertThat(splits(table).get(0).dataFiles()).hasSize(1);
+
assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()).isEqualTo(1);
+ verify(fileIO, never())
+ .newInputStream(argThat(path ->
path.toString().startsWith(source.toString())));
+ }
+
+ @Test
+ void testSameSourceNameInDifferentDirectories() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ externalFile("parquet", "data.parquet", 1);
+ externalFile("parquet", "nested/data.parquet", 2);
+ ExternalFileImporter.importFiles(table, source.toString(),
"dt=p1,hour=1");
+ ExternalFileImporter.importFiles(
+ table, new Path(source, "nested").toString(), "dt=p1,hour=1");
+
+ List<DataFileMeta> metas = splits(table).get(0).dataFiles();
+ assertThat(metas).hasSize(2);
+
assertThat(metas.stream().map(DataFileMeta::fileName).distinct()).hasSize(2);
+ assertThat(metas.stream().map(f ->
f.externalPath().get()).distinct()).hasSize(2);
+
assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()).isEqualTo(2);
+ }
+
+ @Test
+ void testCountsWithExistingData() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ write(table, GenericRow.of(1, fromString("p1"), 1));
+ externalFile("parquet", "imported.parquet", 3, 4);
+ ExternalFileImporter.importFiles(table, source.toString(),
"dt=p1,hour=1");
+ write(table, GenericRow.of(2, fromString("p1"), 1));
+
+ DataSplit split = splits(table).get(0);
+ assertThat(split.dataFiles()).hasSize(3);
+ assertThat(split.rowCount()).isEqualTo(4);
+ assertThat(split.mergedRowCount()).hasValue(4);
+
assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()).isEqualTo(4);
+
assertThat(table.newScan().listPartitionEntries().get(0).recordCount()).isEqualTo(4);
+ }
+
+ @Test
+ void testEmptyDirectoryDoesNotCommit() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ assertThat(ExternalFileImporter.importFiles(table, source.toString(),
"dt=p1,hour=1"))
+ .isZero();
+ assertThat(table.snapshotManager().latestSnapshot()).isNull();
+ }
+
+ @Test
+ void testCompletePartitionIsRequired() throws Exception {
+ FileStoreTable table = createTable("parquet", Collections.emptyMap());
+ for (String partition : Arrays.asList(null, "", "dt=p1",
"dt=p1,hour=1,extra=x")) {
+ assertThatThrownBy(
+ () ->
+ ExternalFileImporter.importFiles(
+ table, source.toString(),
partition))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("exactly the table partition keys");
+ }
+ assertThatThrownBy(
+ () ->
+ ExternalFileImporter.importFiles(
+ table, source.toString(),
"dt=p1,hour=invalid"))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThat(table.snapshotManager().latestSnapshot()).isNull();
+ }
+
+ @Test
+ void testUnsupportedTables() throws Exception {
+ Map<String, String> bucketOptions = new HashMap<>();
+ bucketOptions.put("bucket", "1");
+ bucketOptions.put("bucket-key", "id");
+ FileStoreTable bucketed = createTable("parquet", bucketOptions);
+ assertThatThrownBy(
+ () ->
+ ExternalFileImporter.importFiles(
+ bucketed, source.toString(),
"dt=p1,hour=1"))
+ .hasMessageContaining("bucket = -1");
+ FileStoreTable tracked =
+ createTable("parquet",
Collections.singletonMap("row-tracking.enabled", "true"));
+ assertThatThrownBy(
+ () ->
+ ExternalFileImporter.importFiles(
+ tracked, source.toString(),
"dt=p1,hour=1"))
+ .hasMessageContaining("row tracking");
+ }
+
+ private Path externalFile(String format, String name, int... ids) throws
Exception {
+ FileStoreTable producer = createTable(format, Collections.emptyMap());
+ GenericRow[] rows =
+ Arrays.stream(ids)
+ .mapToObj(id -> GenericRow.of(id,
fromString("source"), 0))
+ .toArray(GenericRow[]::new);
+ write(producer, rows);
+ DataSplit split = splits(producer).get(0);
+ DataFileMeta meta = split.dataFiles().get(0);
+ Path dataFile =
+ producer.store()
+ .pathFactory()
+ .createDataFilePathFactory(split.partition(),
split.bucket())
+ .toPath(meta);
+ Path external = new Path(source, name);
+ fileIO.mkdirs(external.getParent());
+ fileIO.copyFile(dataFile, external, true);
+ return external;
+ }
+
+ private Path invalidFile(String name) throws IOException {
+ java.nio.file.Path path = tempDir.resolve("external").resolve(name);
+ Files.createDirectories(path.getParent());
+ Files.write(path, new byte[] {1, 2, 3});
+ return new Path(path.toUri());
+ }
+
+ private FileStoreTable createTable(String format, Map<String, String>
overrides)
+ throws Exception {
+ Path tablePath = new Path(tempDir.resolve("table-" +
System.nanoTime()).toUri());
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(CoreOptions.FILE_FORMAT.key(), format);
+ options.putAll(overrides);
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("dt", DataTypes.STRING())
+ .field("hour", DataTypes.INT())
+ .build();
+ TableSchema schema =
+ SchemaUtils.forceCommit(
+ new FileSystemSchemaManager(fileIO, tablePath),
+ new Schema(
+ rowType.getFields(),
+ Arrays.asList("dt", "hour"),
+ Collections.emptyList(),
+ options,
+ ""));
+ return FileStoreTableFactory.create(fileIO, tablePath, schema);
+ }
+
+ private List<DataSplit> splits(FileStoreTable table) {
+ return table.newScan().plan().splits().stream()
+ .map(s -> (DataSplit) s)
+ .collect(Collectors.toList());
+ }
+
+ private void write(FileStoreTable table, GenericRow... rows) throws
Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ImportFilesProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ImportFilesProcedure.java
new file mode 100644
index 0000000000..58f1fb9428
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ImportFilesProcedure.java
@@ -0,0 +1,50 @@
+/*
+ * 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.migrate.ExternalFileImporter;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import javax.annotation.Nullable;
+
+/** Imports files from a directory into a partition using external paths. */
+public class ImportFilesProcedure extends ProcedureBase {
+
+ @Override
+ public String identifier() {
+ return "import_files";
+ }
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "location", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(name = "partition", type =
@DataTypeHint("STRING"), isOptional = true)
+ },
+ output = @DataTypeHint("BIGINT"))
+ public Long[] call(
+ ProcedureContext context, String tableId, String location,
@Nullable String partition)
+ throws Exception {
+ return new Long[] {ExternalFileImporter.importFiles(table(tableId),
location, partition)};
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 288431c31a..b9223d0334 100644
---
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -111,3 +111,4 @@ org.apache.paimon.flink.procedure.CreateGlobalIndexProcedure
org.apache.paimon.flink.procedure.VectorSearchProcedure
org.apache.paimon.flink.procedure.FullTextSearchProcedure
org.apache.paimon.flink.procedure.DropGlobalIndexProcedure
+org.apache.paimon.flink.procedure.ImportFilesProcedure
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ImportFilesProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ImportFilesProcedureITCase.java
new file mode 100644
index 0000000000..23ddf9f750
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ImportFilesProcedureITCase.java
@@ -0,0 +1,102 @@
+/*
+ * 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.data.GenericRow;
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** SQL integration tests for {@link ImportFilesProcedure}. */
+class ImportFilesProcedureITCase extends CatalogITCaseBase {
+
+ @Test
+ void testImportIntoPartition() throws Exception {
+ Path location = prepareFiles();
+ sql(
+ "CREATE TABLE T (id INT, dt STRING, `hour` INT) PARTITIONED BY
(dt, `hour`) "
+ + "WITH ('bucket'='-1', 'file.format'='parquet')");
+
+ assertThat(
+ sql("CALL sys.import_files(`table` => 'default.T',
location => '"
+ + location
+ + "', `partition` => 'hour=12,dt=p1')")
+ .get(0)
+ .getField(0))
+ .isEqualTo(1L);
+ FileStoreTable table = paimonTable("T");
+ DataSplit split = (DataSplit) table.newScan().plan().splits().get(0);
+ DataFileMeta file = split.dataFiles().get(0);
+ assertThat(file.externalPath()).isPresent();
+ assertThat(table.fileIO().exists(new
Path(file.externalPath().get()))).isTrue();
+ assertThat(file.rowCount()).isEqualTo(2L);
+ assertThat(sql("SELECT * FROM T ORDER BY id").toString())
+ .isEqualTo("[+I[1, p1, 12], +I[2, p1, 12]]");
+ assertThat(sql("SELECT COUNT(*) FROM
T").get(0).getField(0)).isEqualTo(2L);
+ sql("ALTER TABLE T ADD added INT");
+ assertThat(sql("SELECT COUNT(*) FROM T WHERE added IS
NULL").get(0).getField(0))
+ .isEqualTo(2L);
+ }
+
+ @Test
+ void testImportIntoUnpartitionedTable() throws Exception {
+ Path location = prepareFiles();
+ sql("CREATE TABLE T (id INT) WITH ('bucket'='-1',
'file.format'='parquet')");
+ assertThat(sql("CALL sys.import_files('default.T', '" + location +
"')").get(0).getField(0))
+ .isEqualTo(1L);
+ assertThat(sql("SELECT COUNT(*) FROM
T").get(0).getField(0)).isEqualTo(2L);
+ }
+
+ private Path prepareFiles() throws Exception {
+ sql("CREATE TABLE S (id INT) WITH ('bucket'='-1',
'file.format'='parquet')");
+ FileStoreTable source = paimonTable("S");
+ BatchWriteBuilder builder = source.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(1));
+ write.write(GenericRow.of(2));
+ commit.commit(write.prepareCommit());
+ }
+ Path location = new Path(path, "external");
+ source.fileIO().mkdirs(location);
+ for (Split split : source.newScan().plan().splits()) {
+ DataSplit dataSplit = (DataSplit) split;
+ for (DataFileMeta file : dataSplit.dataFiles()) {
+ Path sourcePath =
+ source.store()
+ .pathFactory()
+ .createDataFilePathFactory(
+ dataSplit.partition(),
dataSplit.bucket())
+ .toPath(file);
+ source.fileIO().copyFile(sourcePath, new Path(location,
file.fileName()), true);
+ }
+ }
+ return location;
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index aebc19389e..61d318a5d9 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -43,6 +43,7 @@ import
org.apache.paimon.spark.procedure.ExpireSnapshotsProcedure;
import org.apache.paimon.spark.procedure.ExpireTagsProcedure;
import org.apache.paimon.spark.procedure.FastForwardProcedure;
import org.apache.paimon.spark.procedure.GrantPermissionProcedure;
+import org.apache.paimon.spark.procedure.ImportFilesProcedure;
import org.apache.paimon.spark.procedure.ListPermissionsProcedure;
import org.apache.paimon.spark.procedure.ListPoliciesProcedure;
import org.apache.paimon.spark.procedure.MarkPartitionDoneProcedure;
@@ -124,6 +125,7 @@ public class SparkProcedures {
procedureBuilders.put("rescale", RescaleProcedure::builder);
procedureBuilders.put("migrate_database",
MigrateDatabaseProcedure::builder);
procedureBuilders.put("migrate_table", MigrateTableProcedure::builder);
+ procedureBuilders.put("import_files", ImportFilesProcedure::builder);
procedureBuilders.put("remove_orphan_files",
RemoveOrphanFilesProcedure::builder);
procedureBuilders.put("remove_orphan_blobs",
RemoveOrphanBlobsProcedure::builder);
procedureBuilders.put("remove_unexisting_files",
RemoveUnexistingFilesProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ImportFilesProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ImportFilesProcedure.java
new file mode 100644
index 0000000000..69bbb46b4f
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ImportFilesProcedure.java
@@ -0,0 +1,101 @@
+/*
+ * 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.spark.procedure;
+
+import org.apache.paimon.migrate.ExternalFileImporter;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+
+import static org.apache.spark.sql.types.DataTypes.LongType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** Imports files from a directory into a partition using external paths. */
+public class ImportFilesProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS = {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("location", StringType),
+ ProcedureParameter.optional("partition", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("imported_files", LongType, false,
Metadata.empty())
+ });
+
+ private ImportFilesProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ Identifier ident = toIdentifier(args.getString(0),
PARAMETERS[0].name());
+ return modifyPaimonTable(
+ ident,
+ table -> {
+ try {
+ long importedFiles =
+ ExternalFileImporter.importFiles(
+ table,
+ args.getString(1),
+ args.isNullAt(2) ? null :
args.getString(2));
+ return new InternalRow[]
{newInternalRow(importedFiles)};
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to import files",
e);
+ }
+ });
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<ImportFilesProcedure>() {
+ @Override
+ public ImportFilesProcedure doBuild() {
+ return new ImportFilesProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "ImportFilesProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ImportFilesProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ImportFilesProcedureTest.scala
new file mode 100644
index 0000000000..6a87eff049
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ImportFilesProcedureTest.scala
@@ -0,0 +1,111 @@
+/*
+ * 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.spark.procedure
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.source.DataSplit
+
+import org.apache.spark.sql.Row
+
+import java.io.File
+
+import scala.collection.JavaConverters._
+
+class ImportFilesProcedureTest extends PaimonSparkTestBase {
+
+ Seq("parquet", "orc", "avro").foreach {
+ format =>
+ test(s"import $format files with real counts into a partition") {
+ withTempDir {
+ dir =>
+ val location = new File(dir, "external").getCanonicalPath
+ spark
+ .sql("SELECT 1 AS id UNION ALL SELECT 2 UNION ALL SELECT 3")
+ .repartition(2)
+ .write
+ .format(format)
+ .save(location)
+ val sourceFiles = new
File(location).listFiles.filter(_.getName.endsWith(s".$format"))
+
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, dt STRING, hour INT)
+ |PARTITIONED BY (dt, hour)
+ |TBLPROPERTIES ('bucket'='-1',
'file.format'='$format')
+ |""".stripMargin)
+ spark.sql("INSERT INTO T VALUES (4, 'p1', 12), (5, 'p2', 13)")
+ spark.sql("CACHE TABLE T")
+ try {
+ checkAnswer(
+ spark.sql(s"""
+ |CALL sys.import_files(
+ | table => 'test.T',
+ | location => '$location',
+ | partition => 'hour=12,dt=p1')
+ |""".stripMargin),
+ Row(sourceFiles.length.toLong)
+ )
+
+ assert(sourceFiles.forall(_.exists()))
+ val table =
+
paimonCatalog.getTable(org.apache.paimon.catalog.Identifier.create("test", "T"))
+ val externalFiles =
table.newReadBuilder.newScan.plan.splits.asScala
+ .flatMap(_.asInstanceOf[DataSplit].dataFiles.asScala)
+ .filter(_.externalPath.isPresent)
+ assert(externalFiles.size == sourceFiles.length)
+ assert(externalFiles.forall(_.rowCount >= 0L))
+ assert(externalFiles.map(_.rowCount).sum == 3L)
+
+ checkAnswer(
+ spark.sql("SELECT * FROM T WHERE dt = 'p1' ORDER BY id"),
+ Seq(Row(1, "p1", 12), Row(2, "p1", 12), Row(3, "p1", 12),
Row(4, "p1", 12)))
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T"), Row(5L))
+ checkAnswer(
+ spark.sql("SELECT dt, COUNT(*) FROM T GROUP BY dt"),
+ Seq(Row("p1", 4L), Row("p2", 1L)))
+ } finally {
+ spark.sql("UNCACHE TABLE T")
+ }
+
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T"), Row(5L))
+ checkAnswer(
+ spark.sql("SELECT dt, COUNT(*) FROM T GROUP BY dt"),
+ Seq(Row("p1", 4L), Row("p2", 1L)))
+ spark.sql("ALTER TABLE T ADD COLUMN added INT")
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T WHERE added IS
NULL"), Row(5L))
+ checkAnswer(
+ spark.sql("SELECT id FROM T WHERE id > 2 ORDER BY id LIMIT 2"),
+ Seq(Row(3), Row(4)))
+ assert(spark.sql("SELECT * FROM T LIMIT 3").collect().length == 3)
+ spark.sql("ANALYZE TABLE T COMPUTE STATISTICS FOR COLUMNS id")
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T"), Row(5L))
+ }
+ }
+ }
+
+ test("import into an unpartitioned table without optional arguments") {
+ withTempDir {
+ dir =>
+ val location = new File(dir, "external").getCanonicalPath
+ spark.sql("SELECT 1 AS id UNION ALL SELECT
2").coalesce(1).write.parquet(location)
+ spark.sql("CREATE TABLE T (id INT) TBLPROPERTIES ('bucket'='-1',
'file.format'='parquet')")
+ checkAnswer(spark.sql(s"CALL sys.import_files('test.T',
'$location')"), Row(1L))
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T"), Row(2L))
+ }
+ }
+}
