From eedecdd3a1c280ca0fb141c93a93ee056934d58f Mon Sep 17 00:00:00 2001 From: Xi Chen Date: Mon, 10 Feb 2025 10:30:30 +0800 Subject: [PATCH 1/5] Data: Handle case where partition location is missing for `TableMigrationUtil` --- .../iceberg/data/TableMigrationUtil.java | 5 +- .../iceberg/data/TestTableMigrationUtil.java | 107 ++++++++++++++++++ 2 files changed, 111 insertions(+), 1 deletion(-) create mode 100644 data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java diff --git a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java index eb1c1a341e2b..8f61096b0e48 100644 --- a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java +++ b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java @@ -164,7 +164,10 @@ public static List listPartition( Path partitionDir = new Path(partitionUri); FileSystem fs = partitionDir.getFileSystem(conf); List fileStatus = - Arrays.stream(fs.listStatus(partitionDir, HIDDEN_PATH_FILTER)) + Arrays.stream( + fs.exists(partitionDir) + ? fs.listStatus(partitionDir, HIDDEN_PATH_FILTER) + : new FileStatus[] {}) .filter(FileStatus::isFile) .collect(Collectors.toList()); DataFile[] datafiles = new DataFile[fileStatus.size()]; diff --git a/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java b/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java new file mode 100644 index 000000000000..2a2b8dbc7942 --- /dev/null +++ b/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.data; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import org.apache.avro.generic.GenericData; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Files; +import org.apache.iceberg.MetricsConfig; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.avro.RandomAvroData; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.parquet.Parquet; +import org.apache.iceberg.parquet.ParquetAvroWriter; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class TestTableMigrationUtil { + private static final Schema SCHEMA = + new Schema( + optional(1, "id", Types.IntegerType.get()), optional(2, "data", Types.StringType.get())); + private static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA).identity("id").build(); + private static final Map PARTITION = Map.of("id", "1"); + private static final String FORMAT = "parquet"; + private static final Configuration CONF = new Configuration(); + + @TempDir protected Path tempTableLocation; + + @Test + void testListPartition() throws IOException { + Path partitionPath = tempTableLocation.resolve("id=1"); + String partitionUri = partitionPath.toUri().toString(); + java.nio.file.Files.createDirectories(partitionPath); + writePartitionFile(partitionPath.toFile()); + + List dataFiles = + TableMigrationUtil.listPartition( + PARTITION, partitionUri, FORMAT, SPEC, CONF, MetricsConfig.getDefault(), null); + assertThat(dataFiles) + .as("List partition with 1 Parquet file should return 1 DataFile") + .hasSize(1); + } + + @Test + void testListEmptyPartition() throws IOException { + Path partitionPath = tempTableLocation.resolve("id=1"); + String partitionUri = partitionPath.toUri().toString(); + java.nio.file.Files.createDirectories(partitionPath); + + List dataFiles = + TableMigrationUtil.listPartition( + PARTITION, partitionUri, FORMAT, SPEC, CONF, MetricsConfig.getDefault(), null); + assertThat(dataFiles).as("List partition with 0 file should return 0 DataFile").isEmpty(); + } + + @Test + void testListPartitionLocationNotExists() { + String partitionUri = tempTableLocation.resolve("id=1").toUri().toString(); + + List dataFiles = + TableMigrationUtil.listPartition( + PARTITION, partitionUri, FORMAT, SPEC, CONF, MetricsConfig.getDefault(), null); + assertThat(dataFiles) + .as("List partition of which location does not exist should return 0 DataFile") + .isEmpty(); + } + + private static void writePartitionFile(File outputDir) throws IOException { + Iterable records = RandomAvroData.generate(SCHEMA, 1000, 54310); + File testFile = File.createTempFile("junit", ".parquet", outputDir); + assertThat(testFile.delete()).as("Delete should succeed").isTrue(); + + try (FileAppender writer = + Parquet.write(Files.localOutput(testFile)) + .schema(SCHEMA) + .createWriterFunc(ParquetAvroWriter::buildWriter) + .build()) { + writer.addAll(records); + } + } +} From 66abe943c9fb616a197273ba0d7d9965b4649c63 Mon Sep 17 00:00:00 2001 From: Xi Chen Date: Sun, 16 Feb 2025 10:33:15 +0800 Subject: [PATCH 2/5] Add end-to-end test to `TestSnapshotTableAction` --- .../actions/TestSnapshotTableAction.java | 21 +++++++++++++++++++ .../actions/TestSnapshotTableAction.java | 20 ++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java index d9c42a07b853..5d8b03b0d527 100644 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java +++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java @@ -20,8 +20,11 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.io.File; import java.io.IOException; import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Files; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.apache.iceberg.ParameterizedTestExtension; @@ -65,4 +68,22 @@ public void testSnapshotWithParallelTasks() throws IOException { .execute(); assertThat(snapshotThreadsIndex.get()).isEqualTo(2); } + + @Test + public void testSnapshotWithEmptyPartitionOrLocationMissingPartition() throws IOException { + File tableLocation = temp.newFolder(); + sql( + "CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet PARTITIONED BY (bucket string) LOCATION '%s'", + SOURCE_NAME, tableLocation.toURI().toString()); + sql("ALTER TABLE %s ADD PARTITION (bucket = 'foo')", SOURCE_NAME); + sql("ALTER TABLE %s ADD PARTITION (bucket = 'bar')", SOURCE_NAME); + + // Delete the 2nd partition location + Path partitionLocation = tableLocation.toPath().resolve("bucket=bar"); + Files.delete(partitionLocation); + + SparkActions.get().snapshotTable(SOURCE_NAME).as(tableName).execute(); + assertThat(scalarSql("SELECT count(*) FROM %s.partitions", selectTarget())) + .as("Should have 0 partition after snapshot") + .isEqualTo(0L); } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java index d9c42a07b853..f330fe07fc35 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.nio.file.Files; +import java.nio.file.Path; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.apache.iceberg.ParameterizedTestExtension; @@ -65,4 +66,23 @@ public void testSnapshotWithParallelTasks() throws IOException { .execute(); assertThat(snapshotThreadsIndex.get()).isEqualTo(2); } + + @TestTemplate + public void testSnapshotWithEmptyPartitionOrLocationMissingPartition() throws IOException { + Path tableLocationPath = Files.createTempDirectory(temp, "junit"); + sql( + "CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet PARTITIONED BY (bucket string) LOCATION '%s'", + SOURCE_NAME, tableLocationPath.toFile().toString()); + sql("ALTER TABLE %s ADD PARTITION (bucket = 'foo')", SOURCE_NAME); + sql("ALTER TABLE %s ADD PARTITION (bucket = 'bar')", SOURCE_NAME); + + // Delete the 2nd partition location + Path partitionLocationPath = tableLocationPath.resolve("bucket=bar"); + Files.delete(partitionLocationPath); + + SparkActions.get().snapshotTable(SOURCE_NAME).as(tableName).execute(); + assertThat(scalarSql("SELECT count(*) FROM %s.partitions", selectTarget())) + .as("Should have 0 partition after snapshot") + .isEqualTo(0L); + } } From ce991f6ac91044b06373c88c2c32da219b16981f Mon Sep 17 00:00:00 2001 From: Xi Chen Date: Wed, 19 Feb 2025 10:42:58 +0800 Subject: [PATCH 3/5] Add log when skipping partition with missing location --- .../iceberg/data/TableMigrationUtil.java | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java index 8f61096b0e48..99b12748026f 100644 --- a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java +++ b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; @@ -45,8 +46,11 @@ import org.apache.iceberg.parquet.ParquetUtil; import org.apache.iceberg.util.Tasks; import org.apache.iceberg.util.ThreadPools; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class TableMigrationUtil { + private static final Logger LOG = LoggerFactory.getLogger(TableMigrationUtil.class); private static final PathFilter HIDDEN_PATH_FILTER = p -> !p.getName().startsWith("_") && !p.getName().startsWith("."); @@ -163,13 +167,19 @@ public static List listPartition( Path partitionDir = new Path(partitionUri); FileSystem fs = partitionDir.getFileSystem(conf); - List fileStatus = - Arrays.stream( - fs.exists(partitionDir) - ? fs.listStatus(partitionDir, HIDDEN_PATH_FILTER) - : new FileStatus[] {}) - .filter(FileStatus::isFile) - .collect(Collectors.toList()); + List fileStatus; + if (fs.exists(partitionDir)) { + fileStatus = + Arrays.stream(fs.listStatus(partitionDir, HIDDEN_PATH_FILTER)) + .filter(FileStatus::isFile) + .collect(Collectors.toList()); + } else { + LOG.info( + "Skipping partition {}: location {} does not exist in the filesystem.", + partition, + partitionUri); + fileStatus = Collections.emptyList(); + } DataFile[] datafiles = new DataFile[fileStatus.size()]; Tasks.Builder task = Tasks.range(fileStatus.size()).stopOnFailure().throwFailureWhenFinished(); From 909f0c2877e061e0bff6ac69d52ef28cadfb7f14 Mon Sep 17 00:00:00 2001 From: Xi Chen Date: Fri, 2 May 2025 17:29:15 +0800 Subject: [PATCH 4/5] Handle FileNotFoundException from caller --- .../iceberg/data/TableMigrationUtil.java | 21 +--- .../iceberg/data/TestTableMigrationUtil.java | 17 +-- .../apache/iceberg/spark/SparkTableUtil.java | 115 +++++++++++++---- .../actions/TestSnapshotTableAction.java | 21 ---- .../source/TestIcebergSourceTablesBase.java | 96 ++++++++++++++ .../apache/iceberg/spark/SparkTableUtil.java | 119 ++++++++++++++---- .../actions/TestSnapshotTableAction.java | 20 --- .../source/TestIcebergSourceTablesBase.java | 95 ++++++++++++++ 8 files changed, 385 insertions(+), 119 deletions(-) diff --git a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java index 99b12748026f..eb1c1a341e2b 100644 --- a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java +++ b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java @@ -21,7 +21,6 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.util.Arrays; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; @@ -46,11 +45,8 @@ import org.apache.iceberg.parquet.ParquetUtil; import org.apache.iceberg.util.Tasks; import org.apache.iceberg.util.ThreadPools; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; public class TableMigrationUtil { - private static final Logger LOG = LoggerFactory.getLogger(TableMigrationUtil.class); private static final PathFilter HIDDEN_PATH_FILTER = p -> !p.getName().startsWith("_") && !p.getName().startsWith("."); @@ -167,19 +163,10 @@ public static List listPartition( Path partitionDir = new Path(partitionUri); FileSystem fs = partitionDir.getFileSystem(conf); - List fileStatus; - if (fs.exists(partitionDir)) { - fileStatus = - Arrays.stream(fs.listStatus(partitionDir, HIDDEN_PATH_FILTER)) - .filter(FileStatus::isFile) - .collect(Collectors.toList()); - } else { - LOG.info( - "Skipping partition {}: location {} does not exist in the filesystem.", - partition, - partitionUri); - fileStatus = Collections.emptyList(); - } + List fileStatus = + Arrays.stream(fs.listStatus(partitionDir, HIDDEN_PATH_FILTER)) + .filter(FileStatus::isFile) + .collect(Collectors.toList()); DataFile[] datafiles = new DataFile[fileStatus.size()]; Tasks.Builder task = Tasks.range(fileStatus.size()).stopOnFailure().throwFailureWhenFinished(); diff --git a/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java b/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java index 2a2b8dbc7942..9fcb2b077883 100644 --- a/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java +++ b/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java @@ -20,8 +20,10 @@ import static org.apache.iceberg.types.Types.NestedField.optional; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; import java.nio.file.Path; import java.util.List; @@ -80,15 +82,16 @@ void testListEmptyPartition() throws IOException { } @Test - void testListPartitionLocationNotExists() { + void testListPartitionMissingFilesFailure() { String partitionUri = tempTableLocation.resolve("id=1").toUri().toString(); - List dataFiles = - TableMigrationUtil.listPartition( - PARTITION, partitionUri, FORMAT, SPEC, CONF, MetricsConfig.getDefault(), null); - assertThat(dataFiles) - .as("List partition of which location does not exist should return 0 DataFile") - .isEmpty(); + assertThatThrownBy( + () -> + TableMigrationUtil.listPartition( + PARTITION, partitionUri, FORMAT, SPEC, CONF, MetricsConfig.getDefault(), null)) + .hasMessageContaining("Unable to list files in partition") + .isInstanceOf(RuntimeException.class) + .hasRootCauseInstanceOf(FileNotFoundException.class); } private static void writePartitionFile(File outputDir) throws IOException { diff --git a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java index 1998f1c61a19..9009200edab7 100644 --- a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java +++ b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java @@ -20,6 +20,7 @@ import static org.apache.spark.sql.functions.col; +import java.io.FileNotFoundException; import java.io.IOException; import java.io.Serializable; import java.net.URI; @@ -105,6 +106,8 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import scala.Function2; import scala.Option; import scala.Some; @@ -120,6 +123,7 @@ * https://github.com/apache/iceberg/blob/apache-iceberg-0.8.0-incubating/spark/src/main/scala/org/apache/iceberg/spark/SparkTableUtil.scala */ public class SparkTableUtil { + private static final Logger LOG = LoggerFactory.getLogger(SparkTableUtil.class); private static final String DUPLICATE_FILE_MESSAGE = "Cannot complete import because data files " @@ -292,34 +296,26 @@ private static List listPartition( SerializableConfiguration conf, MetricsConfig metricsConfig, NameMapping mapping, - int parallelism) { - return TableMigrationUtil.listPartition( - partition.values, - partition.uri, - partition.format, - spec, - conf.get(), - metricsConfig, - mapping, - parallelism); - } - - private static List listPartition( - SparkPartition partition, - PartitionSpec spec, - SerializableConfiguration conf, - MetricsConfig metricsConfig, - NameMapping mapping, + boolean ignoreMissingFiles, ExecutorService service) { - return TableMigrationUtil.listPartition( - partition.values, - partition.uri, - partition.format, - spec, - conf.get(), - metricsConfig, - mapping, - service); + try { + return TableMigrationUtil.listPartition( + partition.values, + partition.uri, + partition.format, + spec, + conf.get(), + metricsConfig, + mapping, + service); + } catch (RuntimeException e) { + if (ignoreMissingFiles && e.getCause() instanceof FileNotFoundException) { + LOG.warn("Ignoring FileNotFoundException when listing partition of {}", partition, e); + return Collections.emptyList(); + } else { + throw e; + } + } } private static SparkPartition toSparkPartition( @@ -524,6 +520,43 @@ public static void importSparkTable( Map partitionFilter, boolean checkDuplicateFiles, ExecutorService service) { + importSparkTable( + spark, + sourceTableIdent, + targetTable, + stagingDir, + partitionFilter, + checkDuplicateFiles, + false, + service); + } + + /** + * Import files from an existing Spark table to an Iceberg table. + * + *

The import uses the Spark session to get table metadata. It assumes no operation is going on + * the original and target table and thus is not thread-safe. + * + * @param spark a Spark session + * @param sourceTableIdent an identifier of the source Spark table + * @param targetTable an Iceberg table where to import the data + * @param stagingDir a staging directory to store temporary manifest files + * @param partitionFilter only import partitions whose values match those in the map, can be + * partially defined + * @param checkDuplicateFiles if true, throw exception if import results in a duplicate data file + * @param ignoreMissingFiles if true, ignore {@link FileNotFoundException} when running {@link + * #listPartition} for the Spark partitions + * @param service executor service to use for file reading + */ + public static void importSparkTable( + SparkSession spark, + TableIdentifier sourceTableIdent, + Table targetTable, + String stagingDir, + Map partitionFilter, + boolean checkDuplicateFiles, + boolean ignoreMissingFiles, + ExecutorService service) { SessionCatalog catalog = spark.sessionState().catalog(); String db = @@ -560,6 +593,7 @@ public static void importSparkTable( spec, stagingDir, checkDuplicateFiles, + ignoreMissingFiles, service); } } @@ -744,6 +778,32 @@ public static void importSparkPartitions( String stagingDir, boolean checkDuplicateFiles, ExecutorService service) { + importSparkPartitions( + spark, partitions, targetTable, spec, stagingDir, checkDuplicateFiles, false, service); + } + + /** + * Import files from given partitions to an Iceberg table. + * + * @param spark a Spark session + * @param partitions partitions to import + * @param targetTable an Iceberg table where to import the data + * @param spec a partition spec + * @param stagingDir a staging directory to store temporary manifest files + * @param checkDuplicateFiles if true, throw exception if import results in a duplicate data file + * @param ignoreMissingFiles if true, ignore {@link FileNotFoundException} when running {@link + * #listPartition} for the Spark partitions + * @param service executor service to use for file reading + */ + public static void importSparkPartitions( + SparkSession spark, + List partitions, + Table targetTable, + PartitionSpec spec, + String stagingDir, + boolean checkDuplicateFiles, + boolean ignoreMissingFiles, + ExecutorService service) { Configuration conf = spark.sessionState().newHadoopConf(); SerializableConfiguration serializableConf = new SerializableConfiguration(conf); int listingParallelism = @@ -771,6 +831,7 @@ public static void importSparkPartitions( serializableConf, metricsConfig, nameMapping, + ignoreMissingFiles, service) .iterator(), Encoders.javaSerialization(DataFile.class)); diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java index 5d8b03b0d527..d9c42a07b853 100644 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java +++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java @@ -20,11 +20,8 @@ import static org.assertj.core.api.Assertions.assertThat; -import java.io.File; import java.io.IOException; import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Files; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.apache.iceberg.ParameterizedTestExtension; @@ -68,22 +65,4 @@ public void testSnapshotWithParallelTasks() throws IOException { .execute(); assertThat(snapshotThreadsIndex.get()).isEqualTo(2); } - - @Test - public void testSnapshotWithEmptyPartitionOrLocationMissingPartition() throws IOException { - File tableLocation = temp.newFolder(); - sql( - "CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet PARTITIONED BY (bucket string) LOCATION '%s'", - SOURCE_NAME, tableLocation.toURI().toString()); - sql("ALTER TABLE %s ADD PARTITION (bucket = 'foo')", SOURCE_NAME); - sql("ALTER TABLE %s ADD PARTITION (bucket = 'bar')", SOURCE_NAME); - - // Delete the 2nd partition location - Path partitionLocation = tableLocation.toPath().resolve("bucket=bar"); - Files.delete(partitionLocation); - - SparkActions.get().snapshotTable(SOURCE_NAME).as(tableName).execute(); - assertThat(scalarSql("SELECT count(*) FROM %s.partitions", selectTarget())) - .as("Should have 0 partition after snapshot") - .isEqualTo(0L); } diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index fd644cc49ca8..0ce107f35134 100644 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -26,11 +26,14 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; import java.io.UncheckedIOException; +import java.nio.file.Path; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.util.Arrays; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -2170,6 +2173,99 @@ public void testTableWithInt96Timestamp() throws IOException { } } + @Test + public void testSparkTableWithMissingFilesFailure() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); + Table table = createTable(tableIdentifier, SCHEMA, SPEC); + + File parquetTablePath = temp.newFolder("table_missing_files"); + String parquetTableLocation = parquetTablePath.toURI().toString(); + spark.sql( + String.format( + "CREATE TABLE parquet_table (data string, id int) " + + "USING parquet PARTITIONED BY (id) LOCATION '%s'", + parquetTableLocation)); + + List records = + Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); + + Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); + inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + + // Add a Spark partition of which location is missing + spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); + Path partitionLocationPath = parquetTablePath.toPath().resolve("id=1234"); + java.nio.file.Files.delete(partitionLocationPath); + + NameMapping mapping = MappingUtil.create(table.schema()); + String mappingJson = NameMappingParser.toJson(mapping); + + table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); + + String stagingLocation = table.location() + "/metadata"; + + assertThatThrownBy( + () -> + SparkTableUtil.importSparkTable( + spark, + new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"), + table, + stagingLocation)) + .hasMessageContaining("Unable to list files in partition") + .isInstanceOf(SparkException.class) + .hasRootCauseInstanceOf(FileNotFoundException.class); + } + + @Test + public void testSparkTableWithIgnoreMissingFilesEnabled() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); + Table table = createTable(tableIdentifier, SCHEMA, SPEC); + + File parquetTableDir = temp.newFolder("table_missing_files"); + String parquetTableLocation = parquetTableDir.toURI().toString(); + spark.sql( + String.format( + "CREATE TABLE parquet_table (data string, id int) " + + "USING parquet PARTITIONED BY (id) LOCATION '%s'", + parquetTableLocation)); + + List records = + Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); + + Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); + inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + + // Add a Spark partition of which location is missing + spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); + Path partitionLocationPath = parquetTableDir.toPath().resolve("id=1234"); + java.nio.file.Files.delete(partitionLocationPath); + + NameMapping mapping = MappingUtil.create(table.schema()); + String mappingJson = NameMappingParser.toJson(mapping); + + table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); + + String stagingLocation = table.location() + "/metadata"; + + SparkTableUtil.importSparkTable( + spark, + new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"), + table, + stagingLocation, + Collections.emptyMap(), + false, + true, + SparkTableUtil.migrationService(1)); + + List partitionsTableRows = + spark + .read() + .format("iceberg") + .load(loadLocation(tableIdentifier, "partitions")) + .collectAsList(); + assertThat(partitionsTableRows).as("Partitions table should have 2 rows").hasSize(2); + } + private void testWithFilter(String filterExpr, TableIdentifier tableIdentifier) { List expected = spark.table("parquet_table").select("tmp_col").filter(filterExpr).collectAsList(); diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java index 67fc9a56a671..e0e969a0a0fe 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java @@ -20,6 +20,7 @@ import static org.apache.spark.sql.functions.col; +import java.io.FileNotFoundException; import java.io.IOException; import java.io.Serializable; import java.net.URI; @@ -105,6 +106,8 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import scala.Function2; import scala.Option; import scala.Some; @@ -120,6 +123,7 @@ * https://github.com/apache/iceberg/blob/apache-iceberg-0.8.0-incubating/spark/src/main/scala/org/apache/iceberg/spark/SparkTableUtil.scala */ public class SparkTableUtil { + private static final Logger LOG = LoggerFactory.getLogger(SparkTableUtil.class); private static final String DUPLICATE_FILE_MESSAGE = "Cannot complete import because data files " @@ -292,34 +296,26 @@ private static List listPartition( SerializableConfiguration conf, MetricsConfig metricsConfig, NameMapping mapping, - int parallelism) { - return TableMigrationUtil.listPartition( - partition.values, - partition.uri, - partition.format, - spec, - conf.get(), - metricsConfig, - mapping, - parallelism); - } - - private static List listPartition( - SparkPartition partition, - PartitionSpec spec, - SerializableConfiguration conf, - MetricsConfig metricsConfig, - NameMapping mapping, + boolean ignoreMissingFiles, ExecutorService service) { - return TableMigrationUtil.listPartition( - partition.values, - partition.uri, - partition.format, - spec, - conf.get(), - metricsConfig, - mapping, - service); + try { + return TableMigrationUtil.listPartition( + partition.values, + partition.uri, + partition.format, + spec, + conf.get(), + metricsConfig, + mapping, + service); + } catch (RuntimeException e) { + if (ignoreMissingFiles && e.getCause() instanceof FileNotFoundException) { + LOG.warn("Ignoring FileNotFoundException when listing partition of {}", partition, e); + return Collections.emptyList(); + } else { + throw e; + } + } } private static SparkPartition toSparkPartition( @@ -528,6 +524,45 @@ public static void importSparkTable( Map partitionFilter, boolean checkDuplicateFiles, ExecutorService service) { + importSparkTable( + spark, + sourceTableIdent, + targetTable, + stagingDir, + partitionFilter, + checkDuplicateFiles, + false, + service); + } + + /** + * Import files from an existing Spark table to an Iceberg table. + * + *

The import uses the Spark session to get table metadata. It assumes no operation is going on + * the original and target table and thus is not thread-safe. + * + * @param spark a Spark session + * @param sourceTableIdent an identifier of the source Spark table + * @param targetTable an Iceberg table where to import the data + * @param stagingDir a staging directory to store temporary manifest files + * @param partitionFilter only import partitions whose values match those in the map, can be + * partially defined + * @param checkDuplicateFiles if true, throw exception if import results in a duplicate data file + * @param ignoreMissingFiles if true, ignore {@link FileNotFoundException} when running {@link + * #listPartition} for the Spark partitions + * @param service executor service to use for file reading. If null, file reading will be + * performed on the current thread. If non-null, the provided ExecutorService will be shutdown + * within this method after file reading is complete. + */ + public static void importSparkTable( + SparkSession spark, + TableIdentifier sourceTableIdent, + Table targetTable, + String stagingDir, + Map partitionFilter, + boolean checkDuplicateFiles, + boolean ignoreMissingFiles, + ExecutorService service) { SessionCatalog catalog = spark.sessionState().catalog(); String db = @@ -564,6 +599,7 @@ public static void importSparkTable( spec, stagingDir, checkDuplicateFiles, + ignoreMissingFiles, service); } } @@ -750,6 +786,34 @@ public static void importSparkPartitions( String stagingDir, boolean checkDuplicateFiles, ExecutorService service) { + importSparkPartitions( + spark, partitions, targetTable, spec, stagingDir, checkDuplicateFiles, false, service); + } + + /** + * Import files from given partitions to an Iceberg table. + * + * @param spark a Spark session + * @param partitions partitions to import + * @param targetTable an Iceberg table where to import the data + * @param spec a partition spec + * @param stagingDir a staging directory to store temporary manifest files + * @param checkDuplicateFiles if true, throw exception if import results in a duplicate data file + * @param ignoreMissingFiles if true, ignore {@link FileNotFoundException} when running {@link + * #listPartition} for the Spark partitions + * @param service executor service to use for file reading. If null, file reading will be + * performed on the current thread. If non-null, the provided ExecutorService will be shutdown + * within this method after file reading is complete. + */ + public static void importSparkPartitions( + SparkSession spark, + List partitions, + Table targetTable, + PartitionSpec spec, + String stagingDir, + boolean checkDuplicateFiles, + boolean ignoreMissingFiles, + ExecutorService service) { Configuration conf = spark.sessionState().newHadoopConf(); SerializableConfiguration serializableConf = new SerializableConfiguration(conf); int listingParallelism = @@ -777,6 +841,7 @@ public static void importSparkPartitions( serializableConf, metricsConfig, nameMapping, + ignoreMissingFiles, service) .iterator(), Encoders.javaSerialization(DataFile.class)); diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java index f330fe07fc35..d9c42a07b853 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java @@ -22,7 +22,6 @@ import java.io.IOException; import java.nio.file.Files; -import java.nio.file.Path; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.apache.iceberg.ParameterizedTestExtension; @@ -66,23 +65,4 @@ public void testSnapshotWithParallelTasks() throws IOException { .execute(); assertThat(snapshotThreadsIndex.get()).isEqualTo(2); } - - @TestTemplate - public void testSnapshotWithEmptyPartitionOrLocationMissingPartition() throws IOException { - Path tableLocationPath = Files.createTempDirectory(temp, "junit"); - sql( - "CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet PARTITIONED BY (bucket string) LOCATION '%s'", - SOURCE_NAME, tableLocationPath.toFile().toString()); - sql("ALTER TABLE %s ADD PARTITION (bucket = 'foo')", SOURCE_NAME); - sql("ALTER TABLE %s ADD PARTITION (bucket = 'bar')", SOURCE_NAME); - - // Delete the 2nd partition location - Path partitionLocationPath = tableLocationPath.resolve("bucket=bar"); - Files.delete(partitionLocationPath); - - SparkActions.get().snapshotTable(SOURCE_NAME).as(tableName).execute(); - assertThat(scalarSql("SELECT count(*) FROM %s.partitions", selectTarget())) - .as("Should have 0 partition after snapshot") - .isEqualTo(0L); - } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 7d38cf55c98c..daeb0430a6fb 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -26,12 +26,14 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; import java.io.UncheckedIOException; import java.nio.file.Path; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.util.Arrays; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -2196,6 +2198,99 @@ public void testTableWithInt96Timestamp() throws IOException { } } + @Test + public void testSparkTableWithMissingFilesFailure() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); + Table table = createTable(tableIdentifier, SCHEMA, SPEC); + + File parquetTableDir = temp.resolve("table_missing_files").toFile(); + String parquetTableLocation = parquetTableDir.toURI().toString(); + spark.sql( + String.format( + "CREATE TABLE parquet_table (data string, id int) " + + "USING parquet PARTITIONED BY (id) LOCATION '%s'", + parquetTableLocation)); + + List records = + Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); + + Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); + inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + + // Add a Spark partition of which location is missing + spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); + Path partitionLocationPath = parquetTableDir.toPath().resolve("id=1234"); + java.nio.file.Files.delete(partitionLocationPath); + + NameMapping mapping = MappingUtil.create(table.schema()); + String mappingJson = NameMappingParser.toJson(mapping); + + table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); + + String stagingLocation = table.location() + "/metadata"; + + assertThatThrownBy( + () -> + SparkTableUtil.importSparkTable( + spark, + new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"), + table, + stagingLocation)) + .hasMessageContaining("Unable to list files in partition") + .isInstanceOf(SparkException.class) + .hasRootCauseInstanceOf(FileNotFoundException.class); + } + + @Test + public void testSparkTableWithIgnoreMissingFilesEnabled() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); + Table table = createTable(tableIdentifier, SCHEMA, SPEC); + + File parquetTableDir = temp.resolve("table_missing_files").toFile(); + String parquetTableLocation = parquetTableDir.toURI().toString(); + spark.sql( + String.format( + "CREATE TABLE parquet_table (data string, id int) " + + "USING parquet PARTITIONED BY (id) LOCATION '%s'", + parquetTableLocation)); + + List records = + Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); + + Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); + inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + + // Add a Spark partition of which location is missing + spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); + Path partitionLocationPath = parquetTableDir.toPath().resolve("id=1234"); + java.nio.file.Files.delete(partitionLocationPath); + + NameMapping mapping = MappingUtil.create(table.schema()); + String mappingJson = NameMappingParser.toJson(mapping); + + table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); + + String stagingLocation = table.location() + "/metadata"; + + SparkTableUtil.importSparkTable( + spark, + new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"), + table, + stagingLocation, + Collections.emptyMap(), + false, + true, + SparkTableUtil.migrationService(1)); + + List partitionsTableRows = + spark + .read() + .format("iceberg") + .load(loadLocation(tableIdentifier, "partitions")) + .collectAsList(); + assertThat(partitionsTableRows).as("Partitions table should have 2 rows").hasSize(2); + } + private void testWithFilter(String filterExpr, TableIdentifier tableIdentifier) { List expected = spark.table("parquet_table").select("tmp_col").filter(filterExpr).collectAsList(); From c63beab51ae0803f6ee48331bdbb3bf9c4faf76e Mon Sep 17 00:00:00 2001 From: Xi Chen Date: Sat, 3 May 2025 22:49:03 +0800 Subject: [PATCH 5/5] Address comments --- .../iceberg/data/TestTableMigrationUtil.java | 2 +- .../source/TestIcebergSourceTablesBase.java | 21 ++++++------------- .../source/TestIcebergSourceTablesBase.java | 21 ++++++------------- 3 files changed, 13 insertions(+), 31 deletions(-) diff --git a/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java b/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java index 9fcb2b077883..2aa21b8674ce 100644 --- a/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java +++ b/data/src/test/java/org/apache/iceberg/data/TestTableMigrationUtil.java @@ -89,7 +89,7 @@ void testListPartitionMissingFilesFailure() { () -> TableMigrationUtil.listPartition( PARTITION, partitionUri, FORMAT, SPEC, CONF, MetricsConfig.getDefault(), null)) - .hasMessageContaining("Unable to list files in partition") + .hasMessageContaining("Unable to list files in partition: " + partitionUri) .isInstanceOf(RuntimeException.class) .hasRootCauseInstanceOf(FileNotFoundException.class); } diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 0ce107f35134..19e8e8146d86 100644 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -2174,7 +2174,7 @@ public void testTableWithInt96Timestamp() throws IOException { } @Test - public void testSparkTableWithMissingFilesFailure() throws IOException { + public void testImportSparkTableWithMissingFilesFailure() throws IOException { TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); Table table = createTable(tableIdentifier, SCHEMA, SPEC); @@ -2190,18 +2190,13 @@ public void testSparkTableWithMissingFilesFailure() throws IOException { Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); - inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + inputDF.write().mode("overwrite").insertInto("parquet_table"); // Add a Spark partition of which location is missing spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); Path partitionLocationPath = parquetTablePath.toPath().resolve("id=1234"); java.nio.file.Files.delete(partitionLocationPath); - NameMapping mapping = MappingUtil.create(table.schema()); - String mappingJson = NameMappingParser.toJson(mapping); - - table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); - String stagingLocation = table.location() + "/metadata"; assertThatThrownBy( @@ -2211,13 +2206,14 @@ public void testSparkTableWithMissingFilesFailure() throws IOException { new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"), table, stagingLocation)) - .hasMessageContaining("Unable to list files in partition") + .hasMessageContaining( + "Unable to list files in partition: " + partitionLocationPath.toFile().toURI()) .isInstanceOf(SparkException.class) .hasRootCauseInstanceOf(FileNotFoundException.class); } @Test - public void testSparkTableWithIgnoreMissingFilesEnabled() throws IOException { + public void testImportSparkTableWithIgnoreMissingFilesEnabled() throws IOException { TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); Table table = createTable(tableIdentifier, SCHEMA, SPEC); @@ -2233,18 +2229,13 @@ public void testSparkTableWithIgnoreMissingFilesEnabled() throws IOException { Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); - inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + inputDF.write().mode("overwrite").insertInto("parquet_table"); // Add a Spark partition of which location is missing spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); Path partitionLocationPath = parquetTableDir.toPath().resolve("id=1234"); java.nio.file.Files.delete(partitionLocationPath); - NameMapping mapping = MappingUtil.create(table.schema()); - String mappingJson = NameMappingParser.toJson(mapping); - - table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); - String stagingLocation = table.location() + "/metadata"; SparkTableUtil.importSparkTable( diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index daeb0430a6fb..225c9540a05e 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -2199,7 +2199,7 @@ public void testTableWithInt96Timestamp() throws IOException { } @Test - public void testSparkTableWithMissingFilesFailure() throws IOException { + public void testImportSparkTableWithMissingFilesFailure() throws IOException { TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); Table table = createTable(tableIdentifier, SCHEMA, SPEC); @@ -2215,18 +2215,13 @@ public void testSparkTableWithMissingFilesFailure() throws IOException { Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); - inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + inputDF.write().mode("overwrite").insertInto("parquet_table"); // Add a Spark partition of which location is missing spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); Path partitionLocationPath = parquetTableDir.toPath().resolve("id=1234"); java.nio.file.Files.delete(partitionLocationPath); - NameMapping mapping = MappingUtil.create(table.schema()); - String mappingJson = NameMappingParser.toJson(mapping); - - table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); - String stagingLocation = table.location() + "/metadata"; assertThatThrownBy( @@ -2236,13 +2231,14 @@ public void testSparkTableWithMissingFilesFailure() throws IOException { new org.apache.spark.sql.catalyst.TableIdentifier("parquet_table"), table, stagingLocation)) - .hasMessageContaining("Unable to list files in partition") + .hasMessageContaining( + "Unable to list files in partition: " + partitionLocationPath.toFile().toURI()) .isInstanceOf(SparkException.class) .hasRootCauseInstanceOf(FileNotFoundException.class); } @Test - public void testSparkTableWithIgnoreMissingFilesEnabled() throws IOException { + public void testImportSparkTableWithIgnoreMissingFilesEnabled() throws IOException { TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test"); Table table = createTable(tableIdentifier, SCHEMA, SPEC); @@ -2258,18 +2254,13 @@ public void testSparkTableWithIgnoreMissingFilesEnabled() throws IOException { Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b")); Dataset inputDF = spark.createDataFrame(records, SimpleRecord.class); - inputDF.select("data", "id").write().mode("overwrite").insertInto("parquet_table"); + inputDF.write().mode("overwrite").insertInto("parquet_table"); // Add a Spark partition of which location is missing spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)"); Path partitionLocationPath = parquetTableDir.toPath().resolve("id=1234"); java.nio.file.Files.delete(partitionLocationPath); - NameMapping mapping = MappingUtil.create(table.schema()); - String mappingJson = NameMappingParser.toJson(mapping); - - table.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, mappingJson).commit(); - String stagingLocation = table.location() + "/metadata"; SparkTableUtil.importSparkTable(