apache · johntomcat7408-cmyk · Jul 21, 2026 · Jul 29, 2026
diff --git a/docs/docs-cn/source/5.application-development/3.connector/12.mongodb.md b/docs/docs-cn/source/5.application-development/3.connector/12.mongodb.md
@@ -0,0 +1,94 @@
+# MongoDB Connector 介绍
+
+MongoDB Connector 支持读取有界 collection，并将数据插入 MongoDB collection。
+
+## 语法
+
+```sql
+CREATE TABLE mongo_source (
+  id BIGINT,
+  name VARCHAR,
+  active BOOLEAN
+) WITH (
+  type = 'mongodb',
+  `geaflow.dsl.mongodb.uri` = 'mongodb://localhost:27017',
+  `geaflow.dsl.mongodb.database` = 'geaflow',
+  `geaflow.dsl.mongodb.collection` = 'source_records',
+  `geaflow.dsl.mongodb.partition.num` = '4',
+  `geaflow.dsl.mongodb.partition.field` = 'id',
+  `geaflow.dsl.mongodb.partition.lowerbound` = '0',
+  `geaflow.dsl.mongodb.partition.upperbound` = '100'
+);
+```
+
+## 参数
+
+| 参数名 | 是否必须 | 默认值 | 描述 |
+| --- | --- | --- | --- |
+| `geaflow.dsl.mongodb.uri` | 是 | - | MongoDB connection string。 |
+| `geaflow.dsl.mongodb.database` | 是 | - | database 名称。 |
+| `geaflow.dsl.mongodb.collection` | 是 | - | collection 名称。 |
+| `geaflow.dsl.mongodb.batch.size` | 否 | `1000` | sink 每批插入的行数。 |
+| `geaflow.dsl.mongodb.partition.num` | 否 | `1` | source 范围分区数上限。 |
+| `geaflow.dsl.mongodb.partition.field` | 分区数大于 1 时 | - | 使用整数边界拆分的数值型字段。 |
+| `geaflow.dsl.mongodb.partition.lowerbound` | 分区数大于 1 时 | - | 范围下界，包含该值。 |
+| `geaflow.dsl.mongodb.partition.upperbound` | 分区数大于 1 时 | - | 范围上界，不包含该值。 |
+
+分区数为 1 时，source 不使用范围条件。分区数大于 1 时，配置范围会拆分为互不重叠的
+`[lowerbound, upperbound)` 查询条件，范围外的文档不会被读取。实际分区数取
+`partition.num` 和 `upperbound - lowerbound` 中的较小值，避免产生空的整数范围。
+
+对于大小窗口，source 使用 bookmark offset 和固定排序。单分区按 `_id` 排序；范围分区按
+分区字段和 `_id` 排序。范围分区读取建议创建对应的升序复合索引，例如
+`{id: 1, _id: 1}`。全量窗口直接流式读取 MongoDB cursor。source 仅支持静态、有界
+collection，读取期间修改数据仍可能造成遗漏或重复。
+
+sink 使用 ordered insert，在达到批量阈值或窗口结束时写入。不支持 upsert 和
+Exactly-once，也不在 MongoDB driver 的重试行为之外增加重试。
+批量失败时，MongoDB 可能已经写入部分文档。
+
+## 类型映射
+
+| GeaFlow 类型 | BSON 类型 |
+| --- | --- |
+| `VARCHAR` | String |
+| `BOOLEAN` | Boolean |
+| `TINYINT`、`SMALLINT`、`INTEGER` | Int32 |
+| `BIGINT` | Int64 |
+| `FLOAT`、`DOUBLE` | Double |
+| `DECIMAL` | Decimal128 |
+| `DATE`、`TIMESTAMP` | Date |
+
+字段缺失和 BSON null 均读取为 null。表结构中的 `_id` 为 `VARCHAR` 时，source 会将
+BSON ObjectId 转为十六进制字符串；sink 写入的字符串 `_id` 保持 BSON string。
+首版不支持数组和嵌套文档。
+
+## 示例
+
+```sql
+CREATE TABLE mongo_source (
+  id BIGINT,
+  name VARCHAR,
+  active BOOLEAN
+) WITH (
+  type = 'mongodb',
+  `geaflow.dsl.mongodb.uri` = 'mongodb://localhost:27017',
+  `geaflow.dsl.mongodb.database` = 'geaflow',
+  `geaflow.dsl.mongodb.collection` = 'source_records'
+);
+
+CREATE TABLE mongo_sink (
+  id BIGINT,
+  name VARCHAR,
+  active BOOLEAN
+) WITH (
+  type = 'mongodb',
+  `geaflow.dsl.mongodb.uri` = 'mongodb://localhost:27017',
+  `geaflow.dsl.mongodb.database` = 'geaflow',
+  `geaflow.dsl.mongodb.collection` = 'sink_records',
+  `geaflow.dsl.mongodb.batch.size` = '500'
+);
+
+INSERT INTO mongo_sink
+SELECT id, name, active FROM mongo_source;
+```
diff --git a/docs/docs-cn/source/5.application-development/3.connector/index.rst b/docs/docs-cn/source/5.application-development/3.connector/index.rst
@@ -17,4 +17,4 @@
    9.pulsar.md
    10.udc.md
    11.doris.md
-
+   12.mongodb.md
diff --git a/docs/docs-en/source/5.application-development/3.connector/12.mongodb.md b/docs/docs-en/source/5.application-development/3.connector/12.mongodb.md
@@ -0,0 +1,96 @@
+# MongoDB Connector
+
+The MongoDB connector reads a bounded collection and inserts rows into a collection.
+
+## Syntax
+
+```sql
+CREATE TABLE mongo_source (
+  id BIGINT,
+  name VARCHAR,
+  active BOOLEAN
+) WITH (
+  type = 'mongodb',
+  `geaflow.dsl.mongodb.uri` = 'mongodb://localhost:27017',
+  `geaflow.dsl.mongodb.database` = 'geaflow',
+  `geaflow.dsl.mongodb.collection` = 'source_records',
+  `geaflow.dsl.mongodb.partition.num` = '4',
+  `geaflow.dsl.mongodb.partition.field` = 'id',
+  `geaflow.dsl.mongodb.partition.lowerbound` = '0',
+  `geaflow.dsl.mongodb.partition.upperbound` = '100'
+);
+```
+
+## Options
+
+| Key | Required | Default | Description |
+| --- | --- | --- | --- |
+| `geaflow.dsl.mongodb.uri` | Yes | - | MongoDB connection string. |
+| `geaflow.dsl.mongodb.database` | Yes | - | Database name. |
+| `geaflow.dsl.mongodb.collection` | Yes | - | Collection name. |
+| `geaflow.dsl.mongodb.batch.size` | No | `1000` | Number of rows in each sink insert batch. |
+| `geaflow.dsl.mongodb.partition.num` | No | `1` | Maximum number of source range partitions. |
+| `geaflow.dsl.mongodb.partition.field` | When partition number is greater than 1 | - | Numeric field split with integer range bounds. |
+| `geaflow.dsl.mongodb.partition.lowerbound` | For multiple partitions | - | Inclusive range lower bound. |
+| `geaflow.dsl.mongodb.partition.upperbound` | For multiple partitions | - | Exclusive range upper bound. |
+
+With one partition, the source scans the collection without a range filter. With multiple
+partitions, the configured range is split into non-overlapping `[lowerbound, upperbound)`
+filters. Documents outside that range are not read. The actual partition count is the smaller
+of `partition.num` and `upperbound - lowerbound`, which avoids empty integer ranges.
+
+For size windows, the source uses bookmark offsets with a stable sort. A single partition sorts
+by `_id`; range partitions sort by the partition field and `_id`. For efficient range-partitioned
+reads, create an ascending compound index on both fields, for example `{id: 1, _id: 1}`.
+All-window reads stream directly from the MongoDB cursor. The source supports static, bounded
+collections; changes made during a read can cause missing or repeated documents.
+
+The sink uses ordered inserts. It flushes when the batch size is reached and when a window
+finishes. It does not provide upsert, exactly-once delivery, or retries beyond the MongoDB driver
+behavior configured by the connection string. A failed batch can be partially written by MongoDB.
+
+## Type mapping
+
+| GeaFlow type | BSON type |
+| --- | --- |
+| `VARCHAR` | String |
+| `BOOLEAN` | Boolean |
+| `TINYINT`, `SMALLINT`, `INTEGER` | Int32 |
+| `BIGINT` | Int64 |
+| `FLOAT`, `DOUBLE` | Double |
+| `DECIMAL` | Decimal128 |
+| `DATE`, `TIMESTAMP` | Date |
+
+Missing fields and BSON null values are read as null. If `_id` is declared as `VARCHAR`, a BSON
+ObjectId is returned as its hexadecimal string. A string `_id` written by the sink remains a
+BSON string. Arrays and nested documents are not supported.
+
+## Example
+
+```sql
+CREATE TABLE mongo_source (
+  id BIGINT,
+  name VARCHAR,
+  active BOOLEAN
+) WITH (
+  type = 'mongodb',
+  `geaflow.dsl.mongodb.uri` = 'mongodb://localhost:27017',
+  `geaflow.dsl.mongodb.database` = 'geaflow',
+  `geaflow.dsl.mongodb.collection` = 'source_records'
+);
+
+CREATE TABLE mongo_sink (
+  id BIGINT,
+  name VARCHAR,
+  active BOOLEAN
+) WITH (
+  type = 'mongodb',
+  `geaflow.dsl.mongodb.uri` = 'mongodb://localhost:27017',
+  `geaflow.dsl.mongodb.database` = 'geaflow',
+  `geaflow.dsl.mongodb.collection` = 'sink_records',
+  `geaflow.dsl.mongodb.batch.size` = '500'
+);
+
+INSERT INTO mongo_sink
+SELECT id, name, active FROM mongo_source;
+```
diff --git a/docs/docs-en/source/5.application-development/3.connector/index.rst b/docs/docs-en/source/5.application-development/3.connector/index.rst
@@ -16,4 +16,5 @@ Connector
    8.hudi.md
    9.pulsar.md
    10.udc.md
-   11.doris.md
+   11.doris.md
+   12.mongodb.md
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-mongodb/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-mongodb/pom.xml
@@ -0,0 +1,85 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.geaflow</groupId>
+        <artifactId>geaflow-dsl-connector</artifactId>
+        <version>0.8.0-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>geaflow-dsl-connector-mongodb</artifactId>
+
+    <properties>
+        <mongodb-driver.version>4.11.1</mongodb-driver.version>
+        <testcontainers.version>1.19.8</testcontainers.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-connector-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.mongodb</groupId>
+            <artifactId>mongodb-driver-sync</artifactId>
+            <version>${mongodb-driver.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>${testng.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>mongodb</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-failsafe-plugin</artifactId>
+                <version>${maven-surefire.version}</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>integration-test</goal>
+                            <goal>verify</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>
diff --git a/...ector-mongodb/src/main/java/org/apache/geaflow/dsl/connector/mongodb/MongoConfigKeys.java b/...ector-mongodb/src/main/java/org/apache/geaflow/dsl/connector/mongodb/MongoConfigKeys.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.geaflow.dsl.connector.mongodb;
+
+import org.apache.geaflow.common.config.ConfigKey;
+import org.apache.geaflow.common.config.ConfigKeys;
+
+public class MongoConfigKeys {
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_URI = ConfigKeys
+        .key("geaflow.dsl.mongodb.uri")
+        .noDefaultValue()
+        .description("MongoDB connection string.");
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_DATABASE = ConfigKeys
+        .key("geaflow.dsl.mongodb.database")
+        .noDefaultValue()
+        .description("MongoDB database name.");
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_COLLECTION = ConfigKeys
+        .key("geaflow.dsl.mongodb.collection")
+        .noDefaultValue()
+        .description("MongoDB collection name.");
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_BATCH_SIZE = ConfigKeys
+        .key("geaflow.dsl.mongodb.batch.size")
+        .defaultValue(1000)
+        .description("MongoDB sink batch size.");
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_PARTITION_NUM = ConfigKeys
+        .key("geaflow.dsl.mongodb.partition.num")
+        .defaultValue(1)
+        .description("MongoDB source partition number.");
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_PARTITION_FIELD = ConfigKeys
+        .key("geaflow.dsl.mongodb.partition.field")
+        .noDefaultValue()
+        .description("MongoDB source range partition field.");
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_PARTITION_LOWERBOUND = ConfigKeys
+        .key("geaflow.dsl.mongodb.partition.lowerbound")
+        .noDefaultValue()
+        .description("Inclusive lower bound for MongoDB source partitions.");
+
+    public static final ConfigKey GEAFLOW_DSL_MONGODB_PARTITION_UPPERBOUND = ConfigKeys
+        .key("geaflow.dsl.mongodb.partition.upperbound")
+        .noDefaultValue()
+        .description("Exclusive upper bound for MongoDB source partitions.");
+
+    private MongoConfigKeys() {
+    }
+}