This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 68e762ecadb Default to parquet in AddFiles for extension less files.
(#40614)
68e762ecadb is described below
commit 68e762ecadbc5e5e351f6881baab4fa11a71ff0c
Author: claudevdm <[email protected]>
AuthorDate: Mon Oct 19 17:54:40 2026 -0400
Default to parquet in AddFiles for extension less files. (#40614)
* Assume files are parquet by default if no extension
* fix messge
---
.../IO_Iceberg_Integration_Tests.json | 2 +-
.../org/apache/beam/sdk/io/iceberg/AddFiles.java | 35 ++++-
.../beam/sdk/io/iceberg/ReadFooterSchema.java | 61 ++++++--
.../org/apache/beam/sdk/io/iceberg/AddFilesIT.java | 80 +++++++++++
.../apache/beam/sdk/io/iceberg/AddFilesTest.java | 157 +++++++++++++++++++++
.../beam/sdk/io/iceberg/ReadFooterSchemaTest.java | 52 +++++++
6 files changed, 364 insertions(+), 23 deletions(-)
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index e1290fa50c3..b73af5e61a4 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 cb54183e0a0..c72479aa39b 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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
}
// ---- 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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
}
}
+ /**
+ * 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 389f03490fe..f111ed4b855 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,60 @@ class ReadFooterSchema extends DoFn<String,
CollectDistinctSchemas.SchemaGroup>
private Callable<ReadResult> 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 unconvertibleSchema(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);
+ }
+
+ 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/AddFilesIT.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
index a1bb7943175..276ea4b0ee9 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 static org.junit.Assert.assertTrue;
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 class AddFilesIT {
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.<String, GenericRecord>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<String> 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<String> 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<Row> 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 0d5a41e0cc4..578d18c539d 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 class AddFilesTest {
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<Record> 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<DataFile> 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
@@ -1814,6 +1914,63 @@ public class AddFilesTest {
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<org.apache.parquet.example.data.Group>
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 4f17bc73840..a6c178a520f 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 class ReadFooterSchemaTest {
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.
*/
@@ -250,6 +269,39 @@ public class ReadFooterSchemaTest {
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<org.apache.parquet.example.data.Group>
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<CollectDistinctSchemas.SchemaGroup> run(String... paths)
{
return pipeline
.apply(Create.of(Arrays.asList(paths)))
