Add NodeProvider interface and ModularHashingNodeProvider

cjjxfli · Dec 23, 2021 · 9e46a70 · 9e46a70
1 parent 9b0712a
commit 9e46a70
Show file tree

Hide file tree

Showing 47 changed files with 182 additions and 100 deletions.
diff --git a/presto-accumulo/src/main/java/com/facebook/presto/accumulo/model/AccumuloSplit.java b/presto-accumulo/src/main/java/com/facebook/presto/accumulo/model/AccumuloSplit.java
@@ -16,6 +16,7 @@
 import com.facebook.presto.accumulo.serializers.AccumuloRowSerializer;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -168,7 +169,7 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return addresses;
     }

diff --git a/presto-atop/src/main/java/com/facebook/presto/atop/AtopSplit.java b/presto-atop/src/main/java/com/facebook/presto/atop/AtopSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -85,7 +86,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         // discard the port number
         return ImmutableList.of(HostAddress.fromString(host.getHostText()));

diff --git a/presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcSplit.java b/presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcSplit.java
@@ -18,6 +18,7 @@
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -108,7 +109,7 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return ImmutableList.of();
     }

diff --git a/presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/BigQuerySplit.java b/presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/BigQuerySplit.java
@@ -16,6 +16,7 @@
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -95,7 +96,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return ImmutableList.of();
     }

diff --git a/presto-blackhole/src/main/java/com/facebook/presto/plugin/blackhole/BlackHoleSplit.java b/presto-blackhole/src/main/java/com/facebook/presto/plugin/blackhole/BlackHoleSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -79,7 +80,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return ImmutableList.of();
     }

diff --git a/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraSplit.java b/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -97,7 +98,7 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return addresses;
     }

diff --git a/presto-druid/src/main/java/com/facebook/presto/druid/DruidSplit.java b/presto-druid/src/main/java/com/facebook/presto/druid/DruidSplit.java
@@ -16,6 +16,7 @@
 import com.facebook.presto.druid.metadata.DruidSegmentInfo;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -106,7 +107,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return address.map(ImmutableList::of).orElse(ImmutableList.of());
     }

diff --git a/presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchSplit.java b/presto-elasticsearch/src/main/java/com/facebook/presto/elasticsearch/ElasticsearchSplit.java
@@ -17,6 +17,7 @@
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -81,7 +82,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return address.map(host -> ImmutableList.of(HostAddress.fromString(host)))
                 .orElseGet(ImmutableList::of);

diff --git a/presto-example-http/src/main/java/com/facebook/presto/example/ExampleSplit.java b/presto-example-http/src/main/java/com/facebook/presto/example/ExampleSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -86,7 +87,7 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return addresses;
     }

diff --git a/presto-google-sheets/src/main/java/com/facebook/presto/google/sheets/SheetsSplit.java b/presto-google-sheets/src/main/java/com/facebook/presto/google/sheets/SheetsSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -70,7 +71,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return hostAddresses;
     }

diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveSplit.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveSplit.java
@@ -17,7 +17,7 @@
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
-import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.SplitWeight;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -34,7 +34,6 @@
 import java.util.OptionalLong;
 import java.util.Set;
 
-import static com.facebook.presto.spi.StandardErrorCode.NO_NODES_AVAILABLE;
 import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.SOFT_AFFINITY;
 import static com.google.common.base.MoreObjects.toStringHelper;
 import static com.google.common.base.Preconditions.checkArgument;
@@ -209,20 +208,10 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
-        if (sortedCandidates == null || sortedCandidates.isEmpty()) {
-            throw new PrestoException(NO_NODES_AVAILABLE, "sortedCandidates is null or empty for HiveSplit");
-        }
-
         if (getNodeSelectionStrategy() == SOFT_AFFINITY) {
-            // Use + 1 as secondary hash for now, would always get a different position from the first hash.
-            int size = sortedCandidates.size();
-            int mod = path.hashCode() % size;
-            int position = mod < 0 ? mod + size : mod;
-            return ImmutableList.of(
-                    sortedCandidates.get(position),
-                    sortedCandidates.get((position + 1) % size));
+            return nodeProvider.get(path, 2);
         }
         return addresses;
     }

diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergSplit.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergSplit.java
@@ -15,7 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
-import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -27,7 +27,6 @@
 import java.util.List;
 import java.util.Map;
 
-import static com.facebook.presto.spi.StandardErrorCode.NO_NODES_AVAILABLE;
 import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.SOFT_AFFINITY;
 import static com.google.common.base.MoreObjects.toStringHelper;
 import static java.util.Objects.requireNonNull;
@@ -106,20 +105,10 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
         return nodeSelectionStrategy;
     }
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
-        if (sortedCandidates == null || sortedCandidates.isEmpty()) {
-            throw new PrestoException(NO_NODES_AVAILABLE, "sortedCandidates is null or empty for HiveSplit");
-        }
-
         if (getNodeSelectionStrategy() == SOFT_AFFINITY) {
-            // Use + 1 as secondary hash for now, would always get a different position from the first hash.
-            int size = sortedCandidates.size();
-            int mod = path.hashCode() % size;
-            int position = mod < 0 ? mod + size : mod;
-            return ImmutableList.of(
-                sortedCandidates.get(position),
-                sortedCandidates.get((position + 1) % size));
+            return nodeProvider.get(path, 2);
         }
         return addresses;
     }

diff --git a/presto-jmx/src/main/java/com/facebook/presto/connector/jmx/JmxSplit.java b/presto-jmx/src/main/java/com/facebook/presto/connector/jmx/JmxSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -59,7 +60,7 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return addresses;
     }

diff --git a/presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaSplit.java b/presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -137,7 +138,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return ImmutableList.of(leader);
     }

diff --git a/presto-kudu/src/main/java/com/facebook/presto/kudu/KuduSplit.java b/presto-kudu/src/main/java/com/facebook/presto/kudu/KuduSplit.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -67,7 +68,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return ImmutableList.of();
     }

diff --git a/presto-local-file/src/main/java/com/facebook/presto/localfile/LocalFileSplit.java b/presto-local-file/src/main/java/com/facebook/presto/localfile/LocalFileSplit.java
@@ -16,6 +16,7 @@
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -71,7 +72,7 @@ public NodeSelectionStrategy getNodeSelectionStrategy()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return ImmutableList.of(address);
     }

diff --git a/...src/main/java/com/facebook/presto/connector/informationSchema/InformationSchemaSplit.java b/...src/main/java/com/facebook/presto/connector/informationSchema/InformationSchemaSplit.java
@@ -16,6 +16,7 @@
 import com.facebook.presto.metadata.QualifiedTablePrefix;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -64,7 +65,7 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return addresses;
     }

diff --git a/presto-main/src/main/java/com/facebook/presto/connector/system/SystemSplit.java b/presto-main/src/main/java/com/facebook/presto/connector/system/SystemSplit.java
@@ -18,6 +18,7 @@
 import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeProvider;
 import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -78,7 +79,7 @@ public List<HostAddress> getAddresses()
     }
 
     @Override
-    public List<HostAddress> getPreferredNodes(List<HostAddress> sortedCandidates)
+    public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     {
         return addresses;
     }