Remove describe_splits_ex dependency

This method from Cassandra thrift API, which is deprecated by itself, is deprecated. Starting from Cassandra 2.1.x the "system.size_estimates" table should be used instead to retrieve token range size estimates. Based on this estimates it is possible to split token range into smaller sub ranges to satisfy the max_split_size criteria.
prestodb · Feb 10, 2017 · b734061 · b734061
1 parent eb7c106
commit b734061
Show file tree

Hide file tree

Showing 11 changed files with 527 additions and 110 deletions.
diff --git a/presto-cassandra/pom.xml b/presto-cassandra/pom.xml
@@ -14,7 +14,7 @@
     <properties>
         <air.main.basedir>${project.parent.basedir}</air.main.basedir>
         <cassandra.version>2.1.6</cassandra.version>
-        <datastax.version>3.0.0</datastax.version>
+        <datastax.version>3.0.5</datastax.version>
     </properties>
 
     <dependencies>
@@ -237,33 +237,45 @@
         <dependency>
             <groupId>org.cassandraunit</groupId>
             <artifactId>cassandra-unit</artifactId>
-            <version>2.0.2.0</version>
+            <version>2.1.9.2</version>
             <scope>test</scope>
             <exclusions>
                 <exclusion>
                     <artifactId>junit</artifactId>
                     <groupId>junit</groupId>
                 </exclusion>
                 <exclusion>
-                    <groupId>org.slf4j</groupId>
-                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.apache.cassandra</groupId>
+                    <artifactId>cassandra-all</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.cassandra</groupId>
+            <artifactId>cassandra-all</artifactId>
+            <version>2.1.12</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.netty</groupId>
+                    <artifactId>netty-all</artifactId>
                 </exclusion>
                 <exclusion>
                     <groupId>org.apache.thrift</groupId>
                     <artifactId>libthrift</artifactId>
                 </exclusion>
                 <exclusion>
-                    <groupId>org.jboss.netty</groupId>
-                    <artifactId>netty</artifactId>
+                    <artifactId>junit</artifactId>
+                    <groupId>junit</groupId>
                 </exclusion>
                 <exclusion>
-                    <groupId>com.datastax.cassandra</groupId>
-                    <artifactId>cassandra-driver-core</artifactId>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-classic</artifactId>
                 </exclusion>
                 <exclusion>
-                    <!-- conflicts with version brought in by presto-main, etc -->
-                    <groupId>com.sun.xml.bind</groupId>
-                    <artifactId>jaxb-impl</artifactId>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-core</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>

diff --git a/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraSession.java b/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraSession.java
@@ -63,6 +63,7 @@
 import static java.util.Comparator.comparing;
 import static java.util.stream.Collectors.toList;
 
+// TODO: Refactor this class to make it be "single responsibility"
 public class CassandraSession
 {
     static final String PRESTO_COMMENT_METADATA = "Presto Metadata:";
@@ -383,6 +384,11 @@ protected Iterable<Row> queryPartitionKeys(CassandraTable table, List<Object> fi
         }
     }
 
+    public <T> T executeWithSession(SessionCallable<T> sessionCallable)
+    {
+        return executeWithSession("", sessionCallable);
+    }
+
     public <T> T executeWithSession(String schemaName, SessionCallable<T> sessionCallable)
     {
         NoHostAvailableException lastException = null;
@@ -402,7 +408,7 @@ public <T> T executeWithSession(String schemaName, SessionCallable<T> sessionCal
         throw lastException;
     }
 
-    private interface SessionCallable<T>
+    public interface SessionCallable<T>
     {
         T executeWithSession(Session session);
     }

diff --git a/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraSplitManager.java b/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraSplitManager.java
@@ -28,7 +28,6 @@
 
 import javax.inject.Inject;
 
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -93,13 +92,7 @@ private List<ConnectorSplit> getSplitsByTokenRange(CassandraTable table, String
         String tokenExpression = table.getTokenExpression();
 
         ImmutableList.Builder<ConnectorSplit> builder = ImmutableList.builder();
-        List<CassandraTokenSplitManager.TokenSplit> tokenSplits;
-        try {
-            tokenSplits = tokenSplitMgr.getSplits(schema, tableName);
-        }
-        catch (IOException e) {
-            throw new RuntimeException(e);
-        }
+        List<CassandraTokenSplitManager.TokenSplit> tokenSplits = tokenSplitMgr.getSplits(schema, tableName);
         for (CassandraTokenSplitManager.TokenSplit tokenSplit : tokenSplits) {
             String condition = buildTokenCondition(tokenExpression, tokenSplit.getStartToken(), tokenSplit.getEndToken());
             List<HostAddress> addresses = new HostAddressFactory().AddressNamesToHostAddressList(tokenSplit.getHosts());