Alluxio cache

core/trino-main/src/main/java/io/trino/testing/TestingNodeManager.java

@@ -25,6 +25,7 @@
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static java.util.Objects.requireNonNull;
 
 public class TestingNodeManager
@@ -35,15 +36,26 @@ public class TestingNodeManager
     private final String environment;
     private final Node localNode;
     private final Set<Node> nodes = new CopyOnWriteArraySet<>();
+    private final boolean scheduleOnCoordinator;
 
     public TestingNodeManager()
     {
         this(TEST_ENVIRONMENT);
     }
 
+    public TestingNodeManager(boolean scheduleOnCoordinator)
+    {
+        this(TEST_ENVIRONMENT, scheduleOnCoordinator);
+    }
+
     public TestingNodeManager(String environment)
     {
-        this(environment, new InternalNode("local", URI.create("local://127.0.0.1"), NodeVersion.UNKNOWN, true), ImmutableSet.of());
+        this(environment, true);
+    }
+
+    public TestingNodeManager(String environment, boolean scheduleOnCoordinator)
+    {
+        this(environment, new InternalNode("local", URI.create("local://127.0.0.1"), NodeVersion.UNKNOWN, true), ImmutableSet.of(), scheduleOnCoordinator);
     }
 
     public TestingNodeManager(Node localNode)
@@ -58,13 +70,14 @@ public TestingNodeManager(List<Node> allNodes)
 
     public TestingNodeManager(Node localNode, Collection<Node> otherNodes)
     {
-        this(TEST_ENVIRONMENT, localNode, otherNodes);
+        this(TEST_ENVIRONMENT, localNode, otherNodes, true);
     }
 
-    public TestingNodeManager(String environment, Node localNode, Collection<Node> otherNodes)
+    public TestingNodeManager(String environment, Node localNode, Collection<Node> otherNodes, boolean scheduleOnCoordinator)
     {
         this.environment = environment;
         this.localNode = requireNonNull(localNode, "localNode is null");
+        this.scheduleOnCoordinator = scheduleOnCoordinator;
         nodes.add(localNode);
         nodes.addAll(otherNodes);
     }
@@ -74,6 +87,11 @@ public void addNode(Node node)
         nodes.add(node);
     }
 
+    public void removeNode(Node node)
+    {
+        nodes.remove(node);
+    }
+
     @Override
     public Set<Node> getAllNodes()
     {
@@ -83,7 +101,12 @@ public Set<Node> getAllNodes()
     @Override
     public Set<Node> getWorkerNodes()
     {
-        return nodes;
+        if (scheduleOnCoordinator) {
+            return nodes;
+        }
+        return nodes.stream()
+                .filter(node -> !node.isCoordinator())
+                .collect(toImmutableSet());
     }
 
     @Override

lib/trino-filesystem-cache-alluxio/pom.xml

@@ -0,0 +1,162 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.trino</groupId>
+        <artifactId>trino-root</artifactId>
+        <version>438-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>trino-filesystem-cache-alluxio</artifactId>
+    <description>Trino Filesystem - Alluxio</description>
+
+    <properties>
+        <air.main.basedir>${project.parent.basedir}</air.main.basedir>
+        <air.compiler.fail-warnings>true</air.compiler.fail-warnings>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.google.errorprone</groupId>
+            <artifactId>error_prone_annotations</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.code.findbugs</groupId>
+                    <artifactId>jsr305</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.inject</groupId>
+            <artifactId>guice</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>configuration</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>stats</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>units</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-filesystem</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>jakarta.validation</groupId>
+            <artifactId>jakarta.validation-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.alluxio</groupId>
+            <artifactId>alluxio-core-client-fs</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.alluxio</groupId>
+            <artifactId>alluxio-core-common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.weakref</groupId>
+            <artifactId>jmxutils</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>slice</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-spi</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>log-manager</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-client</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-filesystem</artifactId>
+            <version>${project.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-main</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-testing</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino.hive</groupId>
+            <artifactId>hive-apache</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>it.unimi.dsi</groupId>
+            <artifactId>fastutil</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

lib/trino-filesystem-cache-alluxio/src/main/java/io/trino/filesystem/alluxio/AlluxioCacheStats.java

@@ -0,0 +1,54 @@
+/*
+ * 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.alluxio;
+
+import com.google.errorprone.annotations.ThreadSafe;
+import io.airlift.stats.DistributionStat;
+import io.trino.filesystem.Location;
+import org.weakref.jmx.Managed;
+import org.weakref.jmx.Nested;
+
+@ThreadSafe
+public class AlluxioCacheStats
+        implements CacheStats
+{
+    private final DistributionStat externalReads = new DistributionStat();
+    private final DistributionStat cacheReads = new DistributionStat();
+
+    @Managed
+    @Nested
+    public DistributionStat getExternalReads()
+    {
+        return externalReads;
+    }
+
+    @Managed
+    @Nested
+    public DistributionStat getCacheReads()
+    {
+        return cacheReads;
+    }
+
+    @Override
+    public void recordCacheRead(Location location, int length)
+    {
+        cacheReads.add(length);
+    }
+
+    @Override
+    public void recordExternalRead(Location location, int length)
+    {
+        externalReads.add(length);
+    }
+}

lib/trino-filesystem-cache-alluxio/src/main/java/io/trino/filesystem/alluxio/AlluxioFileSystemCache.java

@@ -0,0 +1,82 @@
+/*
+ * 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.alluxio;
+
+import alluxio.client.file.CacheContext;
+import alluxio.client.file.URIStatus;
+import alluxio.client.file.cache.CacheManager;
+import alluxio.conf.AlluxioConfiguration;
+import alluxio.wire.FileInfo;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.hash.HashFunction;
+import com.google.common.hash.Hashing;
+import com.google.inject.Inject;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoInput;
+import io.trino.filesystem.TrinoInputFile;
+import io.trino.filesystem.TrinoInputStream;
+import io.trino.filesystem.cache.TrinoFileSystemCache;
+
+import java.io.IOException;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Objects.requireNonNull;
+
+public class AlluxioFileSystemCache
+        implements TrinoFileSystemCache
+{
+    private final CacheManager cacheManager;
+    private final AlluxioConfiguration config;
+    private final CacheStats statistics;
+    private final HashFunction hashFunction = Hashing.murmur3_128();
+
+    @Inject
+    public AlluxioFileSystemCache(CacheManager cacheManager, AlluxioConfiguration config, CacheStats statistics)
+    {
+        this.cacheManager = requireNonNull(cacheManager, "cacheManager is null");
+        this.config = requireNonNull(config, "config is null");
+        this.statistics = requireNonNull(statistics, "statistics is null");
+    }
+
+    @Override
+    public TrinoInput cacheInput(TrinoInputFile delegate, String key)
+            throws IOException
+    {
+        return new AlluxioInput(delegate, uriStatus(delegate, key), cacheManager, config, statistics);
+    }
+
+    @Override
+    public TrinoInputStream cacheStream(TrinoInputFile delegate, String key)
+            throws IOException
+    {
+        return new AlluxioInputStream(delegate, uriStatus(delegate, key), cacheManager, config, statistics);
+    }
+
+    @Override
+    public void expire(Location source)
+            throws IOException
+    {
+    }
+
+    @VisibleForTesting
+    protected URIStatus uriStatus(TrinoInputFile file, String key)
+            throws IOException
+    {
+        FileInfo info = new FileInfo()
+                .setPath(file.location().toString())
+                .setLength(file.length());
+        String cacheIdentifier = hashFunction.hashString(key, UTF_8).toString();
+        return new URIStatus(info, CacheContext.defaults().setCacheIdentifier(cacheIdentifier));
+    }
+}