Java Scan operator raw input reported wrong

Author: Dilli-Babu-Godari

presto-hive/src/main/java/com/facebook/presto/hive/orc/AggregatedOrcPageSource.java

@@ -53,6 +53,7 @@ public class AggregatedOrcPageSource
     private boolean completed;
     private long readTimeNanos;
     private long completedBytes;
+    private long decompressedBytes;
 
     public AggregatedOrcPageSource(List<HiveColumnHandle> columnHandles, Footer footer, TypeManager typeManager, StandardFunctionResolution functionResolution)
     {
@@ -74,6 +75,18 @@ public long getCompletedPositions()
         return 0;
     }
 
+    @Override
+    public long getDecompressedBytes()
+    {
+        return decompressedBytes;
+    }
+
+    @Override
+    public long getDecompressedPositions()
+    {
+        return 0;
+    }
+
     @Override
     public long getReadTimeNanos()
     {
@@ -126,7 +139,10 @@ else if (functionResolution.isMinFunction(functionHandle)) {
 
         completed = true;
         readTimeNanos += System.nanoTime() - start;
-        return new Page(batchSize, blocks);
+        Page page = new Page(batchSize, blocks);
+        // Track decompressed bytes for aggregated page source
+        decompressedBytes += page.getSizeInBytes();
+        return page;
     }
 
     private void writeMinMax(int columnIndex, Type type, HiveType hiveType, BlockBuilder blockBuilder, boolean isMin)

presto-hive/src/main/java/com/facebook/presto/hive/orc/OrcBatchPageSource.java

@@ -64,6 +64,7 @@ public class OrcBatchPageSource
 
     private int batchId;
     private long completedPositions;
+    private long decompressedBytes;
     private boolean closed;
 
     private final OrcAggregatedMemoryContext systemMemoryContext;
@@ -157,6 +158,18 @@ public long getCompletedPositions()
         return completedPositions;
     }
 
+    @Override
+    public long getDecompressedBytes()
+    {
+        return decompressedBytes;
+    }
+
+    @Override
+    public long getDecompressedPositions()
+    {
+        return completedPositions;
+    }
+
     @Override
     public long getReadTimeNanos()
     {
@@ -199,7 +212,10 @@ else if (constantBlocks[fieldId] != null) {
                     blocks[fieldId] = new LazyBlock(batchSize, new OrcBlockLoader(hiveColumnIndexes[fieldId]));
                 }
             }
-            return new Page(batchSize, blocks);
+            Page page = new Page(batchSize, blocks);
+            // Track decompressed bytes - this represents data after decompression but before filtering
+            decompressedBytes += page.getSizeInBytes();
+            return page;
         }
         catch (PrestoException e) {
             closeWithSuppression(e);

presto-hive/src/main/java/com/facebook/presto/hive/orc/OrcSelectivePageSource.java

@@ -51,6 +51,7 @@ public class OrcSelectivePageSource
     private final RowIDCoercer coercer;
     private final boolean supplyRowIDs;
     private final OptionalInt rowIDColumnIndex;
+    private long decompressedBytes;
     private boolean closed;
 
     OrcSelectivePageSource(
@@ -95,6 +96,18 @@ public long getCompletedPositions()
         return recordReader.getReadPositions();
     }
 
+    @Override
+    public long getDecompressedBytes()
+    {
+        return decompressedBytes;
+    }
+
+    @Override
+    public long getDecompressedPositions()
+    {
+        return recordReader.getReadPositions();
+    }
+
     @Override
     public long getReadTimeNanos()
     {
@@ -123,6 +136,10 @@ public Page getNextPage()
             if (page == null) {
                 close();
             }
+            else {
+                // Track decompressed bytes - this represents data after decompression but before filtering
+                decompressedBytes += page.getSizeInBytes();
+            }
             return page;
         }
         catch (InvalidFunctionArgumentException e) {

presto-hive/src/main/java/com/facebook/presto/hive/pagefile/PageFilePageSource.java

@@ -42,6 +42,8 @@ public class PageFilePageSource
     private long completedBytes;
     private long readTimeNanos;
     private long memoryUsageBytes;
+    private long decompressedBytes;
+    private long decompressedPositions;
 
     public PageFilePageSource(
             FSDataInputStream inputStream,
@@ -127,7 +129,10 @@ public Page getNextPage()
         long pageSizeInBytes = page.getSizeInBytes();
         completedBytes += pageSizeInBytes;
         memoryUsageBytes = Math.max(memoryUsageBytes, pageSizeInBytes);
-        return new Page(page.getPositionCount(), blocks);
+        Page resultPage = new Page(page.getPositionCount(), blocks);
+        decompressedBytes += resultPage.getSizeInBytes();
+        decompressedPositions += resultPage.getPositionCount();
+        return resultPage;
     }
 
     @Override
@@ -136,6 +141,18 @@ public long getSystemMemoryUsage()
         return memoryUsageBytes;
     }
 
+    @Override
+    public long getDecompressedBytes()
+    {
+        return decompressedBytes;
+    }
+
+    @Override
+    public long getDecompressedPositions()
+    {
+        return decompressedPositions;
+    }
+
     @Override
     public void close()
             throws IOException

presto-hive/src/main/java/com/facebook/presto/hive/parquet/AggregatedParquetPageSource.java

@@ -60,6 +60,8 @@ public class AggregatedParquetPageSource
     private boolean completed;
     private long readTimeNanos;
     private long completedBytes;
+    private long decompressedBytes;
+    private long decompressedPositions;
 
     public AggregatedParquetPageSource(List<HiveColumnHandle> columnHandles, ParquetMetadata parquetMetadata, TypeManager typeManager, StandardFunctionResolution functionResolution)
     {
@@ -131,7 +133,10 @@ else if (functionResolution.isMinFunction(functionHandle)) {
 
         completed = true;
         readTimeNanos += System.nanoTime() - start;
-        return new Page(batchSize, blocks);
+        Page page = new Page(batchSize, blocks);
+        decompressedBytes += page.getSizeInBytes();
+        decompressedPositions += page.getPositionCount();
+        return page;
     }
 
     private long getRowCountFromParquetMetadata(ParquetMetadata parquetMetadata)
@@ -245,6 +250,18 @@ public long getSystemMemoryUsage()
         return 0;
     }
 
+    @Override
+    public long getDecompressedBytes()
+    {
+        return decompressedBytes;
+    }
+
+    @Override
+    public long getDecompressedPositions()
+    {
+        return decompressedPositions;
+    }
+
     @Override
     public void close()
             throws IOException

presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetPageSource.java

@@ -58,6 +58,7 @@ public class ParquetPageSource
 
     private int batchId;
     private long completedPositions;
+    private long decompressedBytes;
     private boolean closed;
 
     private final RuntimeStats runtimeStats;
@@ -116,6 +117,18 @@ public long getCompletedPositions()
         return completedPositions;
     }
 
+    @Override
+    public long getDecompressedBytes()
+    {
+        return decompressedBytes;
+    }
+
+    @Override
+    public long getDecompressedPositions()
+    {
+        return completedPositions;
+    }
+
     @Override
     public long getReadTimeNanos()
     {
@@ -163,7 +176,10 @@ public Page getNextPage()
                     }
                 }
             }
-            return new Page(batchSize, blocks);
+            Page page = new Page(batchSize, blocks);
+            // Track decompressed bytes - represents data after decompression but before filtering
+            decompressedBytes += page.getSizeInBytes();
+            return page;
         }
         catch (PrestoException e) {
             closeWithSuppression(e);

presto-hive/src/main/java/com/facebook/presto/hive/rcfile/RcFilePageSource.java

@@ -56,6 +56,8 @@ public class RcFilePageSource
 
     private int pageId;
     private long completedPositions;
+    private long decompressedBytes;
+    private long decompressedPositions;
 
     private boolean closed;
 
@@ -149,7 +151,10 @@ public Page getNextPage()
                 }
             }
 
-            return new Page(currentPageSize, blocks);
+            Page page = new Page(currentPageSize, blocks);
+            decompressedBytes += page.getSizeInBytes();
+            decompressedPositions += page.getPositionCount();
+            return page;
         }
         catch (PrestoException e) {
             closeWithSuppression(e);
@@ -193,6 +198,18 @@ public long getSystemMemoryUsage()
         return GUESSED_MEMORY_USAGE;
     }
 
+    @Override
+    public long getDecompressedBytes()
+    {
+        return decompressedBytes;
+    }
+
+    @Override
+    public long getDecompressedPositions()
+    {
+        return decompressedPositions;
+    }
+
     private void closeWithSuppression(Throwable throwable)
     {
         requireNonNull(throwable, "throwable is null");

presto-main-base/src/main/java/com/facebook/presto/operator/PageSourceOperator.java

@@ -31,6 +31,7 @@ public class PageSourceOperator
     private final ConnectorPageSource pageSource;
     private final OperatorContext operatorContext;
     private long completedBytes;
+    private long decompressedBytes;
     private long readTimeNanos;
 
     public PageSourceOperator(ConnectorPageSource pageSource, OperatorContext operatorContext)
@@ -89,12 +90,16 @@ public Page getOutput()
             return null;
         }
 
-        // update operator stats
+        // update operator stats with three-tier metrics
         long endCompletedBytes = pageSource.getCompletedBytes();
+        long endDecompressedBytes = pageSource.getDecompressedBytes();
         long endReadTimeNanos = pageSource.getReadTimeNanos();
+        // Raw input: compressed bytes from storage
         operatorContext.recordRawInputWithTiming(endCompletedBytes - completedBytes, page.getPositionCount(), endReadTimeNanos - readTimeNanos);
-        operatorContext.recordProcessedInput(page.getSizeInBytes(), page.getPositionCount());
+        // Processed input: decompressed bytes (middle tier - after decompression, before filtering)
+        operatorContext.recordProcessedInput(endDecompressedBytes - decompressedBytes, page.getPositionCount());
         completedBytes = endCompletedBytes;
+        decompressedBytes = endDecompressedBytes;
         readTimeNanos = endReadTimeNanos;
 
         // assure the page is in memory before handing to another operator

presto-main-base/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java

@@ -361,7 +361,6 @@ private void recordInputStats()
         long inputBytes = endCompletedBytes - completedBytes;
         long inputBytesReadTime = endReadTimeNanos - readTimeNanos;
         long positionCount = endCompletedPositions - completedPositions;
-        operatorContext.recordProcessedInput(inputBytes, positionCount);
         operatorContext.recordRawInputWithTiming(inputBytes, positionCount, inputBytesReadTime);
         RuntimeStats runtimeStats = pageSource.getRuntimeStats();
         if (runtimeStats != null) {
@@ -392,6 +391,14 @@ private Page recordProcessedInput(Page page)
             }
         }
 
+        // Record processed input: actual block sizes after filtering/projection
+        // If blocks is null, no lazy blocks were found, so use the original page size
+        long processedBytes = (blocks == null) ? page.getSizeInBytes() : blockSizeSum;
+        operatorContext.recordProcessedInput(processedBytes, page.getPositionCount());
+
+        // Record raw input: bytes read from storage before filtering (connector metrics)
+        recordInputStats();
+
         return (blocks == null) ? page : new Page(page.getPositionCount(), blocks);
     }
 

presto-main-base/src/main/java/com/facebook/presto/operator/TableScanOperator.java

@@ -116,6 +116,7 @@ public void noMoreOperators()
 
     private long completedBytes;
     private long completedPositions;
+    private long decompressedBytes;
     private long readTimeNanos;
 
     public TableScanOperator(
@@ -269,7 +270,7 @@ public Page getOutput()
             page = page.getLoadedPage();
         }
 
-        // update operator stats
+        // Record raw input (compressed bytes from storage) and processed input (decompressed bytes)
         recordInputStats();
 
         // updating system memory usage should happen after page is loaded.
@@ -281,15 +282,19 @@ public Page getOutput()
     private void recordInputStats()
     {
         checkState(source != null, "source must not be null");
-        // update operator stats
+        // update operator stats with three-tier metrics
         long endCompletedBytes = source.getCompletedBytes();
         long endCompletedPositions = source.getCompletedPositions();
+        long endDecompressedBytes = source.getDecompressedBytes();
         long endReadTimeNanos = source.getReadTimeNanos();
         long inputBytes = endCompletedBytes - completedBytes;
         long inputBytesReadTime = endReadTimeNanos - readTimeNanos;
         long positionCount = endCompletedPositions - completedPositions;
-        operatorContext.recordProcessedInput(inputBytes, positionCount);
+        long decompressedInputBytes = endDecompressedBytes - decompressedBytes;
+        // Raw input: compressed bytes from storage
         operatorContext.recordRawInputWithTiming(inputBytes, positionCount, inputBytesReadTime);
+        // Processed input: decompressed bytes (middle tier - after decompression, before filtering)
+        operatorContext.recordProcessedInput(decompressedInputBytes, positionCount);
         RuntimeStats runtimeStats = source.getRuntimeStats();
         if (runtimeStats != null) {
             runtimeStats.addMetricValueIgnoreZero(STORAGE_READ_TIME_NANOS, NANO, inputBytesReadTime);
@@ -298,6 +303,7 @@ private void recordInputStats()
         }
         completedBytes = endCompletedBytes;
         completedPositions = endCompletedPositions;
+        decompressedBytes = endDecompressedBytes;
         readTimeNanos = endReadTimeNanos;
     }
 }