Rcfile cleanup

.github/config/labeler-config.yml

@@ -2,7 +2,7 @@
 "tests:hive":
   - lib/trino-orc/**
   - lib/trino-parquet/**
-  - lib/trino-rcfile/**
+  - lib/trino-hive-formats/**
   - plugin/trino-hive-hadoop2/**
   - plugin/trino-hive/**
   - testing/trino-product-tests/**

lib/trino-rcfile/src/main/java/io/trino/rcfile/FileRcFileDataSource.java → lib/trino-filesystem/src/main/java/io/trino/filesystem/local/LocalInput.java

@@ -11,78 +11,65 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.rcfile;
+package io.trino.filesystem.local;
+
+import io.trino.filesystem.TrinoInput;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.io.SeekableInputStream;
 
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 
+import static java.lang.Math.min;
 import static java.util.Objects.requireNonNull;
 
-public class FileRcFileDataSource
-        implements RcFileDataSource
+class LocalInput
+        implements TrinoInput
 {
-    private final File path;
-    private final long size;
+    private final File file;
     private final RandomAccessFile input;
-    private long readTimeNanos;
-    private long readBytes;
-
-    public FileRcFileDataSource(File path)
-            throws IOException
-    {
-        this.path = requireNonNull(path, "path is null");
-        this.size = path.length();
-        this.input = new RandomAccessFile(path, "r");
-    }
 
-    @Override
-    public void close()
+    public LocalInput(File file)
             throws IOException
     {
-        input.close();
-    }
-
-    @Override
-    public long getReadBytes()
-    {
-        return readBytes;
+        this.file = requireNonNull(file, "file is null");
+        this.input = new RandomAccessFile(file, "r");
     }
 
     @Override
-    public long getReadTimeNanos()
+    public SeekableInputStream inputStream()
     {
-        return readTimeNanos;
-    }
-
-    @Override
-    public long getSize()
-    {
-        return size;
+        return Files.localInput(file).newStream();
     }
 
     @Override
     public void readFully(long position, byte[] buffer, int bufferOffset, int bufferLength)
             throws IOException
     {
-        long start = System.nanoTime();
-
         input.seek(position);
         input.readFully(buffer, bufferOffset, bufferLength);
+    }
 
-        readTimeNanos += System.nanoTime() - start;
-        readBytes += bufferLength;
+    @Override
+    public int readTail(byte[] buffer, int bufferOffset, int bufferLength)
+            throws IOException
+    {
+        int readSize = (int) min(file.length(), bufferLength);
+        readFully(file.length() - readSize, buffer, bufferOffset, readSize);
+        return readSize;
     }
 
     @Override
-    public RcFileDataSourceId getId()
+    public void close()
+            throws IOException
     {
-        return new RcFileDataSourceId(path.getPath());
+        input.close();
     }
 
     @Override
     public String toString()
     {
-        return path.getPath();
+        return file.getPath();
     }
 }

lib/trino-filesystem/src/main/java/io/trino/filesystem/local/LocalInputFile.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.filesystem.local;
+
+import io.trino.filesystem.TrinoInput;
+import io.trino.filesystem.TrinoInputFile;
+
+import java.io.File;
+import java.io.IOException;
+
+import static java.util.Objects.requireNonNull;
+
+public class LocalInputFile
+        implements TrinoInputFile
+{
+    private final File file;
+
+    public LocalInputFile(File file)
+    {
+        this.file = requireNonNull(file, "file is null");
+    }
+
+    @Override
+    public TrinoInput newInput()
+            throws IOException
+    {
+        return new LocalInput(file);
+    }
+
+    @Override
+    public long length()
+            throws IOException
+    {
+        return file.length();
+    }
+
+    @Override
+    public long modificationTime()
+            throws IOException
+    {
+        return file.lastModified();
+    }
+
+    @Override
+    public boolean exists()
+            throws IOException
+    {
+        return file.exists();
+    }
+
+    @Override
+    public String location()
+    {
+        return file.getPath();
+    }
+
+    @Override
+    public String toString()
+    {
+        return file.getPath();
+    }
+}

lib/trino-rcfile/src/main/java/io/trino/rcfile/MemoryRcFileDataSource.java → lib/trino-filesystem/src/main/java/io/trino/filesystem/memory/MemoryInput.java

@@ -11,57 +11,54 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.rcfile;
+package io.trino.filesystem.memory;
 
 import io.airlift.slice.Slice;
+import io.trino.filesystem.TrinoInput;
+import org.apache.iceberg.io.SeekableInputStream;
 
+import static java.lang.Math.min;
 import static java.lang.Math.toIntExact;
 import static java.util.Objects.requireNonNull;
 
-public class MemoryRcFileDataSource
-        implements RcFileDataSource
+class MemoryInput
+        implements TrinoInput
 {
-    private final RcFileDataSourceId id;
+    private final String location;
     private final Slice data;
-    private long readBytes;
 
-    public MemoryRcFileDataSource(RcFileDataSourceId id, Slice data)
+    public MemoryInput(String location, Slice data)
     {
-        this.id = requireNonNull(id, "id is null");
+        this.location = requireNonNull(location, "location is null");
         this.data = requireNonNull(data, "data is null");
     }
 
     @Override
-    public RcFileDataSourceId getId()
+    public SeekableInputStream inputStream()
     {
-        return id;
+        return new MemorySeekableInputStream(data);
     }
 
     @Override
-    public long getReadBytes()
+    public void readFully(long position, byte[] buffer, int bufferOffset, int bufferLength)
     {
-        return readBytes;
+        data.getBytes(toIntExact(position), buffer, bufferOffset, bufferLength);
     }
 
     @Override
-    public long getReadTimeNanos()
+    public int readTail(byte[] buffer, int bufferOffset, int bufferLength)
     {
-        return 0;
+        int readSize = min(data.length(), bufferLength);
+        readFully(data.length() - readSize, buffer, bufferOffset, readSize);
+        return readSize;
     }
 
     @Override
-    public long getSize()
-    {
-        return data.length();
-    }
+    public void close() {}
 
     @Override
-    public void readFully(long position, byte[] buffer, int bufferOffset, int bufferLength)
+    public String toString()
     {
-        data.getBytes(toIntExact(position), buffer, bufferOffset, bufferLength);
-        readBytes += bufferLength;
+        return location;
     }
-
-    @Override
-    public void close() {}
 }

lib/trino-filesystem/src/main/java/io/trino/filesystem/memory/MemoryInputFile.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.filesystem.memory;
+
+import io.airlift.slice.Slice;
+import io.trino.filesystem.TrinoInput;
+import io.trino.filesystem.TrinoInputFile;
+
+import java.io.IOException;
+
+import static java.util.Objects.requireNonNull;
+
+public class MemoryInputFile
+        implements TrinoInputFile
+{
+    private final String location;
+    private final Slice data;
+
+    public MemoryInputFile(String location, Slice data)
+    {
+        this.location = requireNonNull(location, "location is null");
+        this.data = requireNonNull(data, "data is null");
+    }
+
+    @Override
+    public TrinoInput newInput()
+            throws IOException
+    {
+        return new MemoryInput(location, data);
+    }
+
+    @Override
+    public long length()
+            throws IOException
+    {
+        return data.length();
+    }
+
+    @Override
+    public long modificationTime()
+            throws IOException
+    {
+        return 0;
+    }
+
+    @Override
+    public boolean exists()
+            throws IOException
+    {
+        return true;
+    }
+
+    @Override
+    public String location()
+    {
+        return location;
+    }
+
+    @Override
+    public String toString()
+    {
+        return location;
+    }
+}