Add support for index joins

weipinghe · Apr 19, 2014 · 7756868 · 7756868
1 parent e27fcac
commit 7756868
Show file tree

Hide file tree

Showing 84 changed files with 3,168 additions and 41 deletions.
diff --git a/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraConnector.java b/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraConnector.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.Connector;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.ConnectorIndexResolver;
 import com.facebook.presto.spi.ConnectorMetadata;
 import com.facebook.presto.spi.ConnectorOutputHandleResolver;
 import com.facebook.presto.spi.ConnectorRecordSetProvider;
@@ -77,4 +78,10 @@ public ConnectorOutputHandleResolver getOutputHandleResolver()
     {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public ConnectorIndexResolver getIndexResolver()
+    {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraHandleResolver.java b/presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraHandleResolver.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.IndexHandle;
 import com.facebook.presto.spi.Split;
 import com.facebook.presto.spi.TableHandle;
 import com.google.common.base.Objects;
@@ -52,6 +53,12 @@ public boolean canHandle(Split split)
         return split instanceof CassandraSplit && ((CassandraSplit) split).getConnectorId().equals(connectorId);
     }
 
+    @Override
+    public boolean canHandle(IndexHandle indexHandle)
+    {
+        return false;
+    }
+
     @Override
     public Class<? extends TableHandle> getTableHandleClass()
     {
@@ -70,6 +77,12 @@ public Class<? extends Split> getSplitClass()
         return CassandraSplit.class;
     }
 
+    @Override
+    public Class<? extends IndexHandle> getIndexHandleClass()
+    {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public String toString()
     {

diff --git a/presto-example-http/src/main/java/com/facebook/presto/example/ExampleConnector.java b/presto-example-http/src/main/java/com/facebook/presto/example/ExampleConnector.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.Connector;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.ConnectorIndexResolver;
 import com.facebook.presto.spi.ConnectorMetadata;
 import com.facebook.presto.spi.ConnectorOutputHandleResolver;
 import com.facebook.presto.spi.ConnectorRecordSetProvider;
@@ -81,4 +82,10 @@ public ConnectorOutputHandleResolver getOutputHandleResolver()
     {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public ConnectorIndexResolver getIndexResolver()
+    {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/presto-example-http/src/main/java/com/facebook/presto/example/ExampleHandleResolver.java b/presto-example-http/src/main/java/com/facebook/presto/example/ExampleHandleResolver.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.IndexHandle;
 import com.facebook.presto.spi.Split;
 import com.facebook.presto.spi.TableHandle;
 
@@ -51,6 +52,12 @@ public boolean canHandle(Split split)
         return split instanceof ExampleSplit && ((ExampleSplit) split).getConnectorId().equals(connectorId);
     }
 
+    @Override
+    public boolean canHandle(IndexHandle indexHandle)
+    {
+        return false;
+    }
+
     @Override
     public Class<? extends TableHandle> getTableHandleClass()
     {
@@ -68,4 +75,10 @@ public Class<? extends Split> getSplitClass()
     {
         return ExampleSplit.class;
     }
+
+    @Override
+    public Class<? extends IndexHandle> getIndexHandleClass()
+    {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveClient.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveClient.java
@@ -27,6 +27,7 @@
 import com.facebook.presto.spi.ConnectorTableMetadata;
 import com.facebook.presto.spi.Domain;
 import com.facebook.presto.spi.FixedSplitSource;
+import com.facebook.presto.spi.IndexHandle;
 import com.facebook.presto.spi.OutputTableHandle;
 import com.facebook.presto.spi.Partition;
 import com.facebook.presto.spi.PartitionResult;
@@ -844,6 +845,12 @@ public boolean canHandle(OutputTableHandle handle)
         return (handle instanceof HiveOutputTableHandle) && ((HiveOutputTableHandle) handle).getClientId().equals(connectorId);
     }
 
+    @Override
+    public boolean canHandle(IndexHandle indexHandle)
+    {
+        return false;
+    }
+
     @Override
     public Class<? extends TableHandle> getTableHandleClass()
     {
@@ -868,6 +875,12 @@ public Class<? extends OutputTableHandle> getOutputTableHandleClass()
         return HiveOutputTableHandle.class;
     }
 
+    @Override
+    public Class<? extends IndexHandle> getIndexHandleClass()
+    {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public String toString()
     {

diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveConnector.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveConnector.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.Connector;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.ConnectorIndexResolver;
 import com.facebook.presto.spi.ConnectorMetadata;
 import com.facebook.presto.spi.ConnectorOutputHandleResolver;
 import com.facebook.presto.spi.ConnectorRecordSetProvider;
@@ -84,4 +85,10 @@ public ConnectorOutputHandleResolver getOutputHandleResolver()
     {
         return outputHandleResolver;
     }
+
+    @Override
+    public ConnectorIndexResolver getIndexResolver()
+    {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/...main/src/main/java/com/facebook/presto/block/dictionary/DictionaryEncodedBlockCursor.java b/...main/src/main/java/com/facebook/presto/block/dictionary/DictionaryEncodedBlockCursor.java
@@ -40,6 +40,18 @@ public DictionaryEncodedBlockCursor(RandomAccessBlock dictionary, BlockCursor id
         checkArgument(idCursor.getType().equals(BIGINT), "Expected bigint cursor but got %s cursor", idCursor.getType());
     }
 
+    public DictionaryEncodedBlockCursor(DictionaryEncodedBlockCursor cursor)
+    {
+        this.dictionary = cursor.dictionary;
+        this.idCursor = cursor.idCursor.duplicate();
+    }
+
+    @Override
+    public BlockCursor duplicate()
+    {
+        return new DictionaryEncodedBlockCursor(this);
+    }
+
     @Override
     public Type getType()
     {

diff --git a/presto-main/src/main/java/com/facebook/presto/block/rle/RunLengthEncodedBlockCursor.java b/presto-main/src/main/java/com/facebook/presto/block/rle/RunLengthEncodedBlockCursor.java
@@ -44,6 +44,19 @@ public RunLengthEncodedBlockCursor(RandomAccessBlock value, int positionCount)
         position = -1;
     }
 
+    public RunLengthEncodedBlockCursor(RunLengthEncodedBlockCursor cursor)
+    {
+        this.value = cursor.value;
+        this.positionCount = cursor.positionCount;
+        this.position = cursor.position;
+    }
+
+    @Override
+    public BlockCursor duplicate()
+    {
+        return new RunLengthEncodedBlockCursor(this);
+    }
+
     @Override
     public Type getType()
     {

diff --git a/presto-main/src/main/java/com/facebook/presto/connector/ConnectorManager.java b/presto-main/src/main/java/com/facebook/presto/connector/ConnectorManager.java
@@ -13,13 +13,15 @@
  */
 package com.facebook.presto.connector;
 
+import com.facebook.presto.index.IndexManager;
 import com.facebook.presto.metadata.HandleResolver;
 import com.facebook.presto.metadata.MetadataManager;
 import com.facebook.presto.metadata.OutputTableHandleResolver;
 import com.facebook.presto.operator.RecordSinkManager;
 import com.facebook.presto.spi.Connector;
 import com.facebook.presto.spi.ConnectorFactory;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.ConnectorIndexResolver;
 import com.facebook.presto.spi.ConnectorMetadata;
 import com.facebook.presto.spi.ConnectorOutputHandleResolver;
 import com.facebook.presto.spi.ConnectorRecordSetProvider;
@@ -47,6 +49,8 @@ public class ConnectorManager
     private final MetadataManager metadataManager;
     private final SplitManager splitManager;
     private final DataStreamManager dataStreamManager;
+    private final IndexManager indexManager;
+
     private final RecordSinkManager recordSinkManager;
     private final HandleResolver handleResolver;
     private final OutputTableHandleResolver outputTableHandleResolver;
@@ -59,6 +63,7 @@ public class ConnectorManager
     public ConnectorManager(MetadataManager metadataManager,
             SplitManager splitManager,
             DataStreamManager dataStreamManager,
+            IndexManager indexManager,
             RecordSinkManager recordSinkManager,
             HandleResolver handleResolver,
             OutputTableHandleResolver outputTableHandleResolver,
@@ -68,6 +73,7 @@ public ConnectorManager(MetadataManager metadataManager,
         this.metadataManager = metadataManager;
         this.splitManager = splitManager;
         this.dataStreamManager = dataStreamManager;
+        this.indexManager = indexManager;
         this.recordSinkManager = recordSinkManager;
         this.handleResolver = handleResolver;
         this.outputTableHandleResolver = outputTableHandleResolver;
@@ -182,5 +188,17 @@ private void addConnector(@Nullable String catalogName, String connectorId, Conn
         if (connectorOutputHandleResolver != null) {
             outputTableHandleResolver.addHandleResolver(connectorId, connectorOutputHandleResolver);
         }
+
+        ConnectorIndexResolver indexResolver = null;
+        try {
+            indexResolver = connector.getIndexResolver();
+            checkNotNull(indexResolver, "Connector %s returned a null index resolver", connectorId);
+        }
+        catch (UnsupportedOperationException ignored) {
+        }
+
+        if (indexResolver != null) {
+            indexManager.addIndexResolver(indexResolver);
+        }
     }
 }
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/NativeConnectorFactory.java b/presto-main/src/main/java/com/facebook/presto/connector/NativeConnectorFactory.java
@@ -19,6 +19,7 @@
 import com.facebook.presto.spi.Connector;
 import com.facebook.presto.spi.ConnectorFactory;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.ConnectorIndexResolver;
 import com.facebook.presto.spi.ConnectorMetadata;
 import com.facebook.presto.spi.ConnectorOutputHandleResolver;
 import com.facebook.presto.spi.ConnectorRecordSetProvider;
@@ -106,6 +107,12 @@ public ConnectorRecordSetProvider getRecordSetProvider()
             {
                 throw new UnsupportedOperationException();
             }
+
+            @Override
+            public ConnectorIndexResolver getIndexResolver()
+            {
+                throw new UnsupportedOperationException();
+            }
         };
     }
 }
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/dual/DualConnector.java b/presto-main/src/main/java/com/facebook/presto/connector/dual/DualConnector.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.connector.InternalConnector;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.ConnectorIndexResolver;
 import com.facebook.presto.spi.ConnectorMetadata;
 import com.facebook.presto.spi.ConnectorOutputHandleResolver;
 import com.facebook.presto.spi.ConnectorRecordSetProvider;
@@ -79,4 +80,10 @@ public ConnectorOutputHandleResolver getOutputHandleResolver()
     {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public ConnectorIndexResolver getIndexResolver()
+    {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/dual/DualHandleResolver.java b/presto-main/src/main/java/com/facebook/presto/connector/dual/DualHandleResolver.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.IndexHandle;
 import com.facebook.presto.spi.Split;
 import com.facebook.presto.spi.TableHandle;
 
@@ -39,6 +40,12 @@ public boolean canHandle(Split split)
         return split instanceof DualSplit;
     }
 
+    @Override
+    public boolean canHandle(IndexHandle indexHandle)
+    {
+        return false;
+    }
+
     @Override
     public Class<? extends TableHandle> getTableHandleClass()
     {
@@ -56,4 +63,10 @@ public Class<? extends Split> getSplitClass()
     {
         return DualSplit.class;
     }
+
+    @Override
+    public Class<? extends IndexHandle> getIndexHandleClass()
+    {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/...java/com/facebook/presto/connector/informationSchema/InformationSchemaHandleResolver.java b/...java/com/facebook/presto/connector/informationSchema/InformationSchemaHandleResolver.java
@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.IndexHandle;
 import com.facebook.presto.spi.Split;
 import com.facebook.presto.spi.TableHandle;
 
@@ -39,6 +40,12 @@ public boolean canHandle(Split split)
         return split instanceof InformationSchemaSplit;
     }
 
+    @Override
+    public boolean canHandle(IndexHandle indexHandle)
+    {
+        return false;
+    }
+
     @Override
     public Class<? extends TableHandle> getTableHandleClass()
     {
@@ -56,4 +63,10 @@ public Class<? extends Split> getSplitClass()
     {
         return InformationSchemaSplit.class;
     }
+
+    @Override
+    public Class<? extends IndexHandle> getIndexHandleClass()
+    {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/presto-main/src/main/java/com/facebook/presto/connector/jmx/JmxConnectorFactory.java b/presto-main/src/main/java/com/facebook/presto/connector/jmx/JmxConnectorFactory.java
@@ -17,6 +17,7 @@
 import com.facebook.presto.spi.Connector;
 import com.facebook.presto.spi.ConnectorFactory;
 import com.facebook.presto.spi.ConnectorHandleResolver;
+import com.facebook.presto.spi.ConnectorIndexResolver;
 import com.facebook.presto.spi.ConnectorMetadata;
 import com.facebook.presto.spi.ConnectorOutputHandleResolver;
 import com.facebook.presto.spi.ConnectorRecordSetProvider;
@@ -94,6 +95,12 @@ public ConnectorRecordSinkProvider getRecordSinkProvider()
                 throw new UnsupportedOperationException();
             }
 
+            @Override
+            public ConnectorIndexResolver getIndexResolver()
+            {
+                throw new UnsupportedOperationException();
+            }
+
             @Override
             public ConnectorOutputHandleResolver getOutputHandleResolver()
             {