Pulsar connector

pom.xml

@@ -118,6 +118,7 @@
         <module>presto-thrift-connector</module>
         <module>presto-matching</module>
         <module>presto-memory-context</module>
+        <module>presto-pulsar</module>
     </modules>
 
     <dependencyManagement>

presto-main/etc/catalog/kafka.properties

@@ -0,0 +1,3 @@
+connector.name=kafka
+kafka.table-names=table1,table2
+kafka.nodes=localhost:12345

presto-main/etc/catalog/pulsar.properties

@@ -0,0 +1,5 @@
+connector.name=pulsar
+pulsar.broker-service-url=http://localhost:9090
+pulsar.zookeeper-uri=localhost:2181
+pulsar.entry-read-batch-size=100
+pulsar.target-num-splits=4

presto-main/etc/config.properties

@@ -39,7 +39,8 @@ plugin.bundles=\
   ../presto-mysql/pom.xml,\
   ../presto-sqlserver/pom.xml, \
   ../presto-postgresql/pom.xml, \
-  ../presto-tpcds/pom.xml
+  ../presto-tpcds/pom.xml, \
+  ../presto-pulsar/pom.xml
 
 presto.version=testversion
 distributed-joins-enabled=true

presto-product-tests/conf/presto/etc/config.properties

@@ -37,7 +37,8 @@ plugin.bundles=\
   ../../../presto-postgresql/pom.xml,\
   ../../../presto-sqlserver/pom.xml,\
   ../../../presto-ml/pom.xml,\
-  ../../../presto-tpcds/pom.xml
+  ../../../presto-tpcds/pom.xml, \
+  ../../../presto-pulsar/pom.xml
 
 presto.version=testversion
 distributed-joins-enabled=true

presto-pulsar/pom.xml

@@ -0,0 +1,153 @@
+<?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">
+    <parent>
+        <artifactId>presto-root</artifactId>
+        <groupId>com.facebook.presto</groupId>
+        <version>0.204-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>presto-pulsar</artifactId>
+    <description>Presto - Apache Pulsar Connector</description>
+    <packaging>presto-plugin</packaging>
+
+    <properties>
+        <air.main.basedir>${project.parent.basedir}</air.main.basedir>
+        <air.check.fail-checkstyle>false</air.check.fail-checkstyle>
+        <air.check.skip-checkstyle>true</air.check.skip-checkstyle>
+        <air.check.skip-dependency>true</air.check.skip-dependency>
+        <air.check.skip-duplicate-finder>true</air.check.skip-duplicate-finder>
+        <avro.version>1.8.2</avro.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>bootstrap</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>json</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>log</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>configuration</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.inject</groupId>
+            <artifactId>guice</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>javax.validation</groupId>
+            <artifactId>validation-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>javax.inject</groupId>
+            <artifactId>javax.inject</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>${avro.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client-admin</artifactId>
+            <version>2.2.0-incubating-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client</artifactId>
+            <version>2.2.0-incubating-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>managed-ledger</artifactId>
+            <version>2.2.0-incubating-SNAPSHOT</version>
+        </dependency>
+        <!-- Presto SPI -->
+        <dependency>
+            <groupId>com.facebook.presto</groupId>
+            <artifactId>presto-spi</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>slice</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>units</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- test dependencies -->
+
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.18.1</version>
+                <configuration>
+                    <argLine>-Xmx4g -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -Xloggc:${basedir}/target/surefire-reports/gc.log -Dlog4j.configuration=${basedir}/test/resources/log4j2-test.xml</argLine>
+                    <classpathDependencyExcludes>
+                        <classpathDependencyExcludes>org.apache.logging.log4j:log4j-1.2-api</classpathDependencyExcludes>
+                    </classpathDependencyExcludes>
+                    <redirectTestOutputToFile>true</redirectTestOutputToFile>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+
+</project>

presto-pulsar/src/main/java/com/facebook/presto/pulsar/AvroSchemaHandler.java

@@ -0,0 +1,59 @@
+/*
+ * 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 com.facebook.presto.pulsar;
+
+import io.airlift.log.Logger;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.DatumReader;
+import org.apache.avro.io.DecoderFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+public class AvroSchemaHandler implements SchemaHandler {
+
+    private final DatumReader<GenericRecord> datumReader;
+
+    private final List<PulsarColumnHandle> columnHandles;
+
+    private static final Logger log = Logger.get(AvroSchemaHandler.class);
+
+    public AvroSchemaHandler(Schema schema, List<PulsarColumnHandle> columnHandles) {
+        this.datumReader = new GenericDatumReader<>(schema);
+        this.columnHandles = columnHandles;
+    }
+
+    @Override
+    public Object deserialize(byte[] bytes) {
+        try {
+            return this.datumReader.read(null, DecoderFactory.get().binaryDecoder(bytes, null));
+        } catch (IOException e) {
+            log.error(e);
+        }
+        return null;
+    }
+
+    @Override
+    public Object extractField(int index, Object currentRecord) {
+        try {
+            GenericRecord record = (GenericRecord) currentRecord;
+            return record.get(this.columnHandles.get(index).getPositionIndex());
+        } catch (Exception ex) {
+            log.error(ex);
+        }
+        return null;
+    }
+}

presto-pulsar/src/main/java/com/facebook/presto/pulsar/JSONSchemaHandler.java

@@ -0,0 +1,83 @@
+/*
+ * 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 com.facebook.presto.pulsar;
+
+import com.facebook.presto.spi.type.Type;
+import io.airlift.log.Logger;
+import io.airlift.slice.Slice;
+import org.apache.pulsar.shade.com.google.gson.JsonElement;
+import org.apache.pulsar.shade.com.google.gson.JsonObject;
+import org.apache.pulsar.shade.com.google.gson.JsonParser;
+
+import java.util.List;
+
+import static com.facebook.presto.spi.type.IntegerType.INTEGER;
+import static com.facebook.presto.spi.type.RealType.REAL;
+import static com.facebook.presto.spi.type.SmallintType.SMALLINT;
+
+public class JSONSchemaHandler implements SchemaHandler {
+
+    private static final Logger log = Logger.get(JSONSchemaHandler.class);
+
+    private List<PulsarColumnHandle> columnHandles;
+
+    private final JsonParser jsonParser = new JsonParser();
+
+    public JSONSchemaHandler(List<PulsarColumnHandle> columnHandles) {
+        this.columnHandles = columnHandles;
+    }
+
+    @Override
+    public Object deserialize(byte[] bytes) {
+        JsonElement jsonElement = this.jsonParser.parse(new String(bytes));
+        return jsonElement.getAsJsonObject();
+    }
+
+    @Override
+    public Object extractField(int index, Object currentRecord) {
+        try {
+            JsonObject jsonObject = (JsonObject) currentRecord;
+            PulsarColumnHandle pulsarColumnHandle = columnHandles.get(index);
+            JsonElement field = jsonObject.get(pulsarColumnHandle.getName());
+            if (field.isJsonNull()) {
+                return null;
+            }
+            Type type = pulsarColumnHandle.getType();
+            Class<?> javaType = type.getJavaType();
+
+            if (javaType == long.class) {
+                if (type.equals(INTEGER)) {
+                    return field.getAsInt();
+                } else if (type.equals(REAL)) {
+                    return field.getAsFloat();
+                } else if (type.equals(SMALLINT)) {
+                    return field.getAsShort();
+                } else {
+                    return field.getAsLong();
+                }
+            } else if (javaType == boolean.class) {
+                return field.getAsBoolean();
+            } else if (javaType == double.class) {
+                return field.getAsDouble();
+            } else if (javaType == Slice.class) {
+                return field.getAsString();
+            } else {
+                return null;
+            }
+        } catch (Exception ex) {
+            log.error(ex);
+        }
+        return null;
+    }
+}