This is an automated email from the ASF dual-hosted git repository. JingsongLi pushed a commit to branch codex/import-external-files in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 8e5b06d2eb466aef46dad22204e3809b5ce586f5 Author: JingsongLi <[email protected]> AuthorDate: Thu Oct 8 12:44:11 2026 +0800 [spark][flink] Add metadata-only external file import procedure --- docs/docs/flink/procedures.md | 2 +- docs/docs/flink/procedures/table-operations.md | 42 +++ docs/docs/spark/procedures.md | 3 +- docs/docs/spark/procedures/migration.md | 42 +++ .../paimon/migrate/ExternalFileImporter.java | 147 ++++++++++ .../paimon/migrate/ExternalFileImporterTest.java | 300 +++++++++++++++++++++ .../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 | 110 ++++++++ 12 files changed, 900 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..8c43093b4e 100644 --- a/docs/docs/flink/procedures/table-operations.md +++ b/docs/docs/flink/procedures/table-operations.md @@ -28,6 +28,48 @@ 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 lists directory entries and obtains file sizes without +opening, copying, renaming, or restoring source files, including archived files. + +**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/archived-files/', + `partition` => 'dt=2026-10-08,hour=12' +); + +-- Unpartitioned table. +CALL sys.import_files('default.unpartitioned_table', 'oss://bucket/archived-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. Files already referenced by the current table +snapshot are rejected. Avoid concurrent imports of the same directory. + +The target must be an append table with `bucket = -1`, without row tracking or data evolution, +and with Iceberg metadata synchronization disabled. Import requires a catalog using filesystem +snapshot commits, such as the filesystem or Hive catalog. Catalog-managed snapshot commits, +including the REST catalog, do not support unknown row counts and reject the import. +Source files must match the target format and non-partition columns; their contents cannot be +validated during import. Partition values come from the supplied specification, and source +locations must be accessible through the table's FileIO. + +File row counts are unknown (`-1`) and column statistics are omitted. Queries require the source +files to be restored and readable. Snapshot row counts remain unknown after import. 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..d0b74c8055 100644 --- a/docs/docs/spark/procedures/migration.md +++ b/docs/docs/spark/procedures/migration.md @@ -90,3 +90,45 @@ 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 only lists directory entries and obtains file sizes; +it does not open, copy, rename, or restore source files. This allows importing archived files whose +contents are unavailable. + +**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/archived-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. Files already referenced +by the current table snapshot are rejected; 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, +and with Iceberg metadata synchronization disabled. Import requires a catalog using filesystem +snapshot commits, such as the filesystem or Hive catalog. Catalog-managed snapshot commits, +including the REST catalog, do not support unknown row counts and reject the import. +Source files must match the target's format and non-partition columns; the procedure cannot validate +their contents. Partition values come from the supplied specification. Source locations must be +accessible through the table's FileIO. + +Row counts are recorded as unknown (`-1`) and column statistics are omitted. Count aggregation +pushdown is disabled for splits containing these files, so queries read their actual contents after +they are restored. Importing does not restore the files. Snapshot row counts remain unknown after +an import. 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..564867c528 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/migrate/ExternalFileImporter.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.migrate; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryWriter; +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.manifest.FileSource; +import org.apache.paimon.stats.SimpleStats; +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.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."); + + List<FileStatus> files = new ArrayList<>(); + for (FileStatus status : fileIO.listStatus(directory)) { + String name = status.getPath().getName(); + if (!status.isDir() + && !name.startsWith(".") + && !name.startsWith("_") + && name.endsWith("." + fileFormat)) { + files.add(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); + } + } + List<DataFileMeta> fileMetas = new ArrayList<>(); + for (FileStatus file : files) { + String externalPath = file.getPath().toString(); + checkArgument( + !existingPaths.contains(externalPath), + "File has already been imported: %s.", + externalPath); + // Manifest identity is independent of the source name, which can collide across dirs. + fileMetas.add( + DataFileMeta.forAppend( + "data-" + UUID.randomUUID() + "." + fileFormat, + file.getLen(), + DataFileMeta.UNKNOWN_ROW_COUNT, + SimpleStats.EMPTY_STATS, + 0, + 0, + fileStoreTable.schema().id(), + Collections.emptyList(), + null, + FileSource.APPEND, + Collections.emptyList(), + externalPath, + null, + null)); + } + + // Only directory/file metadata is accessed. Archived source files are never opened. + 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..ad6fc0e775 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/migrate/ExternalFileImporterTest.java @@ -0,0 +1,300 @@ +/* + * 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.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.BucketEntry; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.PartitionEntry; +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.argThat; +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 for metadata-only imports, including source files that cannot be opened. */ +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("frozen").toUri()); + fileIO.mkdirs(source); + doThrow(new IOException("Source file is frozen")) + .when(fileIO) + .newInputStream(argThat(path -> path.toString().contains("/frozen/"))); + } + + @ParameterizedTest + @ValueSource(strings = {"parquet", "orc", "avro"}) + void testImportWithoutOpeningSourceFiles(String format) throws Exception { + FileStoreTable table = createTable(format, Collections.emptyMap()); + Path first = frozenFile("first." + format); + Path second = frozenFile("second." + format); + frozenFile("_hidden." + format); + frozenFile(".hidden." + format); + frozenFile("unrelated.txt"); + fileIO.mkdirs(new Path(source, "nested")); + frozenFile("nested/ignored." + format); + + assertThat( + ExternalFileImporter.importFiles( + table, source.toString(), "hour=12,dt=2026-10-08")) + .isEqualTo(2); + + List<DataSplit> splits = splits(table); + assertThat(splits).hasSize(1); + DataSplit split = splits.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(DataFileMeta.UNKNOWN_ROW_COUNT); + assertThat(split.mergedRowCount()).isEmpty(); + 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) { + assertThat(meta.fileSize()).isEqualTo(3); + assertThat(meta.schemaId()).isEqualTo(table.schema().id()); + assertThat(meta.rowCount()).isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT); + assertThat(meta.fileFormat()).isEqualTo(format); + assertThat(meta.valueStatsCols()).isEmpty(); + assertThat(meta.valueStats().minValues().getFieldCount()).isZero(); + assertThat(table.fileIO().exists(new Path(meta.externalPath().get()))).isTrue(); + } + assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()) + .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT); + assertThat(table.newScan().listPartitionEntries().get(0).recordCount()) + .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT); + verify(fileIO, never()) + .newInputStream(argThat(path -> path.toString().contains("/frozen/"))); + } + + @Test + void testDuplicateImportDoesNotCommit() throws Exception { + FileStoreTable table = createTable("parquet", Collections.emptyMap()); + frozenFile("first.parquet"); + ExternalFileImporter.importFiles(table, source.toString(), "dt=p1,hour=1"); + long snapshotId = table.snapshotManager().latestSnapshotId(); + frozenFile("second.parquet"); + + 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); + } + + @Test + void testSameSourceNameInDifferentDirectories() throws Exception { + FileStoreTable table = createTable("parquet", Collections.emptyMap()); + frozenFile("data.parquet"); + frozenFile("nested/data.parquet"); + + 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); + } + + @Test + void testUnknownRowCountWithExistingData() throws Exception { + FileStoreTable table = createTable("parquet", Collections.emptyMap()); + write(table, GenericRow.of(1, fromString("p1"), 1)); + frozenFile("imported.parquet"); + 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(DataFileMeta.UNKNOWN_ROW_COUNT); + assertThat(split.mergedRowCount()).isEmpty(); + assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()) + .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT); + assertThat(table.newScan().listPartitionEntries().get(0).recordCount()) + .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT); + + List<ManifestEntry> entries = + split.dataFiles().stream() + .map(f -> ManifestEntry.create(FileKind.ADD, split.partition(), 0, -1, f)) + .collect(Collectors.toList()); + assertThat(BucketEntry.merge(entries).iterator().next().recordCount()) + .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT); + ManifestEntry known = + entries.stream().filter(e -> e.file().rowCount() == 1).findFirst().get(); + ManifestEntry deleteKnown = + ManifestEntry.create(FileKind.DELETE, split.partition(), 0, -1, known.file()); + // A delta that deletes one known row is -1 too, but must still cancel an ADD of one row. + assertThat( + PartitionEntry.fromManifestEntry(deleteKnown) + .merge(PartitionEntry.fromManifestEntry(known)) + .recordCount()) + .isZero(); + assertThat( + BucketEntry.fromManifestEntry(deleteKnown) + .merge(BucketEntry.fromManifestEntry(known)) + .recordCount()) + .isZero(); + } + + @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 frozenFile(String name) throws IOException { + java.nio.file.Path path = tempDir.resolve("frozen").resolve(name); + Files.createDirectories(path.getParent()); + // Invalid file contents ensure an attempted footer read would fail even without the spy. + 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 row) throws Exception { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + 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..db015d8a3e --- /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(DataFileMeta.UNKNOWN_ROW_COUNT); + 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..023cca9c4a --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ImportFilesProcedureTest.scala @@ -0,0 +1,110 @@ +/* + * 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 into a partition and query after restoring files") { + 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 == -1L)) + + 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)) + } + } +}
