Default to parquet in AddFiles for extension less files.

.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
 }

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<ProcessResult> 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<ProcessResult> 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 {

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java

@@ -161,29 +161,60 @@ private static void count(ReadResult result) {
   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()) {

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.<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);