From f3638db4302bff2258cb7ff1feca1afdff458930 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 10:54:32 -0400 Subject: [PATCH 1/2] Assume files are parquet by default if no extension --- .../IO_Iceberg_Integration_Tests.json | 2 +- .../apache/beam/sdk/io/iceberg/AddFiles.java | 35 ++++-- .../beam/sdk/io/iceberg/ReadFooterSchema.java | 52 ++++++--- .../beam/sdk/io/iceberg/AddFilesIT.java | 80 ++++++++++++++ .../beam/sdk/io/iceberg/AddFilesTest.java | 100 ++++++++++++++++++ .../sdk/io/iceberg/ReadFooterSchemaTest.java | 19 ++++ 6 files changed, 265 insertions(+), 23 deletions(-) diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index e1290fa50c32..b73af5e61a43 100644 --- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json +++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 11 + "modification": 1 } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index cb54183e0a07..c72479aa39b3 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -468,6 +468,8 @@ public ConvertToDataFile( static final String PREFIX_ERROR = "File path did not start with the specified prefix"; private static final String UNKNOWN_FORMAT_ERROR = "Could not determine the file's format"; + static final String NOT_PARQUET_ERROR = + UNKNOWN_FORMAT_ERROR + ": the file name has no extension and the file is not Parquet: "; static final String UNKNOWN_PARTITION_ERROR = "Could not determine the file's partition: "; static final String UNREADABLE_SCHEMA_ERROR = "Could not read the file's schema: "; static final String UNCOVERED_ERROR = "Table schema does not cover the file after refresh: "; @@ -626,10 +628,21 @@ private Callable createProcessTask( return () -> { FileFormat format; - try { - format = inferFormat(filePath); - } catch (UnknownFormatException e) { - return errorResult(filePath, UNKNOWN_FORMAT_ERROR, timestamp, window, paneInfo); + @Nullable ParquetMetadata parquetFooter = null; + if (hasNoExtension(filePath)) { + try { + parquetFooter = ParquetFooters.read(filePath); + } catch (Exception e) { + return errorResult( + filePath, NOT_PARQUET_ERROR + errorMessage(e), timestamp, window, paneInfo); + } + format = FileFormat.PARQUET; + } else { + try { + format = inferFormat(filePath); + } catch (UnknownFormatException e) { + return errorResult(filePath, UNKNOWN_FORMAT_ERROR, timestamp, window, paneInfo); + } } // ---- Infrastructure phase. Failures propagate so the runner retries the bundle; @@ -670,8 +683,7 @@ private Callable createProcessTask( } // ---- Per-file phase: every failure below is one error row, never a failed bundle. - @Nullable ParquetMetadata parquetFooter = null; - if (format.equals(FileFormat.PARQUET)) { + if (format.equals(FileFormat.PARQUET) && parquetFooter == null) { try { parquetFooter = ParquetFooters.read(filePath); } catch (Exception e) { @@ -1369,7 +1381,7 @@ static String errorMessage(Throwable e) { return e.getMessage() != null ? e.getMessage() : e.toString(); } - /** Tries to infer other file formats. Defaults to Parquet. */ + /** Attempt to infer file format from extension. */ public static FileFormat inferFormat(String path) { String lowerPath = path.toLowerCase(); @@ -1384,6 +1396,15 @@ public static FileFormat inferFormat(String path) { } } + /** + * A name with no extension, as Hive and other engines write by default, says nothing about the + * format, so the file is read to see whether it is Parquet. A name with a dot names some format. + */ + static boolean hasNoExtension(String path) { + String name = path.substring(path.lastIndexOf('/') + 1); + return !name.isEmpty() && name.indexOf('.') < 0; + } + static class UnknownFormatException extends IllegalArgumentException {} static class UnknownPartitionException extends IllegalStateException { diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java index 389f03490fe4..98d3375d1ea0 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java @@ -161,29 +161,51 @@ private static void count(ReadResult result) { private Callable createReadTask( String filePath, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) { return () -> { - FileFormat format; - try { - format = AddFiles.inferFormat(filePath); - } catch (AddFiles.UnknownFormatException e) { - return new ReadResult(dryRunMarker(UNREADABLE_KEY), false, timestamp, window, paneInfo); - } - if (!format.equals(FileFormat.PARQUET)) { - return new ReadResult( - dryRunMarker(UNCHECKED_FORMAT_KEY), false, timestamp, window, paneInfo); + ParquetMetadata footer; + if (AddFiles.hasNoExtension(filePath)) { + try { + footer = ParquetFooters.read(filePath); + } catch (Exception e) { + return unknownFormat(timestamp, window, paneInfo); + } + } else { + FileFormat format; + try { + format = AddFiles.inferFormat(filePath); + } catch (AddFiles.UnknownFormatException e) { + return unknownFormat(timestamp, window, paneInfo); + } + if (!format.equals(FileFormat.PARQUET)) { + return new ReadResult( + dryRunMarker(UNCHECKED_FORMAT_KEY), false, timestamp, window, paneInfo); + } + try { + footer = ParquetFooters.read(filePath); + } catch (Exception e) { + return unreadableFooter(filePath, e, timestamp, window, paneInfo); + } } try { - ParquetMetadata footer = ParquetFooters.read(filePath); return new ReadResult(FileSchemas.schemaGroup(footer), false, timestamp, window, paneInfo); } catch (Exception e) { - LOG.warn( - "Could not read the footer of {}; the file will not contribute to schema inference: {}", - filePath, - AddFiles.errorMessage(e)); - return new ReadResult(dryRunMarker(UNREADABLE_KEY), true, timestamp, window, paneInfo); + return unreadableFooter(filePath, e, timestamp, window, paneInfo); } }; } + private ReadResult unknownFormat(Instant timestamp, BoundedWindow window, PaneInfo paneInfo) { + return new ReadResult(dryRunMarker(UNREADABLE_KEY), false, timestamp, window, paneInfo); + } + + private ReadResult unreadableFooter( + String filePath, Exception e, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) { + LOG.warn( + "Could not read the footer of {}; the file will not contribute to schema inference: {}", + filePath, + AddFiles.errorMessage(e)); + return new ReadResult(dryRunMarker(UNREADABLE_KEY), true, timestamp, window, paneInfo); + } + /** A dry run counts files that contribute no schema; a real run handles them at registration. */ private CollectDistinctSchemas.@Nullable SchemaGroup dryRunMarker(String key) { if (!config.getDryRun()) { diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index a1bb79431753..276ea4b0ee90 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -33,6 +33,8 @@ import com.google.api.client.http.HttpResponseException; import com.google.api.services.storage.model.StorageObject; import com.google.cloud.storage.Blob; +import com.google.cloud.storage.BlobId; +import com.google.cloud.storage.BlobInfo; import com.google.cloud.storage.Notification; import com.google.cloud.storage.NotificationInfo; import com.google.cloud.storage.Storage; @@ -698,6 +700,84 @@ public void testBatchParquetImportWithSchemaEvolution() throws IOException { s.run().waitUntilFinish(); } + /** + * Hive names Parquet files without an extension (000000_0), and Spark and Hadoop leave a _SUCCESS + * marker beside them. The files register and read back, BigQuery included; the marker is one + * error row and must not seed the table. + */ + @Test + public void testBatchImportOfParquetFilesWithoutExtension() throws IOException { + String parquetDir = format("gs://%s/%s/", DATA_BUCKET, dirName); + String tempDir = format("gs://%s/%s-tmp/", DATA_BUCKET, dirName); + Pipeline q = Pipeline.create(); + org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA); + q.apply(Create.of(TEST_ROWS)) + .setRowSchema(ROW_SCHEMA) + .apply( + MapElements.into(TypeDescriptor.of(GenericRecord.class)) + .via(AvroUtils.getRowToGenericRecordFunction(avroSchema))) + .setCoder(AvroCoder.of(avroSchema)) + .apply( + FileIO.writeDynamic() + .by(record -> String.valueOf(record.get("id"))) + .via(ParquetIO.sink(avroSchema)) + .withNaming( + id -> + (window, pane, numShards, shardIndex, compression) -> + format("%06d_%d", Long.parseLong(id), shardIndex)) + .withTempDirectory(tempDir) + .to(parquetDir) + .withDestinationCoder(StringUtf8Coder.of())); + q.run().waitUntilFinish(); + String marker = parquetDir + "_SUCCESS"; + storage.create( + BlobInfo.newBuilder(BlobId.of(DATA_BUCKET, dirName + "/_SUCCESS")).build(), new byte[0]); + + GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); + List dataFiles = new ArrayList<>(); + for (StorageObject object : gcsUtil.listObjects(DATA_BUCKET, dirName + "/", null).getItems()) { + String path = format("gs://%s/%s", object.getBucket(), object.getName()); + if (!path.equals(marker)) { + dataFiles.add(path); + } + } + LOG.info("Extensionless files: {}", dataFiles); + assertEquals(TEST_ROWS.size(), dataFiles.size()); + for (String path : dataFiles) { + assertFalse(path, path.substring(path.lastIndexOf('/') + 1).contains(".")); + } + List inputs = new ArrayList<>(dataFiles); + inputs.add(marker); + + Pipeline p = Pipeline.create(); + PCollectionRowTuple tuple = + p.apply(Create.of(inputs)) + .apply( + new AddFiles( + IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(), + namespace + "." + destTableName, + null, + null, + null, + TABLE_PROPS, + null, + null)); + PAssert.that(tuple.get("errors")) + .satisfies( + rows -> { + Row row = Iterables.getOnlyElement(rows); + String error = String.valueOf(row.getString("error")); + LOG.info("Marker error row: {}", row); + assertEquals(marker, row.getString("file")); + assertTrue(error, error.startsWith(AddFiles.ConvertToDataFile.NOT_PARQUET_ERROR)); + return null; + }); + p.run().waitUntilFinish(); + + assertTrue(checkTableHasRegisteredParquetFiles(dataFiles)); + checkRecordsInDestinationTable(/* alsoCheckWithBigQueryIO= */ true); + } + private static void writeParquet(List rows, Schema schema, String dir, String tempDir) { Pipeline q = Pipeline.create(); org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(schema); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java index 0d5a41e0cc4b..ea1375457959 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java @@ -18,6 +18,7 @@ package org.apache.beam.sdk.io.iceberg; import static org.apache.beam.sdk.io.iceberg.AddFiles.ConvertToDataFile.FIELD_ID_ERROR; +import static org.apache.beam.sdk.io.iceberg.AddFiles.ConvertToDataFile.NOT_PARQUET_ERROR; import static org.apache.beam.sdk.io.iceberg.AddFiles.ConvertToDataFile.PREFIX_ERROR; import static org.apache.beam.sdk.io.iceberg.AddFiles.ConvertToDataFile.UNKNOWN_PARTITION_ERROR; import static org.apache.beam.sdk.io.iceberg.AddFiles.ConvertToDataFile.getPartitionFromMetrics; @@ -450,6 +451,105 @@ public void testUnknownFormatErrors() throws Exception { pipeline.run().waitUntilFinish(); } + /** Hive names its output files like {@code 000000_0}, with no extension. */ + @Test + public void testExtensionlessParquetFileRegisters() throws Exception { + catalog.createTable(tableId, icebergSchema); + String file = root + "000000_0"; + DataWriter writer = createWriter(file); + writer.write(record(1, "Mark", 20)); + writer.close(); + + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(file)) + .apply( + new AddFiles( + catalogConfig, tableId.toString(), null, null, null, null, null, null)); + PAssert.that(output.get("errors")).empty(); + + pipeline.run().waitUntilFinish(); + + List registered = new ArrayList<>(); + for (FileScanTask task : catalog.loadTable(tableId).newScan().planFiles()) { + registered.add(task.file()); + } + assertEquals(1, registered.size()); + assertEquals(file, registered.get(0).location()); + assertEquals(FileFormat.PARQUET, registered.get(0).format()); + assertEquals(1L, registered.get(0).recordCount()); + } + + /** + * A _SUCCESS marker has no extension either. It must become an error row rather than seed the + * missing table, where a failed footer read would fail the job. + */ + @Test + public void testExtensionlessFileThatIsNotParquetIsAnErrorRow() throws Exception { + File marker = temp.newFile("_SUCCESS"); + + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(marker.getAbsolutePath())) + .apply( + new AddFiles( + catalogConfig, tableId.toString(), null, null, null, null, null, null)); + assertNotParquetErrorFor(output, marker.getAbsolutePath()); + + pipeline.run().waitUntilFinish(); + + assertFalse("the marker must not create the table", catalog.tableExists(tableId)); + } + + @Test + public void testMissingTableIsCreatedFromAnExtensionlessFile() throws Exception { + String file = writeWithoutFieldIds("000000_0", icebergSchema, record(1, "a", 1)); + File marker = temp.newFile("_SUCCESS"); + + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(file, marker.getAbsolutePath())) + .apply(addFiles(ADDITIONS)); + assertNotParquetErrorFor(output, marker.getAbsolutePath()); + + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + + assertEquals(Arrays.asList("id=1 name=a age=1"), readRows("id", "name", "age")); + assertEquals( + "the marker is an unknown format, not a footer that failed to read", + 0, + counted(result, ReadFooterSchema.class, ReadFooterSchema.FOOTER_READ_ERRORS_COUNTER)); + } + + /** Only a name with no extension is left to its contents; inferFormat stays strict. */ + @Test + public void testOnlyAnExtensionlessNameIsReadForItsFormat() { + assertTrue(AddFiles.hasNoExtension("gs://b/wh/sales.db/orders/000000_0")); + assertTrue(AddFiles.hasNoExtension("gs://b/t/part-00000-1a2b3c-c000")); + assertFalse(AddFiles.hasNoExtension("gs://b/t/data.txt")); + assertFalse(AddFiles.hasNoExtension("gs://b/t/.hidden")); + assertFalse(AddFiles.hasNoExtension("gs://b/t/")); + assertEquals(FileFormat.PARQUET, AddFiles.inferFormat("gs://b/t/data.PARQUET")); + assertEquals(FileFormat.ORC, AddFiles.inferFormat("gs://b/t/data.orc")); + assertThrows( + AddFiles.UnknownFormatException.class, () -> AddFiles.inferFormat("gs://b/t/000000_0")); + assertThrows( + AddFiles.UnknownFormatException.class, () -> AddFiles.inferFormat("gs://b/t/data.txt")); + } + + private static void assertNotParquetErrorFor(PCollectionRowTuple output, String file) { + PAssert.that(output.get("errors")) + .satisfies( + rows -> { + Row row = Iterables.getOnlyElement(rows); + String error = String.valueOf(row.getString("error")); + assertEquals(file, row.getString("file")); + assertTrue(error, error.startsWith(NOT_PARQUET_ERROR)); + return null; + }); + } + @Test public void testPartitionPrefixErrors() throws Exception { // Drop unpartitioned table and create a partitioned one diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java index 4f17bc738400..07f9590eef82 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java @@ -164,6 +164,25 @@ public void testNonParquetEmitsNothing() throws IOException { pipeline.run(); } + /** Hive names its output files like this, with no extension. */ + @Test + public void testExtensionlessParquetEmitsSchema() throws IOException { + String file = writeParquet("000000_0", FLAT_SCHEMA, record(FLAT_SCHEMA, "id", 1)); + + assertSchemas(run(file), FLAT_SCHEMA); + + pipeline.run(); + } + + @Test + public void testExtensionlessNonParquetEmitsNothing() throws IOException { + String marker = writeBytes("_SUCCESS", new byte[0]); + + PAssert.that(run(marker)).empty(); + + pipeline.run(); + } + /** * The schema is emitted exactly as declared; which columns proved null-free travels beside it. */ From 84f2faf5f19ead8007c76afa62fb23613719a033 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 16:11:50 -0400 Subject: [PATCH 2/2] fix messge --- .../beam/sdk/io/iceberg/ReadFooterSchema.java | 11 +++- .../beam/sdk/io/iceberg/AddFilesTest.java | 57 +++++++++++++++++++ .../sdk/io/iceberg/ReadFooterSchemaTest.java | 33 +++++++++++ 3 files changed, 100 insertions(+), 1 deletion(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java index 98d3375d1ea0..f111ed4b8554 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java @@ -188,7 +188,7 @@ private Callable createReadTask( try { return new ReadResult(FileSchemas.schemaGroup(footer), false, timestamp, window, paneInfo); } catch (Exception e) { - return unreadableFooter(filePath, e, timestamp, window, paneInfo); + return unconvertibleSchema(filePath, e, timestamp, window, paneInfo); } }; } @@ -206,6 +206,15 @@ private ReadResult unreadableFooter( return new ReadResult(dryRunMarker(UNREADABLE_KEY), true, timestamp, window, paneInfo); } + private ReadResult unconvertibleSchema( + String filePath, Exception e, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) { + LOG.warn( + "Could not convert the schema of {}; the file will not contribute to schema inference: {}", + filePath, + AddFiles.errorMessage(e)); + return new ReadResult(dryRunMarker(UNREADABLE_KEY), false, timestamp, window, paneInfo); + } + /** A dry run counts files that contribute no schema; a real run handles them at registration. */ private CollectDistinctSchemas.@Nullable SchemaGroup dryRunMarker(String key) { if (!config.getDryRun()) { diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java index ea1375457959..578d18c539d2 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java @@ -1914,6 +1914,63 @@ public void testExtraColumnsRegisterWhenEvolutionDisabled() throws Exception { pipeline.run().waitUntilFinish(); } + /** + * Through the whole transform, so the pre-pass sees the file too: its footer reads, so it is not + * a footer read error, and it costs one error row, not the job. + */ + @Test + public void testUnconvertibleSchemaIsOneErrorRowWithEvolution() throws Exception { + catalog.createTable(tableId, icebergSchema); + File legacy = new File(temp.getRoot(), "legacy.parquet"); + writeLegacyRepeated(legacy); + String good = writeWithoutFieldIds("good.parquet", icebergSchema, record(1, "a", 1)); + + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(legacy.getAbsolutePath(), good)) + .apply(addFiles(ADDITIONS)); + PAssert.that(output.get("errors")) + .satisfies( + rows -> { + Row row = Iterables.getOnlyElement(rows); + assertEquals(legacy.getAbsolutePath(), row.getString("file")); + assertTrue( + String.valueOf(row.getString("error")), + String.valueOf(row.getString("error")) + .startsWith(AddFiles.ConvertToDataFile.UNREADABLE_SCHEMA_ERROR)); + return null; + }); + + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + + assertEquals(1, Iterables.size(catalog.loadTable(tableId).newScan().planFiles())); + assertEquals( + 0, counted(result, ReadFooterSchema.class, ReadFooterSchema.FOOTER_READ_ERRORS_COUNTER)); + } + + /** Readable Parquet whose unannotated repeated field Iceberg's converter rejects. */ + private static void writeLegacyRepeated(File file) throws IOException { + org.apache.parquet.schema.MessageType legacy = + org.apache.parquet.schema.Types.buildMessage() + .required(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32) + .named("id") + .repeated(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32) + .named("vals") + .named("root"); + try (org.apache.parquet.hadoop.ParquetWriter writer = + org.apache.parquet.hadoop.example.ExampleParquetWriter.builder( + new org.apache.hadoop.fs.Path(file.getAbsolutePath())) + .withType(legacy) + .build()) { + org.apache.parquet.example.data.Group group = + new org.apache.parquet.example.data.simple.SimpleGroupFactory(legacy).newGroup(); + group.add("id", 1); + group.add("vals", 2); + writer.write(group); + } + } + @Test public void testUnreadableSchemaRoutesToErrorsWithConverterMessage() throws Exception { catalog.createTable(tableId, icebergSchema); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java index 07f9590eef82..a6c178a520f2 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java @@ -269,6 +269,39 @@ private static long counter(PipelineResult result, String name) { return total; } + /** The footer reads, but its schema does not convert: the file contributes no schema. */ + @Test + public void testUnconvertibleSchemaEmitsNothing() throws IOException { + File file = new File(temporaryFolder.getRoot(), "legacy.parquet"); + writeLegacyRepeated(file); + + PAssert.that(run(file.getAbsolutePath())).empty(); + + pipeline.run(); + } + + /** Readable Parquet whose unannotated repeated field Iceberg's converter rejects. */ + private static void writeLegacyRepeated(File file) throws IOException { + org.apache.parquet.schema.MessageType legacy = + org.apache.parquet.schema.Types.buildMessage() + .required(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32) + .named("id") + .repeated(org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32) + .named("vals") + .named("root"); + try (org.apache.parquet.hadoop.ParquetWriter writer = + org.apache.parquet.hadoop.example.ExampleParquetWriter.builder( + new org.apache.hadoop.fs.Path(file.getAbsolutePath())) + .withType(legacy) + .build()) { + org.apache.parquet.example.data.Group group = + new org.apache.parquet.example.data.simple.SimpleGroupFactory(legacy).newGroup(); + group.add("id", 1); + group.add("vals", 2); + writer.write(group); + } + } + private PCollection run(String... paths) { return pipeline .apply(Create.of(Arrays.asList(paths)))