!1454 Add docs about extension execution planner

Merge pull request !1454 from zhousipei/branch1.6_add_docs

hetu-carbondata/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>io.hetu.core</groupId>
     <artifactId>presto-root</artifactId>
-    <version>1.6.0-SNAPSHOT</version>
+    <version>1.6.2-SNAPSHOT</version>
   </parent>
 
   <artifactId>hetu-carbondata</artifactId>

hetu-clickhouse/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-clickhouse</artifactId>

hetu-common/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-common</artifactId>

hetu-cube/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-cube</artifactId>

hetu-datacenter/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-datacenter</artifactId>

hetu-docs/en/admin/extension-execution-planner.md

@@ -0,0 +1,25 @@
+#Extension Physical Execution Planner
+This section describes how to add an extension physical execution planner in openLooKeng. With the extension physical execution planner, openLooKeng can utilize other operator acceleration libraries to speed up the execution of SQL statements.
+
+##Configuration
+To enable extension physical execution feature, the following configs must be added in
+`config.properties`：
+
+``` properties
+extension_execution_planner_enabled=true
+extension_execution_planner_jar_path=file:///xxPath/omni-openLooKeng-adapter-1.6.1-SNAPSHOT.jar
+extension_execution_planner_class_path=nova.hetu.olk.OmniLocalExecutionPlanner
+```
+
+The above attributes are described below:
+
+- `extension_execution_planner_enabled`: Enable extension physical execution feature.
+- `extension_execution_planner_jar_path`: Set the file path of the extension physical execution jar package.
+- `extension_execution_planner_class_path`: Set the package path of extension physical execution generated class in jar。
+
+
+##Usage
+The below command can control the enablement of extension physical execution feature in WebUI or Cli while running openLooKeng:
+```
+set session extension_execution_planner_enabled=true/false
+```

hetu-docs/en/develop/connectors.md

@@ -21,7 +21,7 @@ This interface is too big to list in this documentation, but if you are interest
 connector. If your underlying data source supports schemas, tables and columns, this interface should be straightforward to implement. If you are attempting to adapt something that is not a relational database (as
 the Example HTTP connector does), you may need to get creative about how you map your data source to openLooKeng\'s schema, table, and column concepts.
 
-### ConnectorSplitManger
+### ConnectorSplitManager
 
 The split manager partitions the data for a table into the individual chunks that openLooKeng will distribute to workers for processing. For example, the Hive connector lists the files for each Hive partition and creates
 one or more split per file. For data sources that don\'t have partitioned data, a good strategy here is to simply return a single split for the entire table. This is the strategy employed by the Example HTTP connector.

hetu-docs/en/index.md

@@ -46,6 +46,7 @@ headless: true
     - [Audit Log]({{< relref "./docs/admin/audit-log.md" >}})
     - [Reliable Execution]({{< relref "./docs/admin/reliable-execution.md" >}})
     - [JDBC Data Source Multi-Split Management]({{< relref "./docs/admin/multi-split-for-jdbc-data-source.md" >}})
+    - [Extension Physical Execution Planner]({{< relref "./docs/admin/extension-execution-planner.md" >}})
 
 - [Query Optimizer]("#")
     - [Table Statistics]({{< relref "./docs/optimizer/statistics.md" >}})
@@ -172,6 +173,7 @@ headless: true
     - [SHOW CACHE]({{< relref "./docs/sql/show-cache.md" >}})
     - [SHOW CATALOGS]({{< relref "./docs/sql/show-catalogs.md" >}})
     - [SHOW COLUMNS]({{< relref "./docs/sql/show-columns.md" >}})
+    - [SHOW CREATE CUBE]({{< relref "./docs/sql/show-create-cube.md" >}})
     - [SHOW CREATE TABLE]({{< relref "./docs/sql/show-create-table.md" >}})
     - [SHOW CREATE VIEW]({{< relref "./docs/sql/show-create-view.md" >}})
     - [SHOW FUNCTIONS]({{< relref "./docs/sql/show-functions.md" >}})

hetu-docs/zh/admin/extension-execution-planner.md

@@ -0,0 +1,24 @@
+#扩展物理执行计划
+本节介绍openLooKeng如何添加扩展物理执行计划。通过物理执行计划的扩展，openLooKeng可以使用其他算子加速库来加速SQL语句的执行。
+
+##配置
+在配置文件`config.properties`增加如下配置：
+
+``` properties
+extension_execution_planner_enabled=true
+extension_execution_planner_jar_path=file:///xxPath/omni-openLooKeng-adapter-1.6.1-SNAPSHOT.jar
+extension_execution_planner_class_path=nova.hetu.olk.OmniLocalExecutionPlanner
+```
+
+上述属性说明如下：
+
+- `extension_execution_planner_enabled`：是否开启扩展物理执行计划特性。
+- `extension_execution_planner_jar_path`：指定扩展jar包的文件路径。
+- `extension_execution_planner_class_path`：指定扩展jar包中执行计划生成类的包路径。
+
+
+##使用
+当运行openLooKeng时，可在WebUI或Cli中通过如下命令控制扩展物理执行计划的开启:
+```
+set session extension_execution_planner_enabled=true/false
+```

hetu-docs/zh/develop/connectors.md

@@ -31,7 +31,7 @@
 
 
 
-### ConnectorSplitManger
+### ConnectorSplitManager
 
 分片管理器将表的数据分区成多个块，这些块由 openLooKeng 分发至工作节点进行处理。
 

hetu-docs/zh/index.md

@@ -46,6 +46,8 @@ headless: true
     - [审计日志]({{< relref "./docs/admin/audit-log.md" >}})
     - [可靠查询执行]({{< relref "./docs/admin/reliable-execution.md" >}})
     - [JDBC数据源多分片管理]({{< relref "./docs/admin/multi-split-for-jdbc-data-source.md" >}})
+    - [扩展物理执行计划]({{< relref "./docs/admin/extension-execution-planner.md" >}})
+
 - [查询优化器]("#")
     - [表统计]({{< relref "./docs/optimizer/statistics.md" >}})
     - [EXPLAIN成本]({{< relref "./docs/optimizer/cost-in-explain.md" >}})
@@ -171,6 +173,7 @@ headless: true
     - [SHOW CACHE]({{< relref "./docs/sql/show-cache.md" >}})
     - [SHOW CATALOGS]({{< relref "./docs/sql/show-catalogs.md" >}})
     - [SHOW COLUMNS]({{< relref "./docs/sql/show-columns.md" >}})
+    - [SHOW CREATE CUBE]({{< relref "./docs/sql/show-create-cube.md" >}})
     - [SHOW CREATE TABLE]({{< relref "./docs/sql/show-create-table.md" >}})
     - [SHOW CREATE VIEW]({{< relref "./docs/sql/show-create-view.md" >}})
     - [SHOW FUNCTIONS]({{< relref "./docs/sql/show-functions.md" >}})

hetu-filesystem-client/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-function-namespace-managers/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-function-namespace-managers</artifactId>

hetu-greenplum/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-hana/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-hana</artifactId>

hetu-hazelcast/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-hazelcast</artifactId>

hetu-hbase/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-hbase</artifactId>

hetu-heuristic-index/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-hive-functions/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-kylin/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-kylin</artifactId>

hetu-listener/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-metastore/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-mongodb/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-opengauss/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -157,7 +157,7 @@
         <dependency>
             <groupId>io.hetu.core</groupId>
             <artifactId>presto-postgresql</artifactId>
-            <version>1.6.0-SNAPSHOT</version>
+            <version>1.6.2-SNAPSHOT</version>
             <scope>compile</scope>
             <exclusions>
                 <exclusion>
@@ -176,7 +176,7 @@
         <dependency>
             <groupId>io.hetu.core</groupId>
             <artifactId>presto-postgresql</artifactId>
-            <version>1.6.0-SNAPSHOT</version>
+            <version>1.6.2-SNAPSHOT</version>
             <classifier>tests</classifier>
             <scope>test</scope>
             <exclusions>

hetu-oracle/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-oracle</artifactId>

hetu-seed-store/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-server-rpm/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-server-rpm</artifactId>

hetu-server/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-server</artifactId>

hetu-sql-migration-tool/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-sql-migration-tool</artifactId>

hetu-startree/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-startree</artifactId>

hetu-state-store/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-transport/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hetu-vdm/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-vdm</artifactId>

pom.xml

@@ -10,7 +10,7 @@
 
     <groupId>io.hetu.core</groupId>
     <artifactId>presto-root</artifactId>
-    <version>1.6.0-SNAPSHOT</version>
+    <version>1.6.2-SNAPSHOT</version>
     <packaging>pom</packaging>
 
     <name>presto-root</name>
@@ -1870,4 +1870,8 @@
             </build>
         </profile>
     </profiles>
+
+  <scm>
+    <tag>branch-1.6</tag>
+  </scm>
 </project>

presto-array/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-array</artifactId>

presto-atop/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-atop</artifactId>

presto-base-jdbc/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-base-jdbc</artifactId>

presto-benchmark-driver/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-benchmark-driver</artifactId>

presto-benchmark/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-benchmark</artifactId>

presto-benchto-benchmarks/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-benchto-benchmarks</artifactId>

presto-cli/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-cli</artifactId>

presto-client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-client</artifactId>

presto-elasticsearch/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-elasticsearch</artifactId>

presto-example-http/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-example-http</artifactId>

presto-expressions/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-expressions</artifactId>

presto-geospatial-toolkit/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-geospatial-toolkit</artifactId>

presto-geospatial/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-geospatial</artifactId>

presto-hive-hadoop2/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-hive-hadoop2</artifactId>

presto-hive/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-hive</artifactId>

presto-jdbc/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>hetu-jdbc</artifactId>

presto-jmx/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-jmx</artifactId>

presto-kafka/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-kafka</artifactId>

presto-local-file/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-local-file</artifactId>

presto-main/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-main</artifactId>

presto-main/src/main/java/io/prestosql/SystemSessionProperties.java

@@ -37,6 +37,9 @@ import java.util.Optional;
 import java.util.OptionalInt;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static io.prestosql.spi.HetuConstant.EXTENSION_EXECUTION_PLANNER_CLASS_PATH;
+import static io.prestosql.spi.HetuConstant.EXTENSION_EXECUTION_PLANNER_ENABLED;
+import static io.prestosql.spi.HetuConstant.EXTENSION_EXECUTION_PLANNER_JAR_PATH;
 import static io.prestosql.spi.StandardErrorCode.INVALID_SESSION_PROPERTY;
 import static io.prestosql.spi.session.PropertyMetadata.booleanProperty;
 import static io.prestosql.spi.session.PropertyMetadata.dataSizeProperty;
@@ -810,6 +813,22 @@ public final class SystemSessionProperties
                         SKIP_NON_APPLICABLE_RULES_ENABLED,
                         "Whether to skip applying some selected rules based on query pattern",
                         featuresConfig.isSkipNonApplicableRulesEnabled(),
+                        false),
+                // add extension execution planner and operator
+                stringProperty(
+                        EXTENSION_EXECUTION_PLANNER_JAR_PATH,
+                        "extension execution planner jar path",
+                        hetuConfig.getExtensionExecutionPlannerJarPath(),
+                        false),
+                stringProperty(
+                        EXTENSION_EXECUTION_PLANNER_CLASS_PATH,
+                        "extension execution planner class path",
+                        hetuConfig.getExtensionExecutionPlannerClassPath(),
+                        false),
+                booleanProperty(
+                        EXTENSION_EXECUTION_PLANNER_ENABLED,
+                        "extension execution planner enabled",
+                        hetuConfig.getExtensionExecutionPlannerEnabled(),
                         false));
     }
 
@@ -1418,4 +1437,19 @@ public final class SystemSessionProperties
     {
         return session.getSystemProperty(SKIP_NON_APPLICABLE_RULES_ENABLED, Boolean.class);
     }
+
+    public static Boolean isExtensionExecutionPlannerEnabled(Session session)
+    {
+        return session.getSystemProperty(EXTENSION_EXECUTION_PLANNER_ENABLED, Boolean.class);
+    }
+
+    public static String getExtensionExecutionPlannerJarPath(Session session)
+    {
+        return session.getSystemProperty(EXTENSION_EXECUTION_PLANNER_JAR_PATH, String.class);
+    }
+
+    public static String getExtensionExecutionPlannerClassPath(Session session)
+    {
+        return session.getSystemProperty(EXTENSION_EXECUTION_PLANNER_CLASS_PATH, String.class);
+    }
 }

presto-main/src/main/java/io/prestosql/execution/SqlTaskExecutionFactory.java

@@ -14,6 +14,7 @@
 package io.prestosql.execution;
 
 import io.airlift.concurrent.SetThreadName;
+import io.airlift.log.Logger;
 import io.hetu.core.transport.execution.buffer.PagesSerdeFactory;
 import io.prestosql.Session;
 import io.prestosql.event.SplitMonitor;
@@ -23,12 +24,19 @@ import io.prestosql.memory.QueryContext;
 import io.prestosql.metadata.Metadata;
 import io.prestosql.operator.CommonTableExecutionContext;
 import io.prestosql.operator.TaskContext;
+import io.prestosql.spi.PrestoException;
 import io.prestosql.spi.plan.PlanNodeId;
 import io.prestosql.sql.planner.LocalExecutionPlanner;
 import io.prestosql.sql.planner.LocalExecutionPlanner.LocalExecutionPlan;
 import io.prestosql.sql.planner.PlanFragment;
 import io.prestosql.sql.planner.TypeProvider;
 
+import javax.annotation.Nullable;
+
+import java.lang.reflect.Constructor;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -36,12 +44,19 @@ import java.util.OptionalInt;
 import java.util.concurrent.Executor;
 
 import static com.google.common.base.Throwables.throwIfUnchecked;
+import static io.prestosql.SystemSessionProperties.getExtensionExecutionPlannerClassPath;
+import static io.prestosql.SystemSessionProperties.getExtensionExecutionPlannerJarPath;
 import static io.prestosql.SystemSessionProperties.isExchangeCompressionEnabled;
+import static io.prestosql.SystemSessionProperties.isExtensionExecutionPlannerEnabled;
 import static io.prestosql.execution.SqlTaskExecution.createSqlTaskExecution;
+import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static java.util.Objects.requireNonNull;
 
 public class SqlTaskExecutionFactory
 {
+    private static final Logger log = Logger.get(SqlTaskExecutionFactory.class);
+    private static LocalExecutionPlanner extensionPlanner;
+    private static boolean extensionPlannerInitialized;
     private final Executor taskNotificationExecutor;
 
     private final TaskExecutor taskExecutor;
@@ -83,20 +98,32 @@ public class SqlTaskExecutionFactory
                 consumer,
                 new PagesSerdeFactory(metadata.getFunctionAndTypeManager().getBlockEncodingSerde(), isExchangeCompressionEnabled(session)));
 
-        LocalExecutionPlan localExecutionPlan;
+        LocalExecutionPlan localExecutionPlan = null;
         try (SetThreadName ignored = new SetThreadName("Task-%s", taskStateMachine.getTaskId())) {
             try {
-                localExecutionPlan = planner.plan(
-                        taskContext,
-                        fragment.getRoot(),
-                        TypeProvider.copyOf(fragment.getSymbols()),
-                        fragment.getPartitioningScheme(),
-                        fragment.getStageExecutionDescriptor(),
-                        fragment.getPartitionedSources(),
-                        outputBuffer,
-                        fragment.getFeederCTEId(),
-                        fragment.getFeederCTEParentId(),
-                        cteCtx);
+                if (isExtensionExecutionPlannerEnabled(session)) {
+                    String jarPath = getExtensionExecutionPlannerJarPath(session);
+                    String classPath = getExtensionExecutionPlannerClassPath(session);
+                    if (jarPath != null && !jarPath.equals("") && classPath != null && !classPath.equals("")) {
+                        localExecutionPlan = loadExtensionLocalExecutionPlan(outputBuffer, fragment, taskContext, cteCtx, jarPath, classPath);
+                    }
+                    else {
+                        throw new PrestoException(GENERIC_INTERNAL_ERROR, "Extension execution planner jar path or class path isn't configured correctly");
+                    }
+                }
+                if (localExecutionPlan == null) {
+                    localExecutionPlan = planner.plan(
+                            taskContext,
+                            fragment.getRoot(),
+                            TypeProvider.copyOf(fragment.getSymbols()),
+                            fragment.getPartitioningScheme(),
+                            fragment.getStageExecutionDescriptor(),
+                            fragment.getPartitionedSources(),
+                            outputBuffer,
+                            fragment.getFeederCTEId(),
+                            fragment.getFeederCTEParentId(),
+                            cteCtx);
+                }
             }
             catch (Throwable e) {
                 // planning failed
@@ -115,4 +142,46 @@ public class SqlTaskExecutionFactory
                 taskNotificationExecutor,
                 splitMonitor);
     }
+
+    @Nullable
+    private LocalExecutionPlan loadExtensionLocalExecutionPlan(OutputBuffer outputBuffer, PlanFragment fragment, TaskContext taskContext, Map<String, CommonTableExecutionContext> cteCtx, String jarPath, String classPath)
+    {
+        if (!extensionPlannerInitialized) {
+            try {
+                ExtensionClassLoader extensionClassLoader = new ExtensionClassLoader(jarPath, Thread.currentThread().getContextClassLoader());
+                Thread.currentThread().setContextClassLoader(extensionClassLoader);
+                Class<?> aClass = extensionClassLoader.loadClass(classPath);
+                Constructor<?> constructor = aClass.getConstructor(LocalExecutionPlanner.class);
+                extensionPlanner = (LocalExecutionPlanner) constructor.newInstance(planner);
+                extensionPlannerInitialized = true;
+            }
+            catch (Throwable e) {
+                log.warn("get extension LocalExecutionPlanner failed: %s", e.toString());
+                throw new PrestoException(GENERIC_INTERNAL_ERROR, e);
+            }
+        }
+        if (extensionPlanner != null) {
+            return extensionPlanner.plan(
+                    taskContext,
+                    fragment.getRoot(),
+                    TypeProvider.copyOf(fragment.getSymbols()),
+                    fragment.getPartitioningScheme(),
+                    fragment.getStageExecutionDescriptor(),
+                    fragment.getPartitionedSources(),
+                    outputBuffer,
+                    fragment.getFeederCTEId(),
+                    fragment.getFeederCTEParentId(),
+                    cteCtx);
+        }
+        return null;
+    }
+
+    public static class ExtensionClassLoader
+            extends URLClassLoader
+    {
+        public ExtensionClassLoader(final String path, ClassLoader parent) throws MalformedURLException
+        {
+            super(new URL[] {new URL(path)}, parent);
+        }
+    }
 }

presto-main/src/main/java/io/prestosql/execution/StateMachine.java

@@ -49,6 +49,7 @@ public class StateMachine<T>
     private final Executor executor;
     private final Object lock = new Object();
     private final Set<T> terminalStates;
+    private StateChangeListener tailStateChangeListener;
 
     @GuardedBy("lock")
     private volatile T state;
@@ -279,6 +280,12 @@ public class StateMachine<T>
             inTerminalState = isTerminalState(currentState);
             if (!inTerminalState) {
                 stateChangeListeners.add(stateChangeListener);
+                if (tailStateChangeListener != null) {
+                    if (stateChangeListeners.contains(tailStateChangeListener)) {
+                        stateChangeListeners.remove(tailStateChangeListener);
+                    }
+                    stateChangeListeners.add(tailStateChangeListener);
+                }
             }
         }
 
@@ -287,6 +294,12 @@ public class StateMachine<T>
         safeExecute(() -> stateChangeListener.stateChanged(currentState));
     }
 
+    public void addStateChangeListenerToTail(StateChangeListener<T> stateChangeListener)
+    {
+        tailStateChangeListener = stateChangeListener;
+        addStateChangeListener(stateChangeListener);
+    }
+
     @VisibleForTesting
     boolean isTerminalState(T state)
     {

presto-main/src/main/java/io/prestosql/execution/TaskStateMachine.java

@@ -121,6 +121,15 @@ public class TaskStateMachine
         taskState.addStateChangeListener(stateChangeListener);
     }
 
+    /**
+     * Add listener to the tail, this listener will be notified at last when state changed.
+     * @param stateChangeListener listener of state change.
+     */
+    public void addStateChangeListenerToTail(StateChangeListener<TaskState> stateChangeListener)
+    {
+        taskState.addStateChangeListenerToTail(stateChangeListener);
+    }
+
     @Override
     public String toString()
     {

presto-main/src/main/java/io/prestosql/metadata/FunctionAndTypeManager.java

@@ -213,7 +213,7 @@ public class FunctionAndTypeManager
             Map<String, String> properties)
     {
         requireNonNull(hetuMetaStoreManager, "hetuMetaStoreManager is nll");
-        FunctionNamespaceManagerContextInstance functionNamespaceManagerContextInstance = new FunctionNamespaceManagerContextInstance(hetuMetaStoreManager.getHetuMetastore());
+        FunctionNamespaceManagerContextInstance functionNamespaceManagerContextInstance = new FunctionNamespaceManagerContextInstance(hetuMetaStoreManager.getHetuMetastore(), this);
         requireNonNull(functionNamespaceManagerName, "functionNamespaceManagerName is null");
         FunctionNamespaceManagerFactory factory = functionNamespaceManagerFactories.get(functionNamespaceManagerName);
         checkState(factory != null, "No factory for function namespace manager %s", functionNamespaceManagerName);
@@ -239,6 +239,11 @@ public class FunctionAndTypeManager
         return functionInvokerProvider;
     }
 
+    public Map<String, FunctionNamespaceManager<? extends SqlFunction>> getFunctionNamespaceManagers()
+    {
+        return functionNamespaceManagers;
+    }
+
     public void addFunctionNamespaceFactory(FunctionNamespaceManagerFactory factory)
     {
         if (functionNamespaceManagerFactories.putIfAbsent(factory.getName(), factory) != null) {
@@ -539,6 +544,11 @@ public class FunctionAndTypeManager
 
         Optional<FunctionNamespaceTransactionHandle> transactionHandle = transactionId
                 .map(id -> transactionManager.getFunctionNamespaceTransaction(id, functionName.getCatalogSchemaName().getCatalogName()));
+
+        if (functionNamespaceManager.canResolveFunction()) {
+            return functionNamespaceManager.resolveFunction(transactionHandle, functionName,
+                    parameterTypes.stream().map(TypeSignatureProvider::getTypeSignature).collect(toImmutableList()));
+        }
         Collection<? extends SqlFunction> candidates = functionNamespaceManager.getFunctions(transactionHandle, functionName);
 
         return functionResolver.resolveFunction(functionNamespaceManager, transactionHandle, functionName, parameterTypes, candidates);

presto-main/src/main/java/io/prestosql/metadata/FunctionNamespaceManagerContextInstance.java

@@ -15,6 +15,7 @@ package io.prestosql.metadata;
 
 import io.prestosql.spi.function.FunctionNamespaceManagerContext;
 import io.prestosql.spi.metastore.HetuMetastore;
+import io.prestosql.spi.type.TypeManager;
 
 import java.util.Optional;
 
@@ -23,9 +24,12 @@ public class FunctionNamespaceManagerContextInstance
 {
     private final HetuMetastore hetuMetastore;
 
-    public FunctionNamespaceManagerContextInstance(HetuMetastore hetuMetastore)
+    private final TypeManager typeManager;
+
+    public FunctionNamespaceManagerContextInstance(HetuMetastore hetuMetastore, TypeManager typeManager)
     {
         this.hetuMetastore = hetuMetastore;
+        this.typeManager = typeManager;
     }
 
     @Override
@@ -36,4 +40,13 @@ public class FunctionNamespaceManagerContextInstance
         }
         return Optional.of(hetuMetastore);
     }
+
+    @Override
+    public Optional<TypeManager> getTypeManager()
+    {
+        if (typeManager == null) {
+            return Optional.empty();
+        }
+        return Optional.of(typeManager);
+    }
 }

presto-main/src/main/java/io/prestosql/metadata/InternalBlockEncodingSerde.java

@@ -24,7 +24,7 @@ import java.util.Optional;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Objects.requireNonNull;
 
-final class InternalBlockEncodingSerde
+public class InternalBlockEncodingSerde
         implements BlockEncodingSerde
 {
     private final FunctionAndTypeManager functionAndTypeManager;

presto-main/src/main/java/io/prestosql/operator/EmptyLookupSource.java

@@ -16,7 +16,7 @@ package io.prestosql.operator;
 import io.prestosql.spi.Page;
 import io.prestosql.spi.PageBuilder;
 
-final class EmptyLookupSource
+public final class EmptyLookupSource
         implements LookupSource
 {
     @Override

presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java

@@ -24,6 +24,7 @@ import io.airlift.log.Logger;
 import io.airlift.units.DataSize;
 import io.airlift.units.Duration;
 import io.hetu.core.transport.execution.buffer.PageCodecMarker;
+import io.hetu.core.transport.execution.buffer.PagesSerde;
 import io.hetu.core.transport.execution.buffer.SerializedPage;
 import io.prestosql.failuredetector.FailureDetector;
 import io.prestosql.memory.context.LocalMemoryContext;
@@ -31,6 +32,7 @@ import io.prestosql.operator.HttpPageBufferClient.ClientCallback;
 import io.prestosql.operator.WorkProcessor.ProcessState;
 import io.prestosql.snapshot.MultiInputSnapshotState;
 import io.prestosql.snapshot.QuerySnapshotManager;
+import io.prestosql.spi.Page;
 import io.prestosql.spi.snapshot.BlockEncodingSerdeProvider;
 import org.apache.commons.lang3.tuple.Pair;
 
@@ -122,6 +124,8 @@ public class ExchangeClient
     @GuardedBy("this")
     private long averageBytesPerRequest;
 
+    private List<Page> pages = new ArrayList<>();
+
     private final AtomicBoolean closed = new AtomicBoolean();
     private final AtomicReference<Throwable> failure = new AtomicReference<>();
 
@@ -343,6 +347,19 @@ public class ExchangeClient
                 });
     }
 
+    public List<Page> getPages(String target, PagesSerde pagesSerde)
+    {
+        SerializedPage serializedPage = pollPage(target).getLeft();
+        if (serializedPage == null) {
+            if (isFinished()) {
+                return pages;
+            }
+            return null;
+        }
+        pages.add(pagesSerde.deserialize(serializedPage));
+        return null;
+    }
+
     @Nullable
     public Pair<SerializedPage, String> pollPage(String target)
     {

presto-main/src/main/java/io/prestosql/operator/HashAggregationOperator.java

@@ -117,7 +117,7 @@ public class HashAggregationOperator
         }
 
         @VisibleForTesting
-        HashAggregationOperatorFactory(
+        public HashAggregationOperatorFactory(
                 int operatorId,
                 PlanNodeId planNodeId,
                 List<? extends Type> groupByTypes,

presto-main/src/main/java/io/prestosql/operator/Operator.java

@@ -66,7 +66,10 @@ public interface Operator
     /**
      * For Snapshot - If next output is a marker page, then return it, otherwise return null
      */
-    Page pollMarker();
+    default Page pollMarker()
+    {
+        return null;
+    }
 
     /**
      * After calling this method operator should revoke all reserved revocable memory.

presto-main/src/main/java/io/prestosql/operator/OperatorFactory.java

@@ -13,7 +13,11 @@
  */
 package io.prestosql.operator;
 
+import com.google.common.collect.ImmutableList;
 import io.prestosql.execution.Lifespan;
+import io.prestosql.spi.type.Type;
+
+import java.util.List;
 
 public interface OperatorFactory
 {
@@ -47,4 +51,14 @@ public interface OperatorFactory
     }
 
     OperatorFactory duplicate();
+
+    default boolean isExtensionOperatorFactory()
+    {
+        return false;
+    }
+
+    default List<Type> getSourceTypes()
+    {
+        return ImmutableList.of();
+    }
 }

presto-main/src/main/java/io/prestosql/operator/PageUtils.java

@@ -19,13 +19,13 @@ import io.prestosql.spi.block.LazyBlock;
 
 import java.util.function.LongConsumer;
 
-final class PageUtils
+public final class PageUtils
 {
     private PageUtils()
     {
     }
 
-    static <T> Page recordMaterializedBytes(Page page, LongConsumer sizeInBytesConsumer)
+    public static <T> Page recordMaterializedBytes(Page page, LongConsumer sizeInBytesConsumer)
     {
         // account processed bytes from lazy blocks only when they are loaded
         Block<T>[] blocks = new Block[page.getChannelCount()];

presto-main/src/main/java/io/prestosql/operator/ScanFilterAndProjectOperator.java

@@ -493,7 +493,12 @@ public class ScanFilterAndProjectOperator
             page = recordMaterializedBytes(page, sizeInBytes -> processedBytes += sizeInBytes);
 
             // update operator stats
-            processedPositions += page.getPositionCount();
+            if (pageSource.getCompletedPositionCount().isPresent()) {
+                processedPositions = pageSource.getCompletedPositionCount().getAsLong();
+            }
+            else {
+                processedPositions += page.getPositionCount();
+            }
             physicalBytes = pageSource.getCompletedBytes();
             readTimeNanos = pageSource.getReadTimeNanos();
 

presto-main/src/main/java/io/prestosql/operator/TableScanOperator.java

@@ -267,6 +267,8 @@ public class TableScanOperator
 
     private boolean finished;
 
+    // completedPositionCount is used only if connectorPageSource.getCompletedPositionCount is present.
+    private long completedPositionCount;
     private long completedBytes;
     private long readTimeNanos;
     Optional<TableScanNode> tableScanNodeOptional;
@@ -707,17 +709,34 @@ public class TableScanOperator
         }
 
         Page page = source.getNextPage();
+
+        // if pageSource.getCompletedPositionCount is present, get operator statistics from pageSource
+        if (source.getCompletedPositionCount().isPresent()) {
+            long endCompletedPositionCount = source.getCompletedPositionCount().getAsLong();
+            long endCompletedBytes = source.getCompletedBytes();
+            long endReadTimeNanos = source.getReadTimeNanos();
+            long currentPositionCount = endCompletedPositionCount - completedPositionCount;
+            long currentCompletedBytes = endCompletedBytes - completedBytes;
+            operatorContext.recordPhysicalInputWithTiming(currentCompletedBytes, currentPositionCount, endReadTimeNanos - readTimeNanos);
+            operatorContext.recordProcessedInput(currentCompletedBytes, currentPositionCount);
+            completedPositionCount = endCompletedPositionCount;
+            completedBytes = endCompletedBytes;
+            readTimeNanos = endReadTimeNanos;
+        }
+
         if (page != null) {
             // assure the page is in memory before handing to another operator
             page = page.getLoadedPage();
 
             // update operator stats
-            long endCompletedBytes = source.getCompletedBytes();
-            long endReadTimeNanos = source.getReadTimeNanos();
-            operatorContext.recordPhysicalInputWithTiming(endCompletedBytes - completedBytes, page.getPositionCount(), endReadTimeNanos - readTimeNanos);
-            operatorContext.recordProcessedInput(page.getSizeInBytes(), page.getPositionCount());
-            completedBytes = endCompletedBytes;
-            readTimeNanos = endReadTimeNanos;
+            if (!source.getCompletedPositionCount().isPresent()) {
+                long endCompletedBytes = source.getCompletedBytes();
+                long endReadTimeNanos = source.getReadTimeNanos();
+                operatorContext.recordPhysicalInputWithTiming(endCompletedBytes - completedBytes, page.getPositionCount(), endReadTimeNanos - readTimeNanos);
+                operatorContext.recordProcessedInput(page.getSizeInBytes(), page.getPositionCount());
+                completedBytes = endCompletedBytes;
+                readTimeNanos = endReadTimeNanos;
+            }
 
             // pull bloomFilter from stateStore and filter page
             if (existsCrossFilter) {

presto-main/src/main/java/io/prestosql/operator/TaskContext.java

@@ -40,7 +40,9 @@ import org.joda.time.DateTime;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.OptionalInt;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -111,6 +113,8 @@ public class TaskContext
     private final PagesSerdeFactory serdeFactory;
     private final TaskSnapshotManager snapshotManager;
 
+    private final Map<String, Object> taskExtendProperties = new HashMap<>();
+
     public static TaskContext createTaskContext(
             QueryContext queryContext,
             TaskStateMachine taskStateMachine,
@@ -185,6 +189,16 @@ public class TaskContext
         return snapshotManager;
     }
 
+    public TaskStateMachine getTaskStateMachine()
+    {
+        return taskStateMachine;
+    }
+
+    public Map<String, Object> getTaskExtendProperties()
+    {
+        return taskExtendProperties;
+    }
+
     public PipelineContext addPipelineContext(int pipelineId, boolean inputPipeline, boolean outputPipeline, boolean partitioned)
     {
         PipelineContext pipelineContext = new PipelineContext(

presto-main/src/main/java/io/prestosql/operator/WindowFunctionDefinition.java

@@ -64,6 +64,16 @@ public class WindowFunctionDefinition
         return type;
     }
 
+    public WindowFunctionSupplier getFunctionSupplier()
+    {
+        return functionSupplier;
+    }
+
+    public List<Integer> getArgumentChannels()
+    {
+        return argumentChannels;
+    }
+
     public WindowFunction createWindowFunction()
     {
         return functionSupplier.createWindowFunction(argumentChannels);

presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchange.java

@@ -66,11 +66,11 @@ public class LocalExchange
 {
     private static final Logger LOG = Logger.get(LocalExchange.class);
 
-    private final Supplier<LocalExchanger> exchangerSupplier;
+    protected Supplier<LocalExchanger> exchangerSupplier;
 
-    private final List<LocalExchangeSource> sources;
+    protected final List<LocalExchangeSource> sources;
 
-    private final LocalExchangeMemoryManager memoryManager;
+    protected final LocalExchangeMemoryManager memoryManager;
 
     @GuardedBy("this")
     private boolean allSourcesFinished;
@@ -352,27 +352,27 @@ public class LocalExchange
     @ThreadSafe
     public static class LocalExchangeFactory
     {
-        private final PartitioningHandle partitioning;
-        private final List<Type> types;
-        private final List<Integer> partitionChannels;
-        private final Optional<Integer> partitionHashChannel;
-        private final PipelineExecutionStrategy exchangeSourcePipelineExecutionStrategy;
-        private final DataSize maxBufferedBytes;
-        private final int bufferCount;
-        private final boolean isForMerge;
-        private final AggregationNode.AggregationType aggregationType;
+        protected final PartitioningHandle partitioning;
+        protected final List<Type> types;
+        protected final List<Integer> partitionChannels;
+        protected final Optional<Integer> partitionHashChannel;
+        protected final PipelineExecutionStrategy exchangeSourcePipelineExecutionStrategy;
+        protected final DataSize maxBufferedBytes;
+        protected final int bufferCount;
+        protected final boolean isForMerge;
+        protected final AggregationNode.AggregationType aggregationType;
 
         @GuardedBy("this")
-        private boolean noMoreSinkFactories;
+        protected boolean noMoreSinkFactories;
         // The number of total sink factories are tracked at planning time
         // so that the exact number of sink factory is known by the time execution starts.
         @GuardedBy("this")
-        private int numSinkFactories;
+        protected int numSinkFactories;
 
         @GuardedBy("this")
-        private final Map<Lifespan, LocalExchange> localExchangeMap = new HashMap<>();
+        protected final Map<Lifespan, LocalExchange> localExchangeMap = new HashMap<>();
         @GuardedBy("this")
-        private final List<LocalExchangeSinkFactoryId> closedSinkFactories = new ArrayList<>();
+        protected final List<LocalExchangeSinkFactoryId> closedSinkFactories = new ArrayList<>();
 
         public LocalExchangeFactory(
                 PartitioningHandle partitioning,
@@ -514,7 +514,7 @@ public class LocalExchange
     {
         private final LocalExchange exchange;
 
-        private LocalExchangeSinkFactory(LocalExchange exchange)
+        public LocalExchangeSinkFactory(LocalExchange exchange)
         {
             this.exchange = requireNonNull(exchange, "exchange is null");
         }

presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchangeSinkOperator.java

@@ -117,7 +117,7 @@ public class LocalExchangeSinkOperator
     private final Function<Page, Page> pagePreprocessor;
     private final SingleInputSnapshotState snapshotState;
 
-    LocalExchangeSinkOperator(String id, OperatorContext operatorContext, LocalExchangeSink sink, Function<Page, Page> pagePreprocessor)
+    public LocalExchangeSinkOperator(String id, OperatorContext operatorContext, LocalExchangeSink sink, Function<Page, Page> pagePreprocessor)
     {
         this.id = id;
         this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");

presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchangeSource.java

@@ -66,6 +66,8 @@ public class LocalExchangeSource
 
     private final Object lock = new Object();
 
+    private List<Page> pages = new ArrayList<>();
+
     @GuardedBy("lock")
     private SettableFuture<?> notEmptyFuture = NOT_EMPTY;
 
@@ -99,7 +101,7 @@ public class LocalExchangeSource
         return Collections.unmodifiableSet(inputChannels);
     }
 
-    void addPage(PageReference pageReference, String origin)
+    public void addPage(PageReference pageReference, String origin)
     {
         checkNotHoldsLock();
 
@@ -238,6 +240,19 @@ public class LocalExchangeSource
         return Pair.of(page, origin.orElse(null));
     }
 
+    public List<Page> getPages()
+    {
+        Page page = removePage().getLeft();
+        if (page == null) {
+            if (isFinished()) {
+                return pages;
+            }
+            return null;
+        }
+        pages.add(page);
+        return null;
+    }
+
     public ListenableFuture<?> waitForReading()
     {
         checkNotHoldsLock();

presto-main/src/main/java/io/prestosql/operator/exchange/PageReference.java

@@ -23,7 +23,7 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
 
 @ThreadSafe
-class PageReference
+public class PageReference
 {
     private final Page page;
     private final Runnable onFree;

presto-main/src/main/java/io/prestosql/server/PluginManager.java

@@ -89,6 +89,7 @@ public class PluginManager
             .add("io.airlift.units.")
             .add("org.openjdk.jol.")
             .add("io.prestosql.sql.tree.")
+            .add("nova.hetu.omniruntime.vector.")
             .build();
 
     private static final Logger log = Logger.get(PluginManager.class);

presto-main/src/main/java/io/prestosql/sql/InterpretedFunctionInvoker.java

@@ -19,6 +19,7 @@ import io.prestosql.spi.connector.ConnectorSession;
 import io.prestosql.spi.function.BuiltInScalarFunctionImplementation;
 import io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ArgumentProperty;
 import io.prestosql.spi.function.FunctionHandle;
+import io.prestosql.spi.function.ScalarFunctionImplementation;
 
 import java.lang.invoke.MethodHandle;
 import java.util.ArrayList;
@@ -29,6 +30,7 @@ import static com.google.common.base.Throwables.throwIfUnchecked;
 import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ArgumentType.VALUE_TYPE;
 import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.NullConvention.RETURN_NULL_ON_NULL;
 import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.NullConvention.USE_NULL_FLAG;
+import static io.prestosql.sql.gen.BytecodeUtils.getAllScalarFunctionImplementationChoices;
 import static java.lang.invoke.MethodHandleProxies.asInterfaceInstance;
 import static java.util.Objects.requireNonNull;
 
@@ -48,7 +50,7 @@ public class InterpretedFunctionInvoker
 
     public Object invoke(FunctionHandle functionHandle, ConnectorSession session, List<Object> arguments)
     {
-        return invoke(functionAndTypeManager.getBuiltInScalarFunctionImplementation(functionHandle), session, arguments);
+        return invoke(functionAndTypeManager.getScalarFunctionImplementation(functionHandle), session, arguments);
     }
 
     /**
@@ -56,8 +58,9 @@ public class InterpretedFunctionInvoker
      * <p>
      * Returns a value in the native container type corresponding to the declared SQL return type
      */
-    private Object invoke(BuiltInScalarFunctionImplementation function, ConnectorSession session, List<Object> arguments)
+    private Object invoke(ScalarFunctionImplementation function, ConnectorSession session, List<Object> arguments)
     {
+        BuiltInScalarFunctionImplementation.ScalarImplementationChoice choice = getAllScalarFunctionImplementationChoices(function).get(0);
         MethodHandle method = function.getMethodHandle();
 
         // handle function on instance method, to allow use of fields
@@ -69,15 +72,15 @@ public class InterpretedFunctionInvoker
         List<Object> actualArguments = new ArrayList<>();
         for (int i = 0; i < arguments.size(); i++) {
             Object argument = arguments.get(i);
-            ArgumentProperty argumentProperty = function.getArgumentProperty(i);
+            ArgumentProperty argumentProperty = choice.getArgumentProperty(i);
             if (argumentProperty.getArgumentType() == VALUE_TYPE) {
-                if (function.getArgumentProperty(i).getNullConvention() == RETURN_NULL_ON_NULL) {
+                if (choice.getArgumentProperty(i).getNullConvention() == RETURN_NULL_ON_NULL) {
                     if (argument == null) {
                         return null;
                     }
                     actualArguments.add(argument);
                 }
-                else if (function.getArgumentProperty(i).getNullConvention() == USE_NULL_FLAG) {
+                else if (choice.getArgumentProperty(i).getNullConvention() == USE_NULL_FLAG) {
                     boolean isNull = argument == null;
                     if (isNull) {
                         argument = Defaults.defaultValue(method.type().parameterType(actualArguments.size()));
@@ -103,14 +106,19 @@ public class InterpretedFunctionInvoker
         }
     }
 
-    private static MethodHandle bindInstanceFactory(MethodHandle method, BuiltInScalarFunctionImplementation implementation)
+    private static MethodHandle bindInstanceFactory(MethodHandle method, ScalarFunctionImplementation implementation)
     {
-        if (!implementation.getInstanceFactory().isPresent()) {
+        if (!(implementation instanceof BuiltInScalarFunctionImplementation)) {
+            return method;
+        }
+
+        BuiltInScalarFunctionImplementation builtInImplementation = (BuiltInScalarFunctionImplementation) implementation;
+        if (!builtInImplementation.getInstanceFactory().isPresent()) {
             return method;
         }
 
         try {
-            return method.bindTo(implementation.getInstanceFactory().get().invoke());
+            return method.bindTo(builtInImplementation.getInstanceFactory().get().invoke());
         }
         catch (Throwable throwable) {
             throw propagate(throwable);

presto-main/src/main/java/io/prestosql/sql/gen/BytecodeGeneratorContext.java

@@ -19,6 +19,7 @@ import io.airlift.bytecode.Scope;
 import io.airlift.bytecode.Variable;
 import io.prestosql.metadata.FunctionAndTypeManager;
 import io.prestosql.spi.function.BuiltInScalarFunctionImplementation;
+import io.prestosql.spi.function.ScalarFunctionImplementation;
 import io.prestosql.spi.relation.RowExpression;
 
 import java.util.List;
@@ -85,11 +86,11 @@ public class BytecodeGeneratorContext
     /**
      * Generates a function call with null handling, automatic binding of session parameter, etc.
      */
-    public BytecodeNode generateCall(String name, BuiltInScalarFunctionImplementation function, List<BytecodeNode> arguments)
+    public BytecodeNode generateCall(String name, ScalarFunctionImplementation function, List<BytecodeNode> arguments)
     {
         Optional<BytecodeNode> instance = Optional.empty();
-        if (function.getInstanceFactory().isPresent()) {
-            FieldDefinition field = cachedInstanceBinder.getCachedInstance(function.getInstanceFactory().get());
+        if (function instanceof BuiltInScalarFunctionImplementation && ((BuiltInScalarFunctionImplementation) function).getInstanceFactory().isPresent()) {
+            FieldDefinition field = cachedInstanceBinder.getCachedInstance(((BuiltInScalarFunctionImplementation) function).getInstanceFactory().get());
             instance = Optional.of(scope.getThis().getField(field));
         }
         return generateInvocation(scope, name, function, instance, arguments, callSiteBinder);

presto-main/src/main/java/io/prestosql/sql/gen/BytecodeUtils.java

@@ -31,6 +31,7 @@ import io.prestosql.spi.function.BuiltInScalarFunctionImplementation;
 import io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ArgumentProperty;
 import io.prestosql.spi.function.BuiltInScalarFunctionImplementation.NullConvention;
 import io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ScalarImplementationChoice;
+import io.prestosql.spi.function.ScalarFunctionImplementation;
 import io.prestosql.spi.type.Type;
 import io.prestosql.sql.gen.InputReferenceCompiler.InputReferenceNode;
 
@@ -42,11 +43,18 @@ import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static io.airlift.bytecode.OpCode.NOP;
 import static io.airlift.bytecode.expression.BytecodeExpressions.constantFalse;
 import static io.airlift.bytecode.expression.BytecodeExpressions.constantTrue;
 import static io.airlift.bytecode.expression.BytecodeExpressions.invokeDynamic;
+import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ArgumentProperty.valueTypeArgumentProperty;
 import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ArgumentType.VALUE_TYPE;
+import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.NullConvention.BLOCK_AND_POSITION;
+import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.NullConvention.RETURN_NULL_ON_NULL;
+import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.NullConvention.USE_BOXED_TYPE;
+import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.NullConvention.USE_NULL_FLAG;
+import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ReturnPlaceConvention.STACK;
 import static io.prestosql.sql.gen.Bootstrap.BOOTSTRAP_METHOD;
 import static java.lang.String.format;
 
@@ -159,14 +167,14 @@ public final class BytecodeUtils
                 binding.getType().returnType());
     }
 
-    public static BytecodeNode generateInvocation(Scope scope, String name, BuiltInScalarFunctionImplementation function, Optional<BytecodeNode> instance, List<BytecodeNode> arguments, CallSiteBinder binder)
+    public static BytecodeNode generateInvocation(Scope scope, String name, ScalarFunctionImplementation function, Optional<BytecodeNode> instance, List<BytecodeNode> arguments, CallSiteBinder binder)
     {
         LabelNode end = new LabelNode("end");
         BytecodeBlock block = new BytecodeBlock()
                 .setDescription("invoke " + name);
 
         List<Class<?>> stackTypes = new ArrayList<>();
-        if (function.getInstanceFactory().isPresent()) {
+        if (function instanceof BuiltInScalarFunctionImplementation && ((BuiltInScalarFunctionImplementation) function).getInstanceFactory().isPresent()) {
             checkArgument(instance.isPresent());
         }
 
@@ -177,7 +185,7 @@ public final class BytecodeUtils
         int realParameterIndex = 0;
 
         // Go through all the choices in the function and then pick the best one
-        List<ScalarImplementationChoice> choices = function.getAllChoices();
+        List<ScalarImplementationChoice> choices = getAllScalarFunctionImplementationChoices(function);
         ScalarImplementationChoice bestChoice = null;
         for (ScalarImplementationChoice currentChoice : choices) {
             boolean isValid = true;
@@ -373,4 +381,30 @@ public final class BytecodeUtils
                                 .getVariable(tempValue)
                                 .invokeInterface(Type.class, methodName, void.class, BlockBuilder.class, valueJavaType)));
     }
+
+    public static List<ScalarImplementationChoice> getAllScalarFunctionImplementationChoices(ScalarFunctionImplementation function)
+    {
+        if (function instanceof BuiltInScalarFunctionImplementation) {
+            return ((BuiltInScalarFunctionImplementation) function).getAllChoices();
+        }
+        return ImmutableList.of(new ScalarImplementationChoice(
+                function.isNullable(),
+                function.getInvocationConvention().getArgumentConventions().stream().map(invocationArgumentConvention -> {
+                    switch (invocationArgumentConvention) {
+                        case NEVER_NULL:
+                            return valueTypeArgumentProperty(RETURN_NULL_ON_NULL);
+                        case BOXED_NULLABLE:
+                            return valueTypeArgumentProperty(USE_BOXED_TYPE);
+                        case NULL_FLAG:
+                            return valueTypeArgumentProperty(USE_NULL_FLAG);
+                        case BLOCK_POSITION:
+                            return valueTypeArgumentProperty(BLOCK_AND_POSITION);
+                        default:
+                            throw new UnsupportedOperationException(format("InvocationArgumentConvention %s cannot be a value type", invocationArgumentConvention));
+                    }
+                }).collect(toImmutableList()),
+                STACK,
+                function.getMethodHandle(),
+                Optional.empty()));
+    }
 }

presto-main/src/main/java/io/prestosql/sql/gen/FunctionCallCodeGenerator.java

@@ -17,6 +17,7 @@ import io.airlift.bytecode.BytecodeNode;
 import io.prestosql.metadata.FunctionAndTypeManager;
 import io.prestosql.spi.function.BuiltInScalarFunctionImplementation;
 import io.prestosql.spi.function.FunctionHandle;
+import io.prestosql.spi.function.ScalarFunctionImplementation;
 import io.prestosql.spi.relation.RowExpression;
 import io.prestosql.spi.type.Type;
 
@@ -25,6 +26,7 @@ import java.util.List;
 import java.util.Optional;
 
 import static io.prestosql.spi.function.BuiltInScalarFunctionImplementation.ArgumentType.VALUE_TYPE;
+import static io.prestosql.sql.gen.BytecodeUtils.getAllScalarFunctionImplementationChoices;
 
 public class FunctionCallCodeGenerator
         implements BytecodeGenerator
@@ -34,12 +36,14 @@ public class FunctionCallCodeGenerator
     {
         FunctionAndTypeManager functionAndTypeManager = context.getFunctionManager();
 
-        BuiltInScalarFunctionImplementation function = functionAndTypeManager.getBuiltInScalarFunctionImplementation(functionHandle);
+        ScalarFunctionImplementation function = functionAndTypeManager.getScalarFunctionImplementation(functionHandle);
 
         List<BytecodeNode> argumentsBytecode = new ArrayList<>();
+        BuiltInScalarFunctionImplementation.ScalarImplementationChoice choice = getAllScalarFunctionImplementationChoices(function).get(0);
+
         for (int i = 0; i < arguments.size(); i++) {
             RowExpression argument = arguments.get(i);
-            BuiltInScalarFunctionImplementation.ArgumentProperty argumentProperty = function.getArgumentProperty(i);
+            BuiltInScalarFunctionImplementation.ArgumentProperty argumentProperty = choice.getArgumentProperty(i);
             if (argumentProperty.getArgumentType() == VALUE_TYPE) {
                 argumentsBytecode.add(context.generate(argument));
             }

presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java

@@ -342,37 +342,188 @@ public class LocalExecutionPlanner
 {
     private static final Logger log = Logger.get(LocalExecutionPlanner.class);
 
-    private final Metadata metadata;
-    private final TypeAnalyzer typeAnalyzer;
-    private final Optional<ExplainAnalyzeContext> explainAnalyzeContext;
-    private final PageSourceProvider pageSourceProvider;
-    private final IndexManager indexManager;
-    private final NodePartitioningManager nodePartitioningManager;
-    private final PageSinkManager pageSinkManager;
-    private final ExchangeClientSupplier exchangeClientSupplier;
-    private final ExpressionCompiler expressionCompiler;
-    private final PageFunctionCompiler pageFunctionCompiler;
-    private final JoinFilterFunctionCompiler joinFilterFunctionCompiler;
-    private final DataSize maxIndexMemorySize;
-    private final IndexJoinLookupStats indexJoinLookupStats;
-    private final DataSize maxPartialAggregationMemorySize;
-    private final DataSize maxPagePartitioningBufferSize;
-    private final DataSize maxLocalExchangeBufferSize;
-    private final SpillerFactory spillerFactory;
-    private final SingleStreamSpillerFactory singleStreamSpillerFactory;
-    private final PartitioningSpillerFactory partitioningSpillerFactory;
-    private final PagesIndex.Factory pagesIndexFactory;
-    private final JoinCompiler joinCompiler;
-    private final LookupJoinOperators lookupJoinOperators;
-    private final OrderingCompiler orderingCompiler;
-    private final StateStoreProvider stateStoreProvider;
-    private final NodeInfo nodeInfo;
-    private final CubeManager cubeManager;
-    private final StateStoreListenerManager stateStoreListenerManager;
-    private final DynamicFilterCacheManager dynamicFilterCacheManager;
-    private final HeuristicIndexerManager heuristicIndexerManager;
-    private final FunctionResolution functionResolution;
-    private final LogicalRowExpressions logicalRowExpressions;
+    protected final Metadata metadata;
+    protected final TypeAnalyzer typeAnalyzer;
+    protected final Optional<ExplainAnalyzeContext> explainAnalyzeContext;
+    protected final PageSourceProvider pageSourceProvider;
+    protected final IndexManager indexManager;
+    protected final NodePartitioningManager nodePartitioningManager;
+    protected final PageSinkManager pageSinkManager;
+    protected final ExchangeClientSupplier exchangeClientSupplier;
+    protected final ExpressionCompiler expressionCompiler;
+    protected final PageFunctionCompiler pageFunctionCompiler;
+    protected final JoinFilterFunctionCompiler joinFilterFunctionCompiler;
+    protected final DataSize maxIndexMemorySize;
+    protected final IndexJoinLookupStats indexJoinLookupStats;
+    protected final DataSize maxPartialAggregationMemorySize;
+    protected final DataSize maxPagePartitioningBufferSize;
+    protected final DataSize maxLocalExchangeBufferSize;
+    protected final SpillerFactory spillerFactory;
+    protected final SingleStreamSpillerFactory singleStreamSpillerFactory;
+    protected final PartitioningSpillerFactory partitioningSpillerFactory;
+    protected final PagesIndex.Factory pagesIndexFactory;
+    protected final JoinCompiler joinCompiler;
+    protected final LookupJoinOperators lookupJoinOperators;
+    protected final OrderingCompiler orderingCompiler;
+    protected final StateStoreProvider stateStoreProvider;
+    protected final NodeInfo nodeInfo;
+    protected final CubeManager cubeManager;
+    protected final StateStoreListenerManager stateStoreListenerManager;
+    protected final DynamicFilterCacheManager dynamicFilterCacheManager;
+    protected final HeuristicIndexerManager heuristicIndexerManager;
+    protected final FunctionResolution functionResolution;
+    protected final LogicalRowExpressions logicalRowExpressions;
+    protected final TaskManagerConfig taskManagerConfig;
+
+    public Metadata getMetadata()
+    {
+        return metadata;
+    }
+
+    public TypeAnalyzer getTypeAnalyzer()
+    {
+        return typeAnalyzer;
+    }
+
+    public Optional<ExplainAnalyzeContext> getExplainAnalyzeContext()
+    {
+        return explainAnalyzeContext;
+    }
+
+    public PageSourceProvider getPageSourceProvider()
+    {
+        return pageSourceProvider;
+    }
+
+    public IndexManager getIndexManager()
+    {
+        return indexManager;
+    }
+
+    public NodePartitioningManager getNodePartitioningManager()
+    {
+        return nodePartitioningManager;
+    }
+
+    public PageSinkManager getPageSinkManager()
+    {
+        return pageSinkManager;
+    }
+
+    public ExchangeClientSupplier getExchangeClientSupplier()
+    {
+        return exchangeClientSupplier;
+    }
+
+    public ExpressionCompiler getExpressionCompiler()
+    {
+        return expressionCompiler;
+    }
+
+    public PageFunctionCompiler getPageFunctionCompiler()
+    {
+        return pageFunctionCompiler;
+    }
+
+    public JoinFilterFunctionCompiler getJoinFilterFunctionCompiler()
+    {
+        return joinFilterFunctionCompiler;
+    }
+
+    public DataSize getMaxIndexMemorySize()
+    {
+        return maxIndexMemorySize;
+    }
+
+    public IndexJoinLookupStats getIndexJoinLookupStats()
+    {
+        return indexJoinLookupStats;
+    }
+
+    public DataSize getMaxPartialAggregationMemorySize()
+    {
+        return maxPartialAggregationMemorySize;
+    }
+
+    public DataSize getMaxPagePartitioningBufferSize()
+    {
+        return maxPagePartitioningBufferSize;
+    }
+
+    public DataSize getMaxLocalExchangeBufferSize()
+    {
+        return maxLocalExchangeBufferSize;
+    }
+
+    public SpillerFactory getSpillerFactory()
+    {
+        return spillerFactory;
+    }
+
+    public SingleStreamSpillerFactory getSingleStreamSpillerFactory()
+    {
+        return singleStreamSpillerFactory;
+    }
+
+    public PartitioningSpillerFactory getPartitioningSpillerFactory()
+    {
+        return partitioningSpillerFactory;
+    }
+
+    public PagesIndex.Factory getPagesIndexFactory()
+    {
+        return pagesIndexFactory;
+    }
+
+    public JoinCompiler getJoinCompiler()
+    {
+        return joinCompiler;
+    }
+
+    public LookupJoinOperators getLookupJoinOperators()
+    {
+        return lookupJoinOperators;
+    }
+
+    public OrderingCompiler getOrderingCompiler()
+    {
+        return orderingCompiler;
+    }
+
+    public StateStoreProvider getStateStoreProvider()
+    {
+        return stateStoreProvider;
+    }
+
+    public NodeInfo getNodeInfo()
+    {
+        return nodeInfo;
+    }
+
+    public CubeManager getCubeManager()
+    {
+        return cubeManager;
+    }
+
+    public StateStoreListenerManager getStateStoreListenerManager()
+    {
+        return stateStoreListenerManager;
+    }
+
+    public DynamicFilterCacheManager getDynamicFilterCacheManager()
+    {
+        return dynamicFilterCacheManager;
+    }
+
+    public HeuristicIndexerManager getHeuristicIndexerManager()
+    {
+        return heuristicIndexerManager;
+    }
+
+    public TaskManagerConfig getTaskManagerConfig()
+    {
+        return taskManagerConfig;
+    }
 
     @Inject
     public LocalExecutionPlanner(
@@ -415,6 +566,7 @@ public class LocalExecutionPlanner
         this.pageFunctionCompiler = requireNonNull(pageFunctionCompiler, "pageFunctionCompiler is null");
         this.joinFilterFunctionCompiler = requireNonNull(joinFilterFunctionCompiler, "compiler is null");
         this.indexJoinLookupStats = requireNonNull(indexJoinLookupStats, "indexJoinLookupStats is null");
+        this.taskManagerConfig = taskManagerConfig;
         this.maxIndexMemorySize = requireNonNull(taskManagerConfig, "taskManagerConfig is null").getMaxIndexMemoryUsage();
         this.spillerFactory = requireNonNull(spillerFactory, "spillerFactory is null");
         this.singleStreamSpillerFactory = requireNonNull(singleStreamSpillerFactory, "singleStreamSpillerFactory is null");
@@ -633,7 +785,7 @@ public class LocalExecutionPlanner
         return first instanceof LookupOuterOperatorFactory && isTableScanPipeline(context.outerToJoinMap.get(driverFactory));
     }
 
-    private static void addLookupOuterDrivers(LocalExecutionPlanContext context)
+    protected static void addLookupOuterDrivers(LocalExecutionPlanContext context)
     {
         // For an outer join on the lookup side (RIGHT or FULL) add an additional
         // driver to output the unused rows in the lookup source
@@ -664,29 +816,29 @@ public class LocalExecutionPlanner
         }
     }
 
-    private static class LocalExecutionPlanContext
+    public static class LocalExecutionPlanContext
     {
-        private final TaskContext taskContext;
-        private final TypeProvider types;
-        private final List<DriverFactory> driverFactories;
-        private final Optional<IndexSourceContext> indexSourceContext;
+        protected final TaskContext taskContext;
+        protected final TypeProvider types;
+        protected List<DriverFactory> driverFactories;
+        protected final Optional<IndexSourceContext> indexSourceContext;
 
         // the collector is shared with all subContexts to allow local dynamic filtering
         // with multiple table scans (e.g. co-located joins).
-        private final LocalDynamicFiltersCollector dynamicFiltersCollector;
+        protected final LocalDynamicFiltersCollector dynamicFiltersCollector;
 
         // this is shared with all subContexts
-        private final AtomicInteger nextPipelineId;
+        protected final AtomicInteger nextPipelineId;
 
         private int nextOperatorId;
         private boolean inputDriver = true;
         private OptionalInt driverInstanceCount = OptionalInt.empty();
         private Map<PlanNodeId, OperatorFactory> cteOperationMap = new HashMap<>();
-        private Map<String, CommonTableExecutionContext> cteCtx;
+        protected Map<String, CommonTableExecutionContext> cteCtx;
         private static Map<String, PhysicalOperation> sourceInitialized = new ConcurrentHashMap<>();
         private final PlanNodeId consumerId;
-        private final Optional<PlanFragmentId> feederCTEId;
-        private final Optional<PlanNodeId> feederCTEParentId;
+        protected final Optional<PlanFragmentId> feederCTEId;
+        protected final Optional<PlanNodeId> feederCTEParentId;
 
         // Snapshot: record pipeline that corresponds to the lookup-outer pipeline.
         // This is used to help determine if a lookup-outer pipeline should be treated as a tabel-scan pipeine.
@@ -700,7 +852,7 @@ public class LocalExecutionPlanner
             this(taskContext, types, new ArrayList<>(), Optional.empty(), new LocalDynamicFiltersCollector(taskContext, Optional.of(metadata), dynamicFilterCacheManager), new AtomicInteger(0), feederCTEId, feederCTEParentId, cteCtx);
         }
 
-        private LocalExecutionPlanContext(
+        protected LocalExecutionPlanContext(
                 TaskContext taskContext,
                 TypeProvider types,
                 List<DriverFactory> driverFactories,
@@ -745,11 +897,16 @@ public class LocalExecutionPlanner
             return driverFactory;
         }
 
-        private List<DriverFactory> getDriverFactories()
+        public List<DriverFactory> getDriverFactories()
         {
             return ImmutableList.copyOf(driverFactories);
         }
 
+        public void setDriverFactories(List<DriverFactory> driverFactories)
+        {
+            this.driverFactories = driverFactories;
+        }
+
         public Session getSession()
         {
             return taskContext.getSession();
@@ -780,22 +937,27 @@ public class LocalExecutionPlanner
             return indexSourceContext;
         }
 
-        private int getNextPipelineId()
+        private AtomicInteger getPipelineId()
+        {
+            return nextPipelineId;
+        }
+
+        public int getNextPipelineId()
         {
             return nextPipelineId.getAndIncrement();
         }
 
-        private int getNextOperatorId()
+        public int getNextOperatorId()
         {
             return nextOperatorId++;
         }
 
-        private boolean isInputDriver()
+        public boolean isInputDriver()
         {
             return inputDriver;
         }
 
-        private void setInputDriver(boolean inputDriver)
+        public void setInputDriver(boolean inputDriver)
         {
             this.inputDriver = inputDriver;
         }
@@ -866,7 +1028,7 @@ public class LocalExecutionPlanner
         }
     }
 
-    private static class IndexSourceContext
+    public static class IndexSourceContext
     {
         private final SetMultimap<Symbol, Integer> indexLookupToProbeInput;
 
@@ -918,13 +1080,13 @@ public class LocalExecutionPlanner
         }
     }
 
-    private class Visitor
+    public class Visitor
             extends InternalPlanVisitor<PhysicalOperation, LocalExecutionPlanContext>
     {
-        private final Session session;
-        private final StageExecutionDescriptor stageExecutionDescriptor;
+        protected final Session session;
+        protected final StageExecutionDescriptor stageExecutionDescriptor;
 
-        private Visitor(Session session, StageExecutionDescriptor stageExecutionDescriptor)
+        public Visitor(Session session, StageExecutionDescriptor stageExecutionDescriptor)
         {
             this.session = session;
             this.stageExecutionDescriptor = stageExecutionDescriptor;
@@ -1618,7 +1780,7 @@ public class LocalExecutionPlanner
             }
         }
 
-        private Supplier<List<Map<ColumnHandle, DynamicFilter>>> getDynamicFilterSupplier(Optional<List<List<DynamicFilters.Descriptor>>> dynamicFilters, PlanNode sourceNode, LocalExecutionPlanContext context)
+        protected Supplier<List<Map<ColumnHandle, DynamicFilter>>> getDynamicFilterSupplier(Optional<List<List<DynamicFilters.Descriptor>>> dynamicFilters, PlanNode sourceNode, LocalExecutionPlanContext context)
         {
             if (dynamicFilters.isPresent() && !dynamicFilters.get().isEmpty()) {
                 log.debug("[TableScan] Dynamic filters: %s", dynamicFilters);
@@ -1643,7 +1805,7 @@ public class LocalExecutionPlanner
             return null;
         }
 
-        private RowExpression bindChannels(RowExpression inputExpression, Map<Symbol, Integer> sourceLayout, TypeProvider types)
+        public RowExpression bindChannels(RowExpression inputExpression, Map<Symbol, Integer> sourceLayout, TypeProvider types)
         {
             RowExpression expression = inputExpression;
             Type type = expression.getType();
@@ -1864,12 +2026,12 @@ public class LocalExecutionPlanner
                     stageExecutionDescriptor.isScanGroupedExecution(node.getId()) ? GROUPED_EXECUTION : UNGROUPED_EXECUTION);
         }
 
-        private ImmutableMap<Symbol, Integer> makeLayout(PlanNode node)
+        protected ImmutableMap<Symbol, Integer> makeLayout(PlanNode node)
         {
             return makeLayoutFromOutputSymbols(node.getOutputSymbols());
         }
 
-        private ImmutableMap<Symbol, Integer> makeLayoutFromOutputSymbols(List<Symbol> outputSymbols)
+        protected ImmutableMap<Symbol, Integer> makeLayoutFromOutputSymbols(List<Symbol> outputSymbols)
         {
             ImmutableMap.Builder<Symbol, Integer> outputMappings = ImmutableMap.builder();
             int channel = 0;
@@ -2231,7 +2393,7 @@ public class LocalExecutionPlanner
             return symbols.stream().map(SymbolUtils::toSymbolReference).collect(toImmutableSet());
         }
 
-        private PhysicalOperation createNestedLoopJoin(JoinNode node, LocalExecutionPlanContext context)
+        protected PhysicalOperation createNestedLoopJoin(JoinNode node, LocalExecutionPlanContext context)
         {
             PhysicalOperation probeSource = node.getLeft().accept(this, context);
 
@@ -2459,7 +2621,7 @@ public class LocalExecutionPlanner
             return new PhysicalOperation(operator, outputMappings.build(), context, probeSource);
         }
 
-        private Optional<LocalDynamicFilter> createDynamicFilter(JoinNode node, LocalExecutionPlanContext context, int partitionCount)
+        protected Optional<LocalDynamicFilter> createDynamicFilter(JoinNode node, LocalExecutionPlanContext context, int partitionCount)
         {
             if (!isEnableDynamicFiltering(context.getSession())) {
                 return Optional.empty();
@@ -2625,7 +2787,7 @@ public class LocalExecutionPlanner
             return lookupSourceFactoryManager;
         }
 
-        private JoinFilterFunctionFactory compileJoinFilterFunction(
+        protected JoinFilterFunctionFactory compileJoinFilterFunction(
                 RowExpression filterExpression,
                 Map<Symbol, Integer> probeLayout,
                 Map<Symbol, Integer> buildLayout,
@@ -2636,7 +2798,7 @@ public class LocalExecutionPlanner
             return joinFilterFunctionCompiler.compileJoinFilterFunction(bindChannels(filterExpression, joinSourcesLayout, types), buildLayout.size());
         }
 
-        private int sortExpressionAsSortChannel(
+        public int sortExpressionAsSortChannel(
                 RowExpression sortExpression,
                 Map<Symbol, Integer> probeLayout,
                 Map<Symbol, Integer> buildLayout,
@@ -2682,7 +2844,7 @@ public class LocalExecutionPlanner
             }
         }
 
-        private Map<Symbol, Integer> createJoinSourcesLayout(Map<Symbol, Integer> lookupSourceLayout, Map<Symbol, Integer> probeSourceLayout)
+        protected Map<Symbol, Integer> createJoinSourcesLayout(Map<Symbol, Integer> lookupSourceLayout, Map<Symbol, Integer> probeSourceLayout)
         {
             ImmutableMap.Builder<Symbol, Integer> joinSourcesLayout = ImmutableMap.builder();
             joinSourcesLayout.putAll(lookupSourceLayout);
@@ -3203,19 +3365,19 @@ public class LocalExecutionPlanner
             throw new UnsupportedOperationException("not yet implemented");
         }
 
-        private List<Type> getSourceOperatorTypes(PlanNode node, TypeProvider types)
+        protected List<Type> getSourceOperatorTypes(PlanNode node, TypeProvider types)
         {
             return getSymbolTypes(node.getOutputSymbols(), types);
         }
 
-        private List<Type> getSymbolTypes(List<Symbol> symbols, TypeProvider types)
+        protected List<Type> getSymbolTypes(List<Symbol> symbols, TypeProvider types)
         {
             return symbols.stream()
                     .map(types::get)
                     .collect(toImmutableList());
         }
 
-        private AccumulatorFactory buildAccumulatorFactory(
+        protected AccumulatorFactory buildAccumulatorFactory(
                 PhysicalOperation source,
                 Aggregation aggregation)
         {
@@ -3558,7 +3720,7 @@ public class LocalExecutionPlanner
         };
     }
 
-    private static Function<Page, Page> enforceLayoutProcessor(List<Symbol> expectedLayout, Map<Symbol, Integer> inputLayout)
+    protected static Function<Page, Page> enforceLayoutProcessor(List<Symbol> expectedLayout, Map<Symbol, Integer> inputLayout)
     {
         int[] channels = expectedLayout.stream()
                 .peek(symbol -> checkArgument(inputLayout.containsKey(symbol), "channel not found for symbol: %s", symbol))
@@ -3573,7 +3735,7 @@ public class LocalExecutionPlanner
         return new PageChannelSelector(channels);
     }
 
-    private static List<Integer> getChannelsForSymbols(List<Symbol> symbols, Map<Symbol, Integer> layout)
+    protected static List<Integer> getChannelsForSymbols(List<Symbol> symbols, Map<Symbol, Integer> layout)
     {
         ImmutableList.Builder<Integer> builder = ImmutableList.builder();
         for (Symbol symbol : symbols) {
@@ -3582,7 +3744,7 @@ public class LocalExecutionPlanner
         return builder.build();
     }
 
-    private static Function<Symbol, Integer> channelGetter(PhysicalOperation source)
+    protected static Function<Symbol, Integer> channelGetter(PhysicalOperation source)
     {
         return input -> {
             checkArgument(source.getLayout().containsKey(input));
@@ -3593,7 +3755,7 @@ public class LocalExecutionPlanner
     /**
      * Encapsulates an physical operator plus the mapping of logical symbols to channel/field
      */
-    private static class PhysicalOperation
+    public static class PhysicalOperation
     {
         private final List<OperatorFactory> operatorFactories;
         private final Map<Symbol, Integer> layout;
@@ -3664,7 +3826,7 @@ public class LocalExecutionPlanner
             return layout;
         }
 
-        private List<OperatorFactory> getOperatorFactories()
+        public List<OperatorFactory> getOperatorFactories()
         {
             return operatorFactories;
         }
@@ -3675,7 +3837,7 @@ public class LocalExecutionPlanner
         }
     }
 
-    private static class DriverFactoryParameters
+    protected static class DriverFactoryParameters
     {
         private final LocalExecutionPlanContext subContext;
         private final PhysicalOperation source;

presto-main/src/main/java/io/prestosql/sql/planner/sanity/ExternalFunctionPushDownChecker.java

@@ -17,9 +17,11 @@ package io.prestosql.sql.planner.sanity;
 import io.prestosql.Session;
 import io.prestosql.execution.warnings.WarningCollector;
 import io.prestosql.expressions.DefaultRowExpressionTraversalVisitor;
+import io.prestosql.metadata.FunctionAndTypeManager;
 import io.prestosql.metadata.Metadata;
 import io.prestosql.spi.ErrorCodeSupplier;
 import io.prestosql.spi.PrestoException;
+import io.prestosql.spi.connector.CatalogSchemaName;
 import io.prestosql.spi.plan.AggregationNode;
 import io.prestosql.spi.plan.FilterNode;
 import io.prestosql.spi.plan.JoinNode;
@@ -50,7 +52,6 @@ import java.util.Set;
 import java.util.stream.Collectors;
 
 import static io.prestosql.spi.StandardErrorCode.GENERIC_USER_ERROR;
-import static io.prestosql.spi.connector.CatalogSchemaName.DEFAULT_NAMESPACE;
 import static io.prestosql.sql.relational.OriginalExpressionUtils.isOriginalExpression;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
@@ -66,7 +67,7 @@ public class ExternalFunctionPushDownChecker
     public void validate(PlanNode planNode, Session session, Metadata metadata, TypeAnalyzer typeAnalyzer, TypeProvider types, WarningCollector warningCollector)
     {
         Set<String> set = new HashSet<>();
-        planNode.accept(new ExternalFunctionFinder(), set);
+        planNode.accept(new ExternalFunctionFinder(metadata.getFunctionAndTypeManager()), set);
         if (set.size() > 0) {
             String allErrorFun = set.stream().map(String::toString).collect(Collectors.joining(", "));
             throw new IllegalExternalFunctionUsageException(
@@ -78,6 +79,18 @@ public class ExternalFunctionPushDownChecker
     private static class ExternalFunctionFinder
             extends InternalPlanVisitor<Void, Set<String>>
     {
+        public static List<CatalogSchemaName> functionNamespaces = new ArrayList<>();
+
+        public ExternalFunctionFinder(FunctionAndTypeManager functionAndTypeManager)
+        {
+            for (String catalog : functionAndTypeManager.getFunctionNamespaceManagers().keySet()) {
+                CatalogSchemaName namespace = new CatalogSchemaName(catalog, "default");
+                if (!functionNamespaces.contains(namespace)) {
+                    functionNamespaces.add(namespace);
+                }
+            }
+        }
+
         @Override
         public Void visitPlan(PlanNode node, Set<String> context)
         {
@@ -282,7 +295,7 @@ public class ExternalFunctionPushDownChecker
 
         private static boolean isDefaultFunction(CallExpression callExpression)
         {
-            return DEFAULT_NAMESPACE.equals(callExpression.getFunctionHandle().getFunctionNamespace());
+            return ExternalFunctionFinder.functionNamespaces.contains(callExpression.getFunctionHandle().getFunctionNamespace());
         }
     }
 

presto-main/src/main/java/io/prestosql/utils/HetuConfig.java

@@ -62,10 +62,53 @@ public class HetuConfig
     private Duration splitCacheStateUpdateInterval = new Duration(2, TimeUnit.SECONDS);
     private boolean isTraceStackVisible;
 
+    private String extensionExecutionPlannerJarPath;
+    private String extensionExecutionPlannerClassPath;
+    private boolean extensionExecutionPlannerEnabled;
+
     public HetuConfig()
     {
     }
 
+    public boolean getExtensionExecutionPlannerEnabled()
+    {
+        return extensionExecutionPlannerEnabled;
+    }
+
+    @Config(HetuConstant.EXTENSION_EXECUTION_PLANNER_ENABLED)
+    @ConfigDescription("extension execution planner enable from config")
+    public HetuConfig setExtensionExecutionPlannerEnabled(boolean extensionExecutionPlannerEnabled)
+    {
+        this.extensionExecutionPlannerEnabled = extensionExecutionPlannerEnabled;
+        return this;
+    }
+
+    public String getExtensionExecutionPlannerJarPath()
+    {
+        return extensionExecutionPlannerJarPath;
+    }
+
+    @Config(HetuConstant.EXTENSION_EXECUTION_PLANNER_JAR_PATH)
+    @ConfigDescription("extension execution planner jar path from config")
+    public HetuConfig setExtensionExecutionPlannerJarPath(String extensionExecutionPlannerJarPath)
+    {
+        this.extensionExecutionPlannerJarPath = extensionExecutionPlannerJarPath;
+        return this;
+    }
+
+    public String getExtensionExecutionPlannerClassPath()
+    {
+        return extensionExecutionPlannerClassPath;
+    }
+
+    @Config(HetuConstant.EXTENSION_EXECUTION_PLANNER_CLASS_PATH)
+    @ConfigDescription("extension execution planner class path from config")
+    public HetuConfig setExtensionExecutionPlannerClassPath(String extensionExecutionPlannerClassPath)
+    {
+        this.extensionExecutionPlannerClassPath = extensionExecutionPlannerClassPath;
+        return this;
+    }
+
     @NotNull
     public boolean isFilterEnabled()
     {

presto-main/src/test/java/io/prestosql/operator/window/AbstractTestWindowFunction.java

@@ -30,7 +30,7 @@ public abstract class AbstractTestWindowFunction
     protected LocalQueryRunner queryRunner;
 
     @BeforeClass
-    public final void initTestWindowFunction()
+    public void initTestWindowFunction()
     {
         queryRunner = new LocalQueryRunner(TEST_SESSION);
     }

presto-main/src/test/java/io/prestosql/utils/TestHetuConfig.java

@@ -54,7 +54,10 @@ public class TestHetuConfig
                 .setSplitCacheMapEnabled(false)
                 .setSplitCacheStateUpdateInterval(new Duration(2, TimeUnit.SECONDS))
                 .setTraceStackVisible(false)
-                .setIndexToPreload(""));
+                .setIndexToPreload("")
+                .setExtensionExecutionPlannerEnabled(false)
+                .setExtensionExecutionPlannerJarPath(null)
+                .setExtensionExecutionPlannerClassPath(null));
     }
 
     @Test
@@ -85,6 +88,9 @@ public class TestHetuConfig
                 .put("hetu.split-cache-map.state-update-interval", "5s")
                 .put("stack-trace-visible", "true")
                 .put("hetu.heuristicindex.filter.cache.preload-indices", "idx1,idx2")
+                .put("extension_execution_planner_enabled", "true")
+                .put("extension_execution_planner_jar_path", "")
+                .put("extension_execution_planner_class_path", "")
                 .build();
 
         HetuConfig expected = new HetuConfig()
@@ -111,7 +117,10 @@ public class TestHetuConfig
                 .setSplitCacheMapEnabled(true)
                 .setSplitCacheStateUpdateInterval(new Duration(5, TimeUnit.SECONDS))
                 .setTraceStackVisible(true)
-                .setIndexToPreload("idx1,idx2");
+                .setIndexToPreload("idx1,idx2")
+                .setExtensionExecutionPlannerEnabled(true)
+                .setExtensionExecutionPlannerJarPath("")
+                .setExtensionExecutionPlannerClassPath("");
 
         ConfigAssertions.assertFullMapping(properties, expected);
     }

presto-matching/pom.xml

@@ -18,7 +18,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-matching</artifactId>

presto-memory-context/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-memory-context</artifactId>

presto-memory/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-memory</artifactId>

presto-ml/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-ml</artifactId>

presto-mysql/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-mysql</artifactId>

presto-orc/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-orc</artifactId>

presto-parquet/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-parquet</artifactId>

presto-parser/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-parser</artifactId>

presto-password-authenticators/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-password-authenticators</artifactId>

presto-plugin-toolkit/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-plugin-toolkit</artifactId>

presto-postgresql/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-postgresql</artifactId>

presto-product-tests/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>presto-root</artifactId>
         <groupId>io.hetu.core</groupId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-product-tests</artifactId>

presto-proxy/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-proxy</artifactId>

presto-rcfile/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-rcfile</artifactId>

presto-record-decoder/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.hetu.core</groupId>
         <artifactId>presto-root</artifactId>
-        <version>1.6.0-SNAPSHOT</version>
+        <version>1.6.2-SNAPSHOT</version>
     </parent>
 
     <artifactId>presto-record-decoder</artifactId>