[LAMBERT-205]: Added logs for small file size debugging (apache#9)

* added logs for small file size debugging * added file name * reverted version.txt
adamyasharma2797 · Jul 19, 2024 · cf4ccf8 · cf4ccf8
1 parent 30279b8
commit cf4ccf8
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java b/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
@@ -40,8 +40,11 @@
 import org.apache.iceberg.util.StructProjection;
 import org.apache.iceberg.util.Tasks;
 import org.apache.iceberg.util.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class BaseTaskWriter<T> implements TaskWriter<T> {
+  private static final Logger LOG = LoggerFactory.getLogger(BaseTaskWriter.class);
   private final List<DataFile> completedDataFiles = Lists.newArrayList();
   private final List<DeleteFile> completedDeleteFiles = Lists.newArrayList();
   private final CharSequenceSet referencedDataFiles = CharSequenceSet.empty();
@@ -278,6 +281,7 @@ public void write(T record) throws IOException {
       this.currentRows++;
 
       if (shouldRollToNewFile()) {
+        LOG.info("Closing file: {} as it went beyond target size: {}", currentFile.encryptingOutputFile().location(), targetFileSize);
         closeCurrent();
         openCurrent();
       }
@@ -311,6 +315,7 @@ private boolean shouldRollToNewFile() {
     private void closeCurrent() throws IOException {
       if (currentWriter != null) {
         try {
+          LOG.info("Closing file: {}, File Size: {}, Row Count: {}, targetFileSize: {}", currentFile.encryptingOutputFile().location(), length(currentWriter), currentRows, targetFileSize);
           currentWriter.close();
 
           if (currentRows == 0L) {

diff --git a/core/src/main/java/org/apache/iceberg/io/PartitionedFanoutWriter.java b/core/src/main/java/org/apache/iceberg/io/PartitionedFanoutWriter.java
@@ -24,9 +24,12 @@
 import org.apache.iceberg.PartitionKey;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class PartitionedFanoutWriter<T> extends BaseTaskWriter<T> {
   private final Map<PartitionKey, RollingFileWriter> writers = Maps.newHashMap();
+  private static final Logger LOG = LoggerFactory.getLogger(PartitionedFanoutWriter.class);
 
   protected PartitionedFanoutWriter(
       PartitionSpec spec,
@@ -66,6 +69,8 @@ public void write(T row) throws IOException {
   @Override
   public void close() throws IOException {
     if (!writers.isEmpty()) {
+      LOG.info("Partition Writer Size: {}", writers.size());
+      LOG.info("Partition Writer Keys: {}", writers.keySet());
       for (PartitionKey key : writers.keySet()) {
         writers.get(key).close();
       }