RFC: Write Support (INSERT / CREATE TABLE AS) for v1

[stephaniewang526]

The most requested feature for this extension is write support. Some example answers from a user survey:

"Adding Insert, Update, Delete Operation"
"Multiplexed, bidirectional and plate/mDNS quic discovery to other remote mongodb"

This RFC proposes a simple v1 of write support. Feedback welcome on scope, API surface, and priorities.

---

Scope

Supported	Not in v1
INSERT INTO mongo.collection SELECT ...	UPDATE, DELETE via DML
CREATE TABLE mongo.db.new_coll AS SELECT ...	ON CONFLICT / upsert
COPY ... TO (bulk export to MongoDB)	Schema enforcement on write
Batched insertMany under the hood	Transactions / multi-doc atomics
Flatten reversal (column address_city -> nested address.city)	RETURNING clause

The core value prop: users can ETL into MongoDB from any DuckDB source (Parquet, CSV, Postgres, another Mongo collection, etc.) using standard SQL.

---

Architecture (3 pieces)

1. MongoInsertOperator -- custom PhysicalOperator

A new PhysicalOperator subclass that receives DataChunks from DuckDB's pipeline and calls insertMany on the target collection.

Key design decisions:

• Batch size: accumulate ~1000 docs (configurable), then insertMany with ordered: false for throughput.
• Type mapping: reverse of FlattenDocument -- DuckDB values -> BSON. Reuse the column_name_to_mongo_path map to reconstruct nested docs (e.g., column address_city writes to {"address": {"city": ...}}).
• _id handling: if source data has an _id column, pass it through; otherwise let MongoDB auto-generate ObjectIds.

2. Wire into MongoCatalog::PlanInsert / PlanCreateTableAs

Replace the current throw NotImplementedException(...) stubs:

• PlanInsert: create a MongoInsertOperator, connect the child plan as its input. The target collection comes from LogicalInsert::table.
• PlanCreateTableAs: create the collection (it auto-creates on first insert in MongoDB), then behave like PlanInsert.

The catalog already has connection_string and database_name -- just pass them down.

3. DuckDB-value-to-BSON serializer

A new helper (inverse of FlattenDocument):

bsoncxx::document::value ChunkRowToBSON(
    DataChunk &chunk, idx_t row_idx,
    const vector<string> &column_names,
    const vector<LogicalType> &column_types,
    const unordered_map<string, string> &column_name_to_mongo_path);

Type mapping:

DuckDB type	BSON type
INTEGER/BIGINT	int32/int64
DOUBLE/FLOAT	double
VARCHAR	UTF-8 string
BOOLEAN	bool
TIMESTAMP	DateTime (millis)
DATE	DateTime (millis, midnight)
BLOB	Binary
LIST	Array
STRUCT	Subdocument
NULL	null

Nested reconstruction: if column_name_to_mongo_path["address_city"] == "address.city", split on . and build nested BSON subdocuments, merging sibling fields (e.g., address_city + address_zip -> single address subdoc).

---

Execution flow

For INSERT INTO mongo.db.users SELECT ...:

1. DuckDB parses + plans the query, resolves mongo.db.users via MongoCatalog.
2. MongoCatalog::PlanInsert creates MongoInsertOperator(connection_string, "db", "users", column_info).
3. DuckDB pipeline calls MongoInsertOperator::Sink(chunk) for each output chunk.
4. Sink accumulates rows into a vector<bsoncxx::document::value>, flushes every 1000 docs via collection.insert_many(batch).
5. Finalize() flushes remaining docs, returns row count.
6. Invalidate view cache for the target collection (schema may have changed).

---

Open questions for feedback

1. Flatten reversal by default? Should writing column address_city automatically nest it as address.city in MongoDB, or should that be opt-in?
2. Batch size default -- 1000 docs seems reasonable, but should it be tunable via a session setting?
3. Error handling on partial failure -- if a batch of 1000 fails midway (ordered: false), should we report the count of successful inserts and continue, or abort?
4. Priority for v1.1 -- would you rather see UPDATE/DELETE next, or COPY TO bulk export?

Looking forward to your thoughts!