Fix physical input read time metric for text files

Time taken for reading small files was not included in the
existing implementation

Author: raunaqmorarka

plugin/trino-hive/src/main/java/io/trino/plugin/hive/line/LinePageSource.java

@@ -41,16 +41,18 @@ public class LinePageSource
     private final LineDeserializer deserializer;
     private final LineBuffer lineBuffer;
     private final Location filePath;
+    private final long smallFileReadTimeNanos;
 
     private final PageBuilder pageBuilder;
     private long completedPositions;
 
-    public LinePageSource(LineReader lineReader, LineDeserializer deserializer, LineBuffer lineBuffer, Location filePath)
+    public LinePageSource(LineReader lineReader, LineDeserializer deserializer, LineBuffer lineBuffer, Location filePath, long smallFileReadTimeNanos)
     {
         this.lineReader = requireNonNull(lineReader, "lineReader is null");
         this.deserializer = requireNonNull(deserializer, "deserializer is null");
         this.lineBuffer = requireNonNull(lineBuffer, "lineBuffer is null");
         this.filePath = requireNonNull(filePath, "filePath is null");
+        this.smallFileReadTimeNanos = smallFileReadTimeNanos;
 
         this.pageBuilder = new PageBuilder(deserializer.getTypes());
     }
@@ -105,7 +107,7 @@ public OptionalLong getCompletedPositions()
     @Override
     public long getReadTimeNanos()
     {
-        return lineReader.getReadTimeNanos();
+        return smallFileReadTimeNanos + lineReader.getReadTimeNanos();
     }
 
     @Override

plugin/trino-hive/src/main/java/io/trino/plugin/hive/line/LinePageSourceFactory.java

@@ -132,9 +132,12 @@ private ConnectorPageSource createPageSource(
         TrinoInputFile inputFile = trinoFileSystem.newInputFile(path);
         try {
             // buffer file if small
+            long smallFileReadTimeNanos = 0;
             if (estimatedFileSize < SMALL_FILE_SIZE.toBytes()) {
+                long readStart = System.nanoTime();
                 try (InputStream inputStream = inputFile.newStream()) {
                     byte[] data = inputStream.readAllBytes();
+                    smallFileReadTimeNanos = System.nanoTime() - readStart;
                     inputFile = new MemoryInputFile(path, Slices.wrappedBuffer(data));
                 }
             }
@@ -150,7 +153,7 @@ private ConnectorPageSource createPageSource(
             if (lineReader.isClosed()) {
                 return new EmptyPageSource();
             }
-            return new LinePageSource(lineReader, lineDeserializer, lineReaderFactory.createLineBuffer(), path);
+            return new LinePageSource(lineReader, lineDeserializer, lineReaderFactory.createLineBuffer(), path, smallFileReadTimeNanos);
         }
         catch (TrinoException e) {
             throw e;