added support to query logging using kafka

docker-compose.yml

@@ -75,7 +75,7 @@ services:
       - "9092:9092"
     environment:
       KAFKA_ADVERTISED_HOST_NAME: kafka
-      KAFKA_CREATE_TOPICS: "test:1:1"
+      KAFKA_CREATE_TOPICS: "test:1:1,queries:1:1"
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock

presto/Dockerfile

@@ -1,3 +1,4 @@
 FROM starburstdata/presto
 COPY etc /usr/lib/presto/etc
+COPY lib usr/lib/presto/lib
 CMD /usr/lib/presto/bin/launcher run

presto/etc/catalog/kafka.properties

@@ -1,3 +1,3 @@
 connector.name=kafka
-kafka.table-names=test,hook
+kafka.table-names=test,hook,queries
 kafka.nodes=kafka:9092

presto/etc/event-listener.properties

@@ -0,0 +1,4 @@
+event-listener.name=presto2kafta-query-logging
+kafka-topic-name=queries
+kafka-broker-list=kafka:9092
+service-name=poc-presto

presto/etc/kafka/default.queries.json

@@ -0,0 +1,70 @@
+{
+    "tableName": "queries",
+    "schemaName": "default",
+    "topicName": "queries",
+    "message": {
+        "dataFormat": "json",
+        "fields": [
+             {
+                "name": "service_name",
+                "mapping": "service_name",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "query_id",
+		"mapping": "query_id",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "cpu_time",
+		"mapping": "cpu_time",
+                "type": "BIGINT"
+             },
+             {
+		"name": "wall_time",
+		"mapping": "wall_time",
+                "type": "BIGINT"
+             },
+             {
+		"name": "start_time",
+		"mapping": "start_time",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "end_time",
+		"mapping": "end_time",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "queued_time",
+		"mapping": "queued_time",
+                "type": "BIGINT"
+             },
+             {
+		"name": "query_text",
+		"mapping": "query_text",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "query_status",
+		"mapping": "query_status",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "failure_message",
+		"mapping": "failure_message",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "user",
+		"mapping": "user",
+                "type": "VARCHAR"
+             },
+             {
+		"name": "event_timestamp",
+		"mapping": "event_timestamp",
+                "type": "VARCHAR"
+             }
+        ]
+    }
+}