diff --git a/hetu-docs/en/connector/mongodb.md b/hetu-docs/en/connector/mongodb.md
new file mode 100644
index 000000000..27a997fdd
--- /dev/null
+++ b/hetu-docs/en/connector/mongodb.md
@@ -0,0 +1,248 @@
+# MongoDB Connector
+
+The MongoDB connector allows MongoDB collections to be used as tables in the openLooKeng.
+
+**Note:**
+
+ MongoDB 2.6 and later versions are supported, you are advised to use version 3.0 or later.
+
+## Configuration
+
+To configure the MongoDB connector, create a catalog property file `etc/catalog/mongodb.properties` by referring to the following content and replace the properties as required:
+
+```properties
+connector.name=mongodb
+mongodb.seeds=host1,host:port
+```
+
+### Multiple MongoDB Clusters
+
+Multiple catalogs can be created as required. Therefore, if there is an additional MongoDB cluster, you only need to add another property file with a different name to `etc/catalog` (ensure that it ends with `.properties`). For example, if you name the property file as `sales.properties`, the openLooKeng will create a catalog named `sales` using the configured connector.
+
+## Configuring Properties
+
+The following properties are available:
+
+| Property Name| Description|
+|----------|----------|
+| `mongodb.seeds`| List of all mongod servers|
+| `mongodb.schema-collection`| A collection of schema information|
+| `mongodb.case-insensitive-name-matching`| Case-insensitive matching between database and collection names|
+| `mongodb.credentials`| List of credentials|
+| `mongodb.min-connections-per-host`| Minimum number of the connection pools per host|
+| `mongodb.connections-per-host`| Maximum number of the connection pools per host|
+| `mongodb.max-wait-time`| Maximum waiting time|
+| `mongodb.max-connection-idle-time`| Maximum idle time of connection pooling|
+| `mongodb.connection-timeout`| Communication connection timeout|
+| `mongodb.socket-timeout`| Communication timeout|
+| `mongodb.socket-keep-alive`| Whether to enable the keep-alive function for each communication channel|
+| `mongodb.ssl.enabled`| Using TLS/SSL to connect to mongod/mongos|
+| `mongodb.read-preference`| Read preference|
+| `mongodb.write-concern`| Write policy|
+| `mongodb.required-replica-set`| Name of the required replica set|
+| `mongodb.cursor-batch-size`| Number of elements returned in a batch|
+
+### `mongodb.seeds`
+
+List of all mongod servers in the same replica set, which are separated using commas (,). The format is in `hostname[:port]`. Or list of mongos servers in the same sharded cluster. If port is not specified, port 27017 is used.
+
+This property is mandatory. There is no default value, and at least one seed must be defined.
+
+### `mongodb.schema-collection`
+
+MongoDB is a document-oriented database, and there is no fixed schema information in the system. Therefore, a special collection in each MongoDB database should define the structure for all tables. For more information, see the [Table Definition](./mongodb.md#table-definition) section.
+
+At startup, this connector attempts to guess the type of the field, but the type may not match the collection you created. In this case, you need to manually modify it. `CREATE TABLE` and `CREATE TABLE AS SELECT` are used to create an entry for you.
+
+This property is optional. The default value is `_schema`.
+
+### `mongodb.case-insensitive-name-matching`
+
+Case-insensitive matching between database and collection names.
+
+This property is optional. The default value is `false`.
+
+### `mongodb.credentials`
+
+List of `username:password@collection` credentials separated by commas (,).
+
+This property is optional. There is no default value.
+
+### `mongodb.min-connections-per-host`
+
+Minimum number of connections per host in the MongoClient instance. These connections are retained in the connection pool when they are idle. Over time, the connection pool contains at least this minimum number of connections.
+
+This property is optional. The default value is `0`.
+
+### `mongodb.connections-per-host`
+
+Maximum number of connections per host in the MongoClient instance. These connections are retained in the connection pool when they are idle. Once the connection pool resources are exhausted, any operation that requires a connection is blocked and waits for an available connection.
+
+This property is optional. The default value is `100`.
+
+### `mongodb.max-wait-time`
+
+Maximum time (in milliseconds) that a thread can wait for a connection to become available. The value `0` indicates that the thread will not wait. A negative value indicates that the thread will wait will indefinitely for a connection to become available.
+
+This property is optional. The default value is `120000`.
+
+### `mongodb.connections-timeout`
+
+Connection timeout interval (in milliseconds). The value `0` indicates that no timeout occurs. This property is used only when a new connection is set up.
+
+This property is optional. The default value is `10000`.
+
+### `mongodb.socket-timeout`
+
+Socket timeout interval (in milliseconds). It is used for I/O socket read and write operations.
+
+This property is optional. The default value is `0`, indicating that no timeout occurs.
+
+### `mongodb.socket-keep-alive`
+
+This property controls the socket keep-alive function, which keeps the connection alive through the firewall.
+
+This property is optional. The default value is `false`.
+
+### `mongodb.ssl.enabled`
+
+This property is used to enable the SSL connection with the MongoDB server.
+
+This property is optional. The default value is `false`.
+
+### `mongodb.read-preference`
+
+The read preference are used for query, mapping restoration, aggregation, and counting. The value can be `PRIMARY`, `PRIMARY_PREFERRED`, `SECONDARY`, `SECONDARY_PREFERRED`, or `NEAREST`.
+
+This property is optional. The default value is `PRIMARY`.
+
+### `mongodb.write-concern`
+
+Write policy. The value can be `ACKNOWLEDGED`, `FSYNC_SAFE`, `FSYNCED`, `JOURNAL_SAFEY`, `JOURNALED`, `MAJORITY`, `NORMAL`, `REPLICA_ACKNOWLEDGED`, `REPLICAS_SAFE`, or `UNACKNOWLEDGED`.
+
+This property is optional. The default value is `ACKNOWLEDGED`.
+
+### `mongodb.required-replica-set`
+
+Name of the required replica set. After this property is set, the MongoClient instance performs the following operations:
+
+> - Connect in replica set mode and discover all members in the collection based on the specified server.
+> - Ensure that the collection name reported by all members matches the required collection name.
+> - If any member of the seed list is not part of a replica set with the required name, any requests are rejected.
+
+This property is optional. There is no default value.
+
+### `mongodb.cursor-batch-size`
+
+Limits the number of elements returned in a batch. A cursor typically fetches a batch of result objects and stores them locally. If **batchSize** is set to **0**, the default value of the driver is used. If the value of **batchSize** is positive, the value indicates the size of each batch of objects that are retrieved. The value can be adjusted to optimize performance and limit data transfer. If the value of **batchSize** is negative, it will limit the number of returned objects to the maximum batch size (typically 4 MB), and the cursor will be closed. For example, if **batchSize** is **-10**, the server will return up to 10 documents, return as many documents as possible in 4 MB, and then close the cursor.
+
+**Note**
+
+ Do not set the batch size to 1.
+
+This property is optional. The default value is `0`.
+
+## Table Definition
+
+MongoDB maintains the table definition on the configuration special collection specified by `mongodb.schema-collection`.
+
+**Note**
+
+ The plugin cannot detect a collection deletion.
+ You need to run db.getCollection("_schema").remove({table: delete_table_name}) in the Mongo shell to delete the collection.
+ Or you can delete the collection by running DROP TABLE table_name using openLooKeng.
+
+A collection in a schema consists of MongoDB documents of a table.
+
+ {
+ "table": ...,
+ "fields": [
+ { "name" : ...,
+ "type" : "varchar|bigint|boolean|double|date|array(bigint)|...",
+ "hidden" : false },
+ ...
+ ]
+ }
+ }
+
+| Field| Mandatory or Optional| Type| Description|
+|:----------|:----------|:----------|:----------|
+| `table`| Mandatory| string| Name of the openLooKeng table|
+| `fields`| Mandatory| array| Field definition list. For each column definition, a new column is created in the openLooKeng table.|
+
+The definition of each field is as follows:
+
+ {
+ "name": ...,
+ "type": ...,
+ "hidden": ...
+ }
+
+| Field| Mandatory or Optional| Type| Description|
+|:----------|:----------|:----------|:----------|
+| `name`| Mandatory| string| Name of a column in the openLooKeng table|
+| `type`| Mandatory| string| Type of a column|
+| `hidden`| Optional| boolean| Hides the column from the `DESCRIBE
` and `SELECT *` results. The default value is `false`.|
+
+There is no restriction on the field description of the key or message.
+
+## ObjectId
+
+The MongoDB collection has a special field `_id`. The connector attempts to follow the same rules for this special field, so there will be a hidden field `_id`.
+
+```sql
+ CREATE TABLE IF NOT EXISTS orders (
+ orderkey bigint,
+ orderstatus varchar,
+ totalprice double,
+ orderdate date
+ );
+
+ INSERT INTO orders VALUES(1, 'bad', 50.0, current_date);
+ INSERT INTO orders VALUES(2, 'good', 100.0, current_date);
+ SELECT _id, * FROM orders;
+```
+
+```sql
+ _id | orderkey | orderstatus | totalprice | orderdate
+ -------------------------------------+----------+-------------+------------+------------
+ 55 b1 51 63 38 64 d6 43 8c 61 a9 ce | 1 | bad | 50.0 | 2015-07-23
+ 55 b1 51 67 38 64 d6 43 8c 61 a9 cf | 2 | good | 100.0 | 2015-07-23
+ (2 rows)
+```
+
+```sql
+ SELECT _id, * FROM orders WHERE _id = ObjectId('55b151633864d6438c61a9ce');
+```
+
+```sql
+ _id | orderkey | orderstatus | totalprice | orderdate
+ -------------------------------------+----------+-------------+------------+------------
+ 55 b1 51 63 38 64 d6 43 8c 61 a9 ce | 1 | bad | 50.0 | 2015-07-23
+ (1 row)
+```
+
+The `_id` field can be rendered as a readable value and converted to `VARCHAR`:
+
+```sql
+ SELECT CAST(_id AS VARCHAR), * FROM orders WHERE _id = ObjectId('55b151633864d6438c61a9ce');
+```
+
+```sql
+
+ _id | orderkey | orderstatus | totalprice | orderdate
+ ---------------------------+----------+-------------+------------+------------
+ 55b151633864d6438c61a9ce | 1 | bad | 50.0 | 2015-07-23
+ (1 row)
+```
+
+## Restrictions
+
+\- [Row deletion](../sql/delete.md) is not supported.
+
+\- View creation is not supported.
+
+\- The empty tables and fields created by the openLooKeng cannot be queried in the MongoDB.
+
+\- After a table or field is deleted from the MongoDB, the table or field still exists in the openLooKeng while the values are NULL.
+
diff --git a/hetu-docs/zh/connector/mongodb.md b/hetu-docs/zh/connector/mongodb.md
new file mode 100644
index 000000000..33fcbefe4
--- /dev/null
+++ b/hetu-docs/zh/connector/mongodb.md
@@ -0,0 +1,249 @@
+
+# MongoDB连接器
+
+MongoDB连接器允许将MongoDB集合作为openLooKeng中的表使用。
+
+**注意**
+
+ 支持MongoDB 2.6+,但建议使用3.0或更高版本。
+
+## 配置
+
+要配置MongoDB连接器,参考以下内容创建目录属性文件`etc/catalog/mongodb.properties`,并根据需要替换这些属性:
+
+``` properties
+connector.name=mongodb
+mongodb.seeds=host1,host:port
+```
+
+### 多个MongoDB集群
+
+可以根据需要创建多个目录,因此,如果有额外的MongoDB集群,只需添加另一个不同的名称的属性文件到`etc/catalog`中(确保它以`.properties`结尾)。例如,如果将属性文件命名为`sales.properties`,openLooKeng将使用配置的连接器创建一个名为`sales`的目录。
+
+## 配置属性
+
+支持的配置属性:
+
+| 属性名称| 说明|
+|----------|----------|
+| `mongodb.seeds`| 所有mongod服务器的列表|
+| `mongodb.schema-collection`| 一个包含schema信息的集合|
+| `mongodb.case-insensitive-name-matching`| 不区分大小写匹配数据库和集合名称|
+| `mongodb.credentials`| 认证列表|
+| `mongodb.min-connections-per-host`| 每个主机的最小连接池大小|
+| `mongodb.connections-per-host`| 每个主机的连接池的最大大小|
+| `mongodb.max-wait-time`| 最大等待时间|
+| `mongodb.max-connection-idle-time`| 池化连接的最大空闲时间|
+| `mongodb.connection-timeout`| 通信连接超时|
+| `mongodb.socket-timeout`| 通信超时|
+| `mongodb.socket-keep-alive`| 是否在每个通信上使能keep-alive|
+| `mongodb.ssl.enabled`| 使用TLS/SSL连接到mongod/mongos|
+| `mongodb.read-preference`| 读偏好|
+| `mongodb.write-concern`| 写入策略|
+| `mongodb.required-replica-set`| 所需的副本集名称|
+| `mongodb.cursor-batch-size`| 批量返回的元素数|
+
+### `mongodb.seeds`
+
+同一副本集中所有mongod服务器列表,以逗号分隔 ,格式如``hostname[:port]``,或者同一个分片集群的mongos服务器列表。如果不指定port,则使用27017端口。
+
+此属性是必需的;没有默认值,并且至少必须定义一个seed。
+
+### `mongodb.schema-collection`
+
+由于MongoDB是文档数据库,因此系统中没有固定的schema信息。因此,每个MongoDB数据库中的一个特殊集合应定义所有表的架构。有关详细信息,请参阅[表格定义](./mongodb.md#表格定义)部分。
+
+在启动时,此连接器尝试猜测字段的类型,但可能与用户创建的集合可不匹配。在这种情况下,需要手动修改它。`CREATE TABLE`和`CREATE TABLE AS SELECT`会为用户创建一个条目。
+
+该属性是可选的;默认值为`_schema`。
+
+### `mongodb.case-insensitive-name-matching`
+
+不区分大小写匹配数据库和集合名称。
+
+该属性是可选的;默认值为`false`。
+
+### `mongodb.credentials`
+
+以逗号分隔的`username:password@collection`认证列表。
+
+该属性是可选的;没有默认值。
+
+### `mongodb.min-connectenions-per-host`
+
+此MongoClient实例每个主机的最小连接数。这些连接在空闲时将保留在连接池中。当这些连接空闲时,它们将保存在连接池中,并且随着时间的推移,连接池将确保至少包含这个最小数目的连接。
+
+该属性是可选的;默认值为`0`。
+
+### `mongodb.connections-per-host`
+
+此MongoClient实例每个主机允许的最大连接数。这些连接在空闲时将保留在连接池中。一旦连接池资源耗尽,任何需要连接的操作都将阻塞等待可用连接。
+
+该属性是可选的;默认值为`100`。
+
+### `mongodb.max-wait-time`
+
+线程可以等待连接变为可用状态的最大等待时间(以毫秒为单位)。值`0`表示它不会等待。负值表示无限期地等待连接变为可用。
+
+该属性是可选的;默认值为`120000`。
+
+### `mongodb.connections-timeout`
+
+连接超时(以毫秒为单位)。值`0`表示没有超时。仅在建立新连接时使用。
+
+该属性是可选的;默认值为`10000`。
+
+### `mongodb.socket-timeout`
+
+套接字超时(以毫秒为单位)。它用于I / O套接字读取和写入操作。
+
+该属性是可选的;默认值为`0`并且表示没有超时。
+
+### `mongodb.socket-keep-alive`
+
+此标志控制套接字保持活动功能,该功能通过防火墙保持连接活动。
+
+该属性是可选的;默认值为`false`。
+
+### `mongodb.ssl.enabled`
+
+此标志用于启用与MongoDB服务器的SSL连接。
+
+该属性是可选的;默认值为`false`。
+
+### `mongodb.read-preference`
+
+读偏好用于查询、映射还原、聚合和计数。可配置的值有`PRIMARY`,`PRIMARY_PREFERRED`,`SECONDARY`,`SECONDARY_PREFERRED`和`NEAREST`。
+
+该属性是可选的;默认值为`PRIMARY`。
+
+### `mongodb.write-concern`
+
+写入策略。可配置的值有`ACKNOWLEDGED`,`FSYNC_SAFE`,`FSYNCED`,`JOURNAL_SAFEY`,`JOURNALED`,`MAJORITY`,`NORMAL`,`REPLICA_ACKNOWLEDGED`,`REPLICAS_SAFE`和`UNACKNOWLEDGED`。
+
+该属性是可选的;默认值为`ACKNOWLEDGED`。
+
+### `mongodb.required-replica-set`
+
+所需的副本集名称。设置此选项后,MongoClient实例将执行以下操作:
+
+>- 以副本集模式连接,并根据给定的服务器发现集合中的所有成员
+>- 确保所有成员报告的集合名称与所需的集合名称匹配。
+>- 如果seed列表的任何成员不是具有必需名称的副本集的一部分,则拒绝处理任何请求。
+
+该属性是可选的。没有默认值。
+
+### `mongodb.cursor-batch-size`
+
+限制一批中返回的元素数。游标通常会获取一批结果对象并将其存储在本地。
+如果batchSize为0,将使用驱动程序的默认值。
+如果batchSize为正,则其值表示检索到的每批对象的大小。可以对其进行调整以优化性能并限制数据传输。
+如果batchSize为负,它将限制返回的对象数量,这些对象的数量在最大批处理大小限制内(通常为4MB),并且游标将被关闭。例如,如果batchSize为-10,则服务器将最多返回10个文档,并尽可能多地返回4MB中可以容纳的文档,然后关闭游标。
+
+**注意**
+
+ 请勿将批量大小设置为1。
+
+该属性是可选的;默认值为`0`。
+
+## 表格定义
+
+MongoDB在`mongodb.schema-collection`指定的配置特殊集合上维护表格定义。
+
+**注意**
+
+ 插件无法检测到集合被删除。
+ 需要在Mongo shell中通过db.getCollection("_schema").remove({ table: delete_table_name })删除集合。
+ 或者通过使用openLooKeng运行DROP TABLE table_name来删除集合。
+
+schema中的集合由表的MongoDB文档组成。
+
+ {
+ "table": ...,
+ "fields": [
+ { "name" : ...,
+ "type" : "varchar|bigint|boolean|double|date|array(bigint)|...",
+ "hidden" : false },
+ ...
+ ]
+ }
+ }
+
+| 字段| 是否必填| 类型| 说明|
+|:----------|:----------|:----------|:----------|
+| `table`| 必填| string| openLooKeng表名称。|
+| `fields`| 必填| array| 字段定义列表。每个字段定义在openLooKeng表中创建一个新列。|
+
+每个字段定义:
+
+ {
+ "name": ...,
+ "type": ...,
+ "hidden": ...
+ }
+
+| 字段| 是否必填| 类型| 说明|
+|:----------|:----------|:----------|:----------|
+| `name`| 必填| string| openLooKeng表中的列名。|
+| `type`| 必填| string| 列的类型。|
+| `hidden`| 可选| boolean| 从`DESCRIBE
`和`SELECT *`中隐藏该列。默认为`false`。|
+
+密钥或消息的字段描述没有限制。
+
+## ObjectId
+
+MongoDB集合具有特殊字段`_id`。连接器尝试对此特殊字段遵循相同的规则,因此将存在隐藏字段`_id`。
+```sql
+ CREATE TABLE IF NOT EXISTS orders (
+ orderkey bigint,
+ orderstatus varchar,
+ totalprice double,
+ orderdate date
+ );
+
+ INSERT INTO orders VALUES(1, 'bad', 50.0, current_date);
+ INSERT INTO orders VALUES(2, 'good', 100.0, current_date);
+ SELECT _id, * FROM orders;
+```
+```sql
+ _id | orderkey | orderstatus | totalprice | orderdate
+ -------------------------------------+----------+-------------+------------+------------
+ 55 b1 51 63 38 64 d6 43 8c 61 a9 ce | 1 | bad | 50.0 | 2015-07-23
+ 55 b1 51 67 38 64 d6 43 8c 61 a9 cf | 2 | good | 100.0 | 2015-07-23
+ (2 rows)
+```
+```sql
+ SELECT _id, * FROM orders WHERE _id = ObjectId('55b151633864d6438c61a9ce');
+```
+```sql
+ _id | orderkey | orderstatus | totalprice | orderdate
+ -------------------------------------+----------+-------------+------------+------------
+ 55 b1 51 63 38 64 d6 43 8c 61 a9 ce | 1 | bad | 50.0 | 2015-07-23
+ (1 row)
+```
+
+可以将`_id`字段呈现为可读值,并转换为`VARCHAR`:
+
+```sql
+ SELECT CAST(_id AS VARCHAR), * FROM orders WHERE _id = ObjectId('55b151633864d6438c61a9ce');
+```
+```sql
+
+ _id | orderkey | orderstatus | totalprice | orderdate
+ ---------------------------+----------+-------------+------------+------------
+ 55b151633864d6438c61a9ce | 1 | bad | 50.0 | 2015-07-23
+ (1 row)
+```
+
+限制
+-----------
+>- 不支持使用[删除行](../sql/delete.md)
+>- 不支持视图创建
+>- MongoDB查询不到openLooKeng新建的空表和字段。
+>- MongoDB删除表、字段后,openLooKeng中表格或者字段仍存在,但是值为NULL。
+
+
+
+
+
+
diff --git a/hetu-mongodb/pom.xml b/hetu-mongodb/pom.xml
new file mode 100644
index 000000000..81d21a73a
--- /dev/null
+++ b/hetu-mongodb/pom.xml
@@ -0,0 +1,200 @@
+
+
+
+ presto-root
+ io.hetu.core
+ 1.0.0-SNAPSHOT
+
+ 4.0.0
+
+ hetu-mongodb
+ Hetu - mongodb Connector
+ hetu-plugin
+
+
+ ${project.parent.basedir}
+ 3.6.0
+ 4.0.32.Final
+
+
+
+
+ org.mongodb
+ mongo-java-driver
+ ${mongo-java.version}
+
+
+
+ joda-time
+ joda-time
+
+
+
+ javax.validation
+ validation-api
+
+
+
+ io.airlift
+ bootstrap
+
+
+
+ io.airlift
+ json
+
+
+
+ io.airlift
+ log
+
+
+
+ io.airlift
+ configuration
+
+
+
+ com.google.guava
+ guava
+
+
+
+ com.google.inject
+ guice
+
+
+
+ javax.inject
+ javax.inject
+
+
+
+ com.fasterxml.jackson.core
+ jackson-core
+
+
+
+ com.fasterxml.jackson.core
+ jackson-databind
+
+
+
+
+ io.airlift
+ log-manager
+ runtime
+
+
+
+
+ io.hetu.core
+ presto-spi
+ provided
+
+
+
+ io.airlift
+ slice
+ provided
+
+
+
+ com.fasterxml.jackson.core
+ jackson-annotations
+ provided
+
+
+
+ org.openjdk.jol
+ jol-core
+ provided
+
+
+
+
+ io.hetu.core
+ presto-tests
+ test
+
+
+
+ io.hetu.core
+ presto-main
+ test
+
+
+
+ io.hetu.core
+ presto-main
+ test-jar
+ test
+
+
+
+ io.hetu.core
+ presto-tpch
+ test
+
+
+
+ io.airlift.tpch
+ tpch
+ test
+
+
+
+ io.airlift
+ testing
+ test
+
+
+
+ org.testcontainers
+ mongodb
+ 1.15.0
+
+
+ org.slf4j
+ slf4j-api
+
+
+ com.fasterxml.jackson.core
+ jackson-annotations
+
+
+ net.java.dev.jna
+ jna
+
+
+
+
+
+ net.java.dev.jna
+ jna
+ 5.5.0
+ test
+
+
+
+ org.testng
+ testng
+ test
+
+
+
+ org.assertj
+ assertj-core
+ test
+
+
+
+ io.netty
+ netty-transport
+ ${netty.version}
+ test
+
+
+
\ No newline at end of file
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoClientConfig.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoClientConfig.java
new file mode 100644
index 000000000..48f418916
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoClientConfig.java
@@ -0,0 +1,324 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableList;
+import com.mongodb.MongoCredential;
+import com.mongodb.ServerAddress;
+import io.airlift.configuration.Config;
+import io.airlift.configuration.DefunctConfig;
+
+import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
+import javax.validation.constraints.Size;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.mongodb.MongoCredential.createCredential;
+
+@DefunctConfig("mongodb.connection-per-host")
+public class MongoClientConfig
+{
+ private static final Splitter SPLITTER = Splitter.on(',').trimResults().omitEmptyStrings();
+ private static final Splitter PORT_SPLITTER = Splitter.on(':').trimResults().omitEmptyStrings();
+
+ private String schemaCollection = "_schema";
+ private boolean caseInsensitiveNameMatching;
+ private List seeds = ImmutableList.of();
+ private List credentials = ImmutableList.of();
+
+ private int minConnectionsPerHost;
+ private int connectionsPerHost = 100;
+ private int maxWaitTime = 120_000;
+ private int connectionTimeout = 10_000;
+ private int socketTimeout;
+ private int maxConnectionIdleTime;
+ private boolean socketKeepAlive;
+ private boolean sslEnabled;
+
+ // query configurations
+ private int cursorBatchSize; // use driver default
+
+ private ReadPreferenceType readPreference = ReadPreferenceType.PRIMARY;
+ private WriteConcernType writeConcern = WriteConcernType.ACKNOWLEDGED;
+ private String requiredReplicaSetName;
+ private String implicitRowFieldPrefix = "_pos";
+
+ @NotNull
+ public String getSchemaCollection()
+ {
+ return schemaCollection;
+ }
+
+ @Config("mongodb.schema-collection")
+ public MongoClientConfig setSchemaCollection(String schemaCollection)
+ {
+ this.schemaCollection = schemaCollection;
+ return this;
+ }
+
+ public boolean isCaseInsensitiveNameMatching()
+ {
+ return caseInsensitiveNameMatching;
+ }
+
+ @Config("mongodb.case-insensitive-name-matching")
+ public MongoClientConfig setCaseInsensitiveNameMatching(boolean caseInsensitiveNameMatching)
+ {
+ this.caseInsensitiveNameMatching = caseInsensitiveNameMatching;
+ return this;
+ }
+
+ @NotNull
+ @Size(min = 1)
+ public List getSeeds()
+ {
+ return seeds;
+ }
+
+ @Config("mongodb.seeds")
+ public MongoClientConfig setSeeds(String commaSeparatedList)
+ {
+ this.seeds = buildSeeds(SPLITTER.split(commaSeparatedList));
+ return this;
+ }
+
+ public MongoClientConfig setSeeds(String... seeds)
+ {
+ this.seeds = buildSeeds(Arrays.asList(seeds));
+ return this;
+ }
+
+ @NotNull
+ public List getCredentials()
+ {
+ return credentials;
+ }
+
+ @Config("mongodb.credentials")
+ public MongoClientConfig setCredentials(String credentials)
+ {
+ this.credentials = buildCredentials(SPLITTER.split(credentials));
+ return this;
+ }
+
+ public MongoClientConfig setCredentials(String... credentials)
+ {
+ this.credentials = buildCredentials(Arrays.asList(credentials));
+ return this;
+ }
+
+ private List buildSeeds(Iterable hostPorts)
+ {
+ ImmutableList.Builder builder = ImmutableList.builder();
+ for (String hostPort : hostPorts) {
+ List values = PORT_SPLITTER.splitToList(hostPort);
+ checkArgument(values.size() == 1 || values.size() == 2, "Invalid ServerAddress format. Requires host[:port]");
+ if (values.size() == 1) {
+ builder.add(new ServerAddress(values.get(0)));
+ }
+ else {
+ builder.add(new ServerAddress(values.get(0), Integer.parseInt(values.get(1))));
+ }
+ }
+ return builder.build();
+ }
+
+ private List buildCredentials(Iterable userPasses)
+ {
+ ImmutableList.Builder builder = ImmutableList.builder();
+ for (String userPassDatabase : userPasses) {
+ int lastIndex = userPassDatabase.lastIndexOf('@');
+ checkArgument(lastIndex > 0, "Invalid Credential format. Requires user:password@database");
+ String userPass = userPassDatabase.substring(0, lastIndex);
+ String database = userPassDatabase.substring(lastIndex + 1);
+
+ int firstIndex = userPass.indexOf(':');
+ checkArgument(firstIndex > 0, "Invalid Credential format. Requires user:password@database");
+ String user = userPass.substring(0, firstIndex);
+ String password = userPass.substring(firstIndex + 1);
+
+ builder.add(createCredential(user, database, password.toCharArray()));
+ }
+ return builder.build();
+ }
+
+ @Min(0)
+ public int getMinConnectionsPerHost()
+ {
+ return minConnectionsPerHost;
+ }
+
+ @Config("mongodb.min-connections-per-host")
+ public MongoClientConfig setMinConnectionsPerHost(int minConnectionsPerHost)
+ {
+ this.minConnectionsPerHost = minConnectionsPerHost;
+ return this;
+ }
+
+ @Min(1)
+ public int getConnectionsPerHost()
+ {
+ return connectionsPerHost;
+ }
+
+ @Config("mongodb.connections-per-host")
+ public MongoClientConfig setConnectionsPerHost(int connectionsPerHost)
+ {
+ this.connectionsPerHost = connectionsPerHost;
+ return this;
+ }
+
+ @Min(0)
+ public int getMaxWaitTime()
+ {
+ return maxWaitTime;
+ }
+
+ @Config("mongodb.max-wait-time")
+ public MongoClientConfig setMaxWaitTime(int maxWaitTime)
+ {
+ this.maxWaitTime = maxWaitTime;
+ return this;
+ }
+
+ @Min(0)
+ public int getConnectionTimeout()
+ {
+ return connectionTimeout;
+ }
+
+ @Config("mongodb.connection-timeout")
+ public MongoClientConfig setConnectionTimeout(int connectionTimeout)
+ {
+ this.connectionTimeout = connectionTimeout;
+ return this;
+ }
+
+ @Min(0)
+ public int getSocketTimeout()
+ {
+ return socketTimeout;
+ }
+
+ @Config("mongodb.socket-timeout")
+ public MongoClientConfig setSocketTimeout(int socketTimeout)
+ {
+ this.socketTimeout = socketTimeout;
+ return this;
+ }
+
+ public boolean getSocketKeepAlive()
+ {
+ return socketKeepAlive;
+ }
+
+ @Config("mongodb.socket-keep-alive")
+ public MongoClientConfig setSocketKeepAlive(boolean socketKeepAlive)
+ {
+ this.socketKeepAlive = socketKeepAlive;
+ return this;
+ }
+
+ @NotNull
+ public ReadPreferenceType getReadPreference()
+ {
+ return readPreference;
+ }
+
+ @Config("mongodb.read-preference")
+ public MongoClientConfig setReadPreference(ReadPreferenceType readPreference)
+ {
+ this.readPreference = readPreference;
+ return this;
+ }
+
+ @NotNull
+ public WriteConcernType getWriteConcern()
+ {
+ return writeConcern;
+ }
+
+ @Config("mongodb.write-concern")
+ public MongoClientConfig setWriteConcern(WriteConcernType writeConcern)
+ {
+ this.writeConcern = writeConcern;
+ return this;
+ }
+
+ public String getRequiredReplicaSetName()
+ {
+ return requiredReplicaSetName;
+ }
+
+ @Config("mongodb.required-replica-set")
+ public MongoClientConfig setRequiredReplicaSetName(String requiredReplicaSetName)
+ {
+ this.requiredReplicaSetName = requiredReplicaSetName;
+ return this;
+ }
+
+ public int getCursorBatchSize()
+ {
+ return cursorBatchSize;
+ }
+
+ @Config("mongodb.cursor-batch-size")
+ public MongoClientConfig setCursorBatchSize(int cursorBatchSize)
+ {
+ this.cursorBatchSize = cursorBatchSize;
+ return this;
+ }
+
+ @NotNull
+ public String getImplicitRowFieldPrefix()
+ {
+ return implicitRowFieldPrefix;
+ }
+
+ @Config("mongodb.implicit-row-field-prefix")
+ public MongoClientConfig setImplicitRowFieldPrefix(String implicitRowFieldPrefix)
+ {
+ this.implicitRowFieldPrefix = implicitRowFieldPrefix;
+ return this;
+ }
+
+ public boolean getSslEnabled()
+ {
+ return this.sslEnabled;
+ }
+
+ @Config("mongodb.ssl.enabled")
+ public MongoClientConfig setSslEnabled(boolean sslEnabled)
+ {
+ this.sslEnabled = sslEnabled;
+ return this;
+ }
+
+ @Min(0)
+ public int getMaxConnectionIdleTime()
+ {
+ return maxConnectionIdleTime;
+ }
+
+ @Config("mongodb.max-connection-idle-time")
+ public MongoClientConfig setMaxConnectionIdleTime(int maxConnectionIdleTime)
+ {
+ this.maxConnectionIdleTime = maxConnectionIdleTime;
+ return this;
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoClientModule.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoClientModule.java
new file mode 100644
index 000000000..5835835d6
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoClientModule.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.google.inject.Binder;
+import com.google.inject.Module;
+import com.google.inject.Provides;
+import com.google.inject.Scopes;
+import com.mongodb.MongoClient;
+import com.mongodb.MongoClientOptions;
+import io.prestosql.spi.type.TypeManager;
+
+import javax.inject.Singleton;
+
+import static io.airlift.configuration.ConfigBinder.configBinder;
+import static java.util.Objects.requireNonNull;
+
+public class MongoClientModule
+ implements Module
+{
+ @Override
+ public void configure(Binder binder)
+ {
+ binder.bind(MongoConnector.class).in(Scopes.SINGLETON);
+ binder.bind(MongoSplitManager.class).in(Scopes.SINGLETON);
+ binder.bind(MongoPageSourceProvider.class).in(Scopes.SINGLETON);
+ binder.bind(MongoPageSinkProvider.class).in(Scopes.SINGLETON);
+
+ configBinder(binder).bindConfig(MongoClientConfig.class);
+ }
+
+ @Singleton
+ @Provides
+ public static MongoSession createMongoSession(TypeManager typeManager, MongoClientConfig config)
+ {
+ requireNonNull(config, "config is null");
+
+ MongoClientOptions.Builder options = MongoClientOptions.builder();
+
+ options.connectionsPerHost(config.getConnectionsPerHost())
+ .connectTimeout(config.getConnectionTimeout())
+ .socketTimeout(config.getSocketTimeout())
+ .socketKeepAlive(config.getSocketKeepAlive())
+ .sslEnabled(config.getSslEnabled())
+ .maxWaitTime(config.getMaxWaitTime())
+ .maxConnectionIdleTime(config.getMaxConnectionIdleTime())
+ .minConnectionsPerHost(config.getMinConnectionsPerHost())
+ .readPreference(config.getReadPreference().getReadPreference())
+ .writeConcern(config.getWriteConcern().getWriteConcern());
+
+ if (config.getRequiredReplicaSetName() != null) {
+ options.requiredReplicaSetName(config.getRequiredReplicaSetName());
+ }
+
+ MongoClient client = new MongoClient(config.getSeeds(), config.getCredentials(), options.build());
+
+ return new MongoSession(
+ typeManager,
+ client,
+ config);
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoColumnHandle.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoColumnHandle.java
new file mode 100644
index 000000000..364688099
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoColumnHandle.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.prestosql.spi.connector.ColumnHandle;
+import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.type.Type;
+import org.bson.Document;
+
+import java.util.Objects;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+import static java.util.Objects.requireNonNull;
+
+public class MongoColumnHandle
+ implements ColumnHandle
+{
+ private final String name;
+ private final Type type;
+ private final boolean hidden;
+
+ @JsonCreator
+ public MongoColumnHandle(
+ @JsonProperty("name") String name,
+ @JsonProperty("columnType") Type type,
+ @JsonProperty("hidden") boolean hidden)
+ {
+ this.name = requireNonNull(name, "name is null");
+ this.type = requireNonNull(type, "columnType is null");
+ this.hidden = hidden;
+ }
+
+ @JsonProperty
+ public String getName()
+ {
+ return name;
+ }
+
+ @JsonProperty("columnType")
+ public Type getType()
+ {
+ return type;
+ }
+
+ @JsonProperty
+ public boolean isHidden()
+ {
+ return hidden;
+ }
+
+ public ColumnMetadata toColumnMetadata()
+ {
+ return new ColumnMetadata(name, type, null, hidden);
+ }
+
+ public Document getDocument()
+ {
+ return new Document().append("name", name)
+ .append("type", type.getTypeSignature().toString())
+ .append("hidden", hidden);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(name, type, hidden);
+ }
+
+ @Override
+ public boolean equals(Object obj)
+ {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ MongoColumnHandle other = (MongoColumnHandle) obj;
+ return Objects.equals(name, other.name) &&
+ Objects.equals(type, other.type) &&
+ Objects.equals(hidden, other.hidden);
+ }
+
+ @Override
+ public String toString()
+ {
+ return toStringHelper(this)
+ .add("name", name)
+ .add("type", type)
+ .add("hidden", hidden)
+ .toString();
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoConnector.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoConnector.java
new file mode 100644
index 000000000..2c717e31e
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoConnector.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import io.prestosql.spi.connector.Connector;
+import io.prestosql.spi.connector.ConnectorMetadata;
+import io.prestosql.spi.connector.ConnectorPageSinkProvider;
+import io.prestosql.spi.connector.ConnectorPageSourceProvider;
+import io.prestosql.spi.connector.ConnectorSplitManager;
+import io.prestosql.spi.connector.ConnectorTransactionHandle;
+import io.prestosql.spi.transaction.IsolationLevel;
+
+import javax.inject.Inject;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static io.prestosql.spi.transaction.IsolationLevel.READ_UNCOMMITTED;
+import static io.prestosql.spi.transaction.IsolationLevel.checkConnectorSupports;
+import static java.util.Objects.requireNonNull;
+
+public class MongoConnector
+ implements Connector
+{
+ private final MongoSession mongoSession;
+ private final MongoSplitManager splitManager;
+ private final MongoPageSourceProvider pageSourceProvider;
+ private final MongoPageSinkProvider pageSinkProvider;
+
+ private final ConcurrentMap transactions = new ConcurrentHashMap<>();
+
+ @Inject
+ public MongoConnector(
+ MongoSession mongoSession,
+ MongoSplitManager splitManager,
+ MongoPageSourceProvider pageSourceProvider,
+ MongoPageSinkProvider pageSinkProvider)
+ {
+ this.mongoSession = mongoSession;
+ this.splitManager = requireNonNull(splitManager, "splitManager is null");
+ this.pageSourceProvider = requireNonNull(pageSourceProvider, "pageSourceProvider is null");
+ this.pageSinkProvider = requireNonNull(pageSinkProvider, "pageSinkProvider is null");
+ }
+
+ @Override
+ public ConnectorTransactionHandle beginTransaction(IsolationLevel isolationLevel, boolean readOnly)
+ {
+ checkConnectorSupports(READ_UNCOMMITTED, isolationLevel);
+ MongoTransactionHandle transaction = new MongoTransactionHandle();
+ transactions.put(transaction, new MongoMetadata(mongoSession));
+ return transaction;
+ }
+
+ @Override
+ public boolean isSingleStatementWritesOnly()
+ {
+ return true;
+ }
+
+ @Override
+ public ConnectorMetadata getMetadata(ConnectorTransactionHandle transaction)
+ {
+ MongoMetadata metadata = transactions.get(transaction);
+ checkArgument(metadata != null, "no such transaction: %s", transaction);
+ return metadata;
+ }
+
+ @Override
+ public void commit(ConnectorTransactionHandle transaction)
+ {
+ checkArgument(transactions.remove(transaction) != null, "no such transaction: %s", transaction);
+ }
+
+ @Override
+ public void rollback(ConnectorTransactionHandle transaction)
+ {
+ MongoMetadata metadata = transactions.remove(transaction);
+ checkArgument(metadata != null, "no such transaction: %s", transaction);
+ metadata.rollback();
+ }
+
+ @Override
+ public ConnectorSplitManager getSplitManager()
+ {
+ return splitManager;
+ }
+
+ @Override
+ public ConnectorPageSourceProvider getPageSourceProvider()
+ {
+ return pageSourceProvider;
+ }
+
+ @Override
+ public ConnectorPageSinkProvider getPageSinkProvider()
+ {
+ return pageSinkProvider;
+ }
+
+ @Override
+ public void shutdown()
+ {
+ mongoSession.shutdown();
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoConnectorFactory.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoConnectorFactory.java
new file mode 100644
index 000000000..5374c20c4
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoConnectorFactory.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.google.inject.Injector;
+import io.airlift.bootstrap.Bootstrap;
+import io.airlift.json.JsonModule;
+import io.prestosql.spi.connector.Connector;
+import io.prestosql.spi.connector.ConnectorContext;
+import io.prestosql.spi.connector.ConnectorFactory;
+import io.prestosql.spi.connector.ConnectorHandleResolver;
+import io.prestosql.spi.type.TypeManager;
+
+import java.util.Map;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static java.util.Objects.requireNonNull;
+
+public class MongoConnectorFactory
+ implements ConnectorFactory
+{
+ private final String name;
+
+ public MongoConnectorFactory(String name)
+ {
+ checkArgument(!isNullOrEmpty(name), "name is null or empty");
+ this.name = name;
+ }
+
+ @Override
+ public String getName()
+ {
+ return name;
+ }
+
+ @Override
+ public ConnectorHandleResolver getHandleResolver()
+ {
+ return new MongoHandleResolver();
+ }
+
+ @Override
+ public Connector create(String catalogName, Map config, ConnectorContext context)
+ {
+ requireNonNull(config, "config is null");
+
+ Bootstrap app = new Bootstrap(
+ new JsonModule(),
+ new MongoClientModule(),
+ binder -> binder.bind(TypeManager.class).toInstance(context.getTypeManager()));
+
+ Injector injector = app.strictConfig().doNotInitializeLogging()
+ .setRequiredConfigurationProperties(config)
+ .initialize();
+
+ return injector.getInstance(MongoConnector.class);
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoHandleResolver.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoHandleResolver.java
new file mode 100644
index 000000000..28a6579f0
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoHandleResolver.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import io.prestosql.spi.connector.ColumnHandle;
+import io.prestosql.spi.connector.ConnectorHandleResolver;
+import io.prestosql.spi.connector.ConnectorInsertTableHandle;
+import io.prestosql.spi.connector.ConnectorOutputTableHandle;
+import io.prestosql.spi.connector.ConnectorSplit;
+import io.prestosql.spi.connector.ConnectorTableHandle;
+import io.prestosql.spi.connector.ConnectorTransactionHandle;
+
+public class MongoHandleResolver
+ implements ConnectorHandleResolver
+{
+ public MongoHandleResolver()
+ {
+ }
+
+ @Override
+ public Class extends ConnectorTransactionHandle> getTransactionHandleClass()
+ {
+ return MongoTransactionHandle.class;
+ }
+
+ @Override
+ public Class extends ConnectorTableHandle> getTableHandleClass()
+ {
+ return MongoTableHandle.class;
+ }
+
+ @Override
+ public Class extends ColumnHandle> getColumnHandleClass()
+ {
+ return MongoColumnHandle.class;
+ }
+
+ @Override
+ public Class extends ConnectorSplit> getSplitClass()
+ {
+ return MongoSplit.class;
+ }
+
+ @Override
+ public Class extends ConnectorOutputTableHandle> getOutputTableHandleClass()
+ {
+ return MongoOutputTableHandle.class;
+ }
+
+ @Override
+ public Class extends ConnectorInsertTableHandle> getInsertTableHandleClass()
+ {
+ return MongoInsertTableHandle.class;
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoIndex.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoIndex.java
new file mode 100644
index 000000000..164c05422
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoIndex.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.google.common.collect.ImmutableList;
+import com.mongodb.client.ListIndexesIterable;
+import io.prestosql.spi.block.SortOrder;
+import org.bson.Document;
+
+import java.util.List;
+import java.util.Optional;
+
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+public class MongoIndex
+{
+ private final String name;
+ private final List keys;
+ private final boolean unique;
+
+ public static List parse(ListIndexesIterable indexes)
+ {
+ ImmutableList.Builder builder = ImmutableList.builder();
+ for (Document index : indexes) {
+ // TODO: v, ns, sparse fields
+ Document key = (Document) index.get("key");
+ String name = index.getString("name");
+ boolean unique = index.getBoolean("unique", false);
+
+ if (key.containsKey("_fts")) { // Full Text Search
+ continue;
+ }
+ builder.add(new MongoIndex(name, parseKey(key), unique));
+ }
+
+ return builder.build();
+ }
+
+ private static List parseKey(Document key)
+ {
+ ImmutableList.Builder builder = ImmutableList.builder();
+
+ for (String name : key.keySet()) {
+ Object value = key.get(name);
+ if (value instanceof Number) {
+ int order = ((Number) value).intValue();
+ checkState(order == 1 || order == -1, "Unknown index sort order");
+ builder.add(new MongodbIndexKey(name, order == 1 ? SortOrder.ASC_NULLS_LAST : SortOrder.DESC_NULLS_LAST));
+ }
+ else if (value instanceof String) {
+ builder.add(new MongodbIndexKey(name, (String) value));
+ }
+ else {
+ throw new UnsupportedOperationException("Unknown index type: " + value.toString());
+ }
+ }
+
+ return builder.build();
+ }
+
+ public MongoIndex(String name, List keys, boolean unique)
+ {
+ this.name = name;
+ this.keys = keys;
+ this.unique = unique;
+ }
+
+ public String getName()
+ {
+ return name;
+ }
+
+ public List getKeys()
+ {
+ return keys;
+ }
+
+ public boolean isUnique()
+ {
+ return unique;
+ }
+
+ public static class MongodbIndexKey
+ {
+ private final String name;
+ private final Optional sortOrder;
+ private final Optional type;
+
+ public MongodbIndexKey(String name, SortOrder sortOrder)
+ {
+ this(name, Optional.of(sortOrder), Optional.empty());
+ }
+
+ public MongodbIndexKey(String name, String type)
+ {
+ this(name, Optional.empty(), Optional.of(type));
+ }
+
+ public MongodbIndexKey(String name, Optional sortOrder, Optional type)
+ {
+ this.name = requireNonNull(name, "name is null");
+ this.sortOrder = sortOrder;
+ this.type = type;
+ }
+
+ public String getName()
+ {
+ return name;
+ }
+
+ public Optional getSortOrder()
+ {
+ return sortOrder;
+ }
+
+ public Optional getType()
+ {
+ return type;
+ }
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoInsertTableHandle.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoInsertTableHandle.java
new file mode 100644
index 000000000..8fefb8146
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoInsertTableHandle.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableList;
+import io.prestosql.spi.connector.ConnectorInsertTableHandle;
+import io.prestosql.spi.connector.SchemaTableName;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+public class MongoInsertTableHandle
+ implements ConnectorInsertTableHandle
+{
+ private final SchemaTableName schemaTableName;
+ private final List columns;
+
+ @JsonCreator
+ public MongoInsertTableHandle(
+ @JsonProperty("schemaTableName") SchemaTableName schemaTableName,
+ @JsonProperty("columns") List columns)
+ {
+ this.schemaTableName = requireNonNull(schemaTableName, "schemaTableName is null");
+ this.columns = ImmutableList.copyOf(requireNonNull(columns, "columns is null"));
+ }
+
+ @JsonProperty
+ public SchemaTableName getSchemaTableName()
+ {
+ return schemaTableName;
+ }
+
+ @JsonProperty
+ public List getColumns()
+ {
+ return columns;
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoMetadata.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoMetadata.java
new file mode 100644
index 000000000..b887f6997
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoMetadata.java
@@ -0,0 +1,310 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.airlift.log.Logger;
+import io.airlift.slice.Slice;
+import io.hetu.core.plugin.mongodb.MongoIndex.MongodbIndexKey;
+import io.prestosql.spi.connector.ColumnHandle;
+import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.connector.ConnectorInsertTableHandle;
+import io.prestosql.spi.connector.ConnectorMetadata;
+import io.prestosql.spi.connector.ConnectorNewTableLayout;
+import io.prestosql.spi.connector.ConnectorOutputMetadata;
+import io.prestosql.spi.connector.ConnectorOutputTableHandle;
+import io.prestosql.spi.connector.ConnectorSession;
+import io.prestosql.spi.connector.ConnectorTableHandle;
+import io.prestosql.spi.connector.ConnectorTableMetadata;
+import io.prestosql.spi.connector.ConnectorTableProperties;
+import io.prestosql.spi.connector.Constraint;
+import io.prestosql.spi.connector.ConstraintApplicationResult;
+import io.prestosql.spi.connector.LocalProperty;
+import io.prestosql.spi.connector.NotFoundException;
+import io.prestosql.spi.connector.SchemaTableName;
+import io.prestosql.spi.connector.SchemaTablePrefix;
+import io.prestosql.spi.connector.SortingProperty;
+import io.prestosql.spi.connector.TableNotFoundException;
+import io.prestosql.spi.predicate.TupleDomain;
+import io.prestosql.spi.statistics.ComputedStatistics;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Locale.ENGLISH;
+import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toList;
+
+public class MongoMetadata
+ implements ConnectorMetadata
+{
+ private static final Logger log = Logger.get(MongoMetadata.class);
+
+ private final MongoSession mongoSession;
+
+ private final AtomicReference rollbackAction = new AtomicReference<>();
+
+ public MongoMetadata(MongoSession mongoSession)
+ {
+ this.mongoSession = requireNonNull(mongoSession, "mongoSession is null");
+ }
+
+ @Override
+ public List listSchemaNames(ConnectorSession session)
+ {
+ return mongoSession.getAllSchemas();
+ }
+
+ @Override
+ public MongoTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName)
+ {
+ requireNonNull(tableName, "tableName is null");
+ try {
+ return mongoSession.getTable(tableName).getTableHandle();
+ }
+ catch (TableNotFoundException e) {
+ log.debug(e, "Table(%s) not found", tableName);
+ return null;
+ }
+ }
+
+ @Override
+ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
+ {
+ requireNonNull(tableHandle, "tableHandle is null");
+ SchemaTableName tableName = getTableName(tableHandle);
+ return getTableMetadata(session, tableName);
+ }
+
+ @Override
+ public List listTables(ConnectorSession session, Optional optionalSchemaName)
+ {
+ List schemaNames = optionalSchemaName.map(ImmutableList::of)
+ .orElseGet(() -> (ImmutableList) listSchemaNames(session));
+ ImmutableList.Builder tableNames = ImmutableList.builder();
+ for (String schemaName : schemaNames) {
+ for (String tableName : mongoSession.getAllTables(schemaName)) {
+ tableNames.add(new SchemaTableName(schemaName, tableName.toLowerCase(ENGLISH)));
+ }
+ }
+ return tableNames.build();
+ }
+
+ @Override
+ public Map getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
+ {
+ MongoTableHandle table = (MongoTableHandle) tableHandle;
+ List columns = mongoSession.getTable(table.getSchemaTableName()).getColumns();
+
+ ImmutableMap.Builder columnHandles = ImmutableMap.builder();
+ for (MongoColumnHandle columnHandle : columns) {
+ columnHandles.put(columnHandle.getName(), columnHandle);
+ }
+ return columnHandles.build();
+ }
+
+ @Override
+ public Map> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
+ {
+ requireNonNull(prefix, "prefix is null");
+ ImmutableMap.Builder> columns = ImmutableMap.builder();
+ for (SchemaTableName tableName : listTables(session, prefix)) {
+ try {
+ columns.put(tableName, getTableMetadata(session, tableName).getColumns());
+ }
+ catch (NotFoundException e) {
+ // table disappeared during listing operation
+ }
+ }
+ return columns.build();
+ }
+
+ private List listTables(ConnectorSession session, SchemaTablePrefix prefix)
+ {
+ if (!prefix.getTable().isPresent()) {
+ return listTables(session, prefix.getSchema());
+ }
+ return ImmutableList.of(prefix.toSchemaTableName());
+ }
+
+ @Override
+ public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle)
+ {
+ return ((MongoColumnHandle) columnHandle).toColumnMetadata();
+ }
+
+ @Override
+ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, boolean ignoreExisting)
+ {
+ mongoSession.createTable(tableMetadata.getTable(), buildColumnHandles(tableMetadata));
+ }
+
+ @Override
+ public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle)
+ {
+ MongoTableHandle table = (MongoTableHandle) tableHandle;
+
+ mongoSession.dropTable(table.getSchemaTableName());
+ }
+
+ @Override
+ public void renameTable(ConnectorSession session, ConnectorTableHandle tableHandle, SchemaTableName newTableName)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnMetadata column)
+ {
+ mongoSession.addColumn(((MongoTableHandle) tableHandle).getSchemaTableName(), column);
+ }
+
+ @Override
+ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional layout)
+ {
+ List columns = buildColumnHandles(tableMetadata);
+
+ mongoSession.createTable(tableMetadata.getTable(), columns);
+
+ setRollback(() -> mongoSession.dropTable(tableMetadata.getTable()));
+
+ return new MongoOutputTableHandle(
+ tableMetadata.getTable(),
+ columns.stream().filter(c -> !c.isHidden()).collect(toList()));
+ }
+
+ @Override
+ public Optional finishCreateTable(ConnectorSession session, ConnectorOutputTableHandle tableHandle, Collection fragments, Collection computedStatistics)
+ {
+ clearRollback();
+ return Optional.empty();
+ }
+
+ @Override
+ public ConnectorInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle)
+ {
+ MongoTableHandle table = (MongoTableHandle) tableHandle;
+ List columns = mongoSession.getTable(table.getSchemaTableName()).getColumns();
+
+ return new MongoInsertTableHandle(
+ table.getSchemaTableName(),
+ columns.stream().filter(c -> !c.isHidden()).collect(toList()));
+ }
+
+ @Override
+ public Optional finishInsert(ConnectorSession session, ConnectorInsertTableHandle insertHandle, Collection fragments, Collection computedStatistics)
+ {
+ return Optional.empty();
+ }
+
+ @Override
+ public boolean usesLegacyTableLayouts()
+ {
+ return false;
+ }
+
+ @Override
+ public ConnectorTableProperties getTableProperties(ConnectorSession session, ConnectorTableHandle table)
+ {
+ MongoTableHandle tableHandle = (MongoTableHandle) table;
+
+ Optional> partitioningColumns = Optional.empty(); //TODO: sharding key
+ ImmutableList.Builder> localProperties = ImmutableList.builder();
+
+ MongoTable tableInfo = mongoSession.getTable(tableHandle.getSchemaTableName());
+ Map columns = getColumnHandles(session, tableHandle);
+
+ for (MongoIndex index : tableInfo.getIndexes()) {
+ for (MongodbIndexKey key : index.getKeys()) {
+ if (!key.getSortOrder().isPresent()) {
+ continue;
+ }
+ if (columns.get(key.getName()) != null) {
+ localProperties.add(new SortingProperty<>(columns.get(key.getName()), key.getSortOrder().get()));
+ }
+ }
+ }
+
+ return new ConnectorTableProperties(
+ TupleDomain.all(),
+ Optional.empty(),
+ partitioningColumns,
+ Optional.empty(),
+ localProperties.build());
+ }
+
+ @Override
+ public Optional> applyFilter(ConnectorSession session, ConnectorTableHandle table, Constraint constraint)
+ {
+ MongoTableHandle handle = (MongoTableHandle) table;
+
+ TupleDomain oldDomain = handle.getConstraint();
+ TupleDomain newDomain = oldDomain.intersect(constraint.getSummary());
+ if (oldDomain.equals(newDomain)) {
+ return Optional.empty();
+ }
+
+ handle = new MongoTableHandle(
+ handle.getSchemaTableName(),
+ newDomain);
+
+ return Optional.of(new ConstraintApplicationResult<>(handle, constraint.getSummary()));
+ }
+
+ private void setRollback(Runnable action)
+ {
+ checkState(rollbackAction.compareAndSet(null, action), "rollback action is already set");
+ }
+
+ private void clearRollback()
+ {
+ rollbackAction.set(null);
+ }
+
+ public void rollback()
+ {
+ Optional.ofNullable(rollbackAction.getAndSet(null)).ifPresent(Runnable::run);
+ }
+
+ private static SchemaTableName getTableName(ConnectorTableHandle tableHandle)
+ {
+ return ((MongoTableHandle) tableHandle).getSchemaTableName();
+ }
+
+ private ConnectorTableMetadata getTableMetadata(ConnectorSession session, SchemaTableName tableName)
+ {
+ MongoTableHandle tableHandle = mongoSession.getTable(tableName).getTableHandle();
+
+ List columns = ImmutableList.copyOf(
+ getColumnHandles(session, tableHandle).values().stream()
+ .map(MongoColumnHandle.class::cast)
+ .map(MongoColumnHandle::toColumnMetadata)
+ .collect(toList()));
+
+ return new ConnectorTableMetadata(tableName, columns);
+ }
+
+ private static List buildColumnHandles(ConnectorTableMetadata tableMetadata)
+ {
+ return tableMetadata.getColumns().stream()
+ .map(m -> new MongoColumnHandle(m.getName(), m.getType(), m.isHidden()))
+ .collect(toList());
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoOutputTableHandle.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoOutputTableHandle.java
new file mode 100644
index 000000000..977b08a10
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoOutputTableHandle.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableList;
+import io.prestosql.spi.connector.ConnectorOutputTableHandle;
+import io.prestosql.spi.connector.SchemaTableName;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+public class MongoOutputTableHandle
+ implements ConnectorOutputTableHandle
+{
+ private final SchemaTableName schemaTableName;
+ private final List columns;
+
+ @JsonCreator
+ public MongoOutputTableHandle(
+ @JsonProperty("schemaTableName") SchemaTableName schemaTableName,
+ @JsonProperty("columns") List columns)
+ {
+ this.schemaTableName = requireNonNull(schemaTableName, "schemaTableName is null");
+ this.columns = ImmutableList.copyOf(requireNonNull(columns, "columns is null"));
+ }
+
+ @JsonProperty
+ public SchemaTableName getSchemaTableName()
+ {
+ return schemaTableName;
+ }
+
+ @JsonProperty
+ public List getColumns()
+ {
+ return columns;
+ }
+}
diff --git a/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoPageSink.java b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoPageSink.java
new file mode 100644
index 000000000..cbc1ffccb
--- /dev/null
+++ b/hetu-mongodb/src/main/java/io/hetu/core/plugin/mongodb/MongoPageSink.java
@@ -0,0 +1,256 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.hetu.core.plugin.mongodb;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.primitives.Shorts;
+import com.google.common.primitives.SignedBytes;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.model.InsertManyOptions;
+import io.airlift.slice.Slice;
+import io.prestosql.spi.Page;
+import io.prestosql.spi.PrestoException;
+import io.prestosql.spi.StandardErrorCode;
+import io.prestosql.spi.block.Block;
+import io.prestosql.spi.connector.ConnectorPageSink;
+import io.prestosql.spi.connector.SchemaTableName;
+import io.prestosql.spi.type.BigintType;
+import io.prestosql.spi.type.BooleanType;
+import io.prestosql.spi.type.CharType;
+import io.prestosql.spi.type.DateType;
+import io.prestosql.spi.type.DecimalType;
+import io.prestosql.spi.type.DoubleType;
+import io.prestosql.spi.type.IntegerType;
+import io.prestosql.spi.type.NamedTypeSignature;
+import io.prestosql.spi.type.RealType;
+import io.prestosql.spi.type.SmallintType;
+import io.prestosql.spi.type.TimeType;
+import io.prestosql.spi.type.TimestampType;
+import io.prestosql.spi.type.TimestampWithTimeZoneType;
+import io.prestosql.spi.type.TinyintType;
+import io.prestosql.spi.type.Type;
+import io.prestosql.spi.type.TypeSignatureParameter;
+import io.prestosql.spi.type.VarbinaryType;
+import io.prestosql.spi.type.VarcharType;
+import org.bson.Document;
+import org.bson.types.Binary;
+import org.bson.types.ObjectId;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static io.hetu.core.plugin.mongodb.ObjectIdType.OBJECT_ID;
+import static io.hetu.core.plugin.mongodb.TypeUtils.isArrayType;
+import static io.hetu.core.plugin.mongodb.TypeUtils.isMapType;
+import static io.hetu.core.plugin.mongodb.TypeUtils.isRowType;
+import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.prestosql.spi.type.Chars.padSpaces;
+import static io.prestosql.spi.type.DateTimeEncoding.unpackMillisUtc;
+import static io.prestosql.spi.type.Decimals.readBigDecimal;
+import static java.lang.Float.intBitsToFloat;
+import static java.lang.Math.toIntExact;
+import static java.util.Collections.unmodifiableList;
+import static java.util.Collections.unmodifiableMap;
+import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+
+public class MongoPageSink
+ implements ConnectorPageSink
+{
+ private final MongoSession mongoSession;
+ private final SchemaTableName schemaTableName;
+ private final List columns;
+ private final String implicitPrefix;
+
+ public MongoPageSink(
+ MongoClientConfig config,
+ MongoSession mongoSession,
+ SchemaTableName schemaTableName,
+ List columns)
+ {
+ this.mongoSession = mongoSession;
+ this.schemaTableName = schemaTableName;
+ this.columns = columns;
+ this.implicitPrefix = requireNonNull(config.getImplicitRowFieldPrefix(), "config.getImplicitRowFieldPrefix() is null");
+ }
+
+ @Override
+ public CompletableFuture> appendPage(Page page)
+ {
+ MongoCollection collection = mongoSession.getCollection(schemaTableName);
+ List batch = new ArrayList<>(page.getPositionCount());
+
+ for (int position = 0; position < page.getPositionCount(); position++) {
+ Document doc = new Document();
+
+ for (int channel = 0; channel < page.getChannelCount(); channel++) {
+ MongoColumnHandle column = columns.get(channel);
+ doc.append(column.getName(), getObjectValue(columns.get(channel).getType(), page.getBlock(channel), position));
+ }
+ batch.add(doc);
+ }
+
+ collection.insertMany(batch, new InsertManyOptions().ordered(true));
+ return NOT_BLOCKED;
+ }
+
+ private Object getObjectValue(Type type, Block block, int position)
+ {
+ if (block.isNull(position)) {
+ if (type.equals(OBJECT_ID)) {
+ return new ObjectId();
+ }
+ return null;
+ }
+
+ if (type.equals(OBJECT_ID)) {
+ return new ObjectId(block.getSlice(position, 0, block.getSliceLength(position)).getBytes());
+ }
+ if (type.equals(BooleanType.BOOLEAN)) {
+ return type.getBoolean(block, position);
+ }
+ if (type.equals(BigintType.BIGINT)) {
+ return type.getLong(block, position);
+ }
+ if (type.equals(IntegerType.INTEGER)) {
+ return toIntExact(type.getLong(block, position));
+ }
+ if (type.equals(SmallintType.SMALLINT)) {
+ return Shorts.checkedCast(type.getLong(block, position));
+ }
+ if (type.equals(TinyintType.TINYINT)) {
+ return SignedBytes.checkedCast(type.getLong(block, position));
+ }
+ if (type.equals(RealType.REAL)) {
+ return intBitsToFloat(toIntExact(type.getLong(block, position)));
+ }
+ if (type.equals(DoubleType.DOUBLE)) {
+ return type.getDouble(block, position);
+ }
+ if (type instanceof VarcharType) {
+ return type.getSlice(block, position).toStringUtf8();
+ }
+ if (type instanceof CharType) {
+ return padSpaces(type.getSlice(block, position), ((CharType) type)).toStringUtf8();
+ }
+ if (type.equals(VarbinaryType.VARBINARY)) {
+ return new Binary(type.getSlice(block, position).getBytes());
+ }
+ if (type.equals(DateType.DATE)) {
+ long days = type.getLong(block, position);
+ return new Date(TimeUnit.DAYS.toMillis(days));
+ }
+ if (type.equals(TimeType.TIME)) {
+ long millisUtc = type.getLong(block, position);
+ return new Date(millisUtc);
+ }
+ if (type.equals(TimestampType.TIMESTAMP)) {
+ long millisUtc = type.getLong(block, position);
+ return new Date(millisUtc);
+ }
+ if (type.equals(TimestampWithTimeZoneType.TIMESTAMP_WITH_TIME_ZONE)) {
+ long millisUtc = unpackMillisUtc(type.getLong(block, position));
+ return new Date(millisUtc);
+ }
+ if (type instanceof DecimalType) {
+ return readBigDecimal((DecimalType) type, block, position);
+ }
+ if (isArrayType(type)) {
+ Type elementType = type.getTypeParameters().get(0);
+
+ Block arrayBlock = (Block) block.getObject(position, Block.class);
+
+ List