(iceberg) branch main updated: Parquet: Log corrupted parquet filenames to trace bad nodes that may have written them. (#13108)

[pv...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new 91dff9886e Parquet: Log corrupted parquet filenames to trace bad nodes that may have written them. (#13108)
91dff9886e is described below

commit 91dff9886e6e6c494f6a970129f4c08487c98a0d
Author: Dhruv Pratap <dp...@netflix.com>
AuthorDate: Thu May 22 06:39:47 2025 -0500

    Parquet: Log corrupted parquet filenames to trace bad nodes that may have written them. (#13108)
---
 .../org/apache/iceberg/parquet/ParquetReader.java  | 38 +++++++++++++++-------
 1 file changed, 27 insertions(+), 11 deletions(-)

diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetReader.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetReader.java
index e8ee90fdeb..0498adc442 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetReader.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetReader.java
@@ -32,7 +32,10 @@ import org.apache.iceberg.mapping.NameMapping;
 import org.apache.parquet.ParquetReadOptions;
 import org.apache.parquet.column.page.PageReadStore;
 import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.schema.MessageType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class ParquetReader<T> extends CloseableGroup implements CloseableIterable<T> {
   private final InputFile input;
@@ -94,6 +97,8 @@ public class ParquetReader<T> extends CloseableGroup implements CloseableIterabl
   }
 
   private static class FileIterator<T> implements CloseableIterator<T> {
+    private static final Logger LOG = LoggerFactory.getLogger(FileIterator.class);
+
     private final ParquetFileReader reader;
     private final boolean[] shouldSkip;
     private final ParquetValueReader<T> model;
@@ -120,18 +125,29 @@ public class ParquetReader<T> extends CloseableGroup implements CloseableIterabl
 
     @Override
     public T next() {
-      if (valuesRead >= nextRowGroupStart) {
-        advance();
-      }
-
-      if (reuseContainers) {
-        this.last = model.read(last);
-      } else {
-        this.last = model.read(null);
+      try {
+        if (valuesRead >= nextRowGroupStart) {
+          advance();
+        }
+
+        if (reuseContainers) {
+          this.last = model.read(last);
+        } else {
+          this.last = model.read(null);
+        }
+
+        valuesRead += 1;
+
+        return last;
+      } catch (ParquetDecodingException e) {
+        if (reader != null) {
+          // Knowing the exact parquet file is essential for tracing bad nodes
+          // that produced the corrupt file, parquet lib doesn't do this today.
+          LOG.error("Error decoding Parquet file {}", reader.getFile(), e);
+        }
+
+        throw e;
       }
-      valuesRead += 1;
-
-      return last;
     }
 
     private void advance() {