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

Author: dhruv-pratap

parquet/src/main/java/org/apache/iceberg/parquet/ParquetReader.java

@@ -32,7 +32,10 @@
 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 CloseableIterator<T> iterator() {
   }
 
   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 boolean hasNext() {
 
     @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() {