Data connectors: shared batched ingest with upsert mode (and fix missing `auto_id` in the MySQL connector)

[rajarshidattapy]

Area: packages/moss-data-connector/*

Problem

1. ingest.py is copied into 9 packages and only supports create-from-scratch

All of these have almost identical src/ingest.py files: _template, sqlite, mysql, mongodb, supabase, dynamodb, huggingface, s3, zeroentropy. Each one does:

docs = list(source)                      # materialise the whole source in memory
...
return await client.create_index(index_name, docs, model_id=model_id)

(e.g. moss-connector-sqlite/src/ingest.py:32-36, moss-connector-mongodb/src/ingest.py:32-36)

This causes three problems:

• Memory: list(source) loads the whole table, collection, or dataset before anything is sent. For huggingface, dynamodb, or a large MongoDB collection, that can mean millions of rows in RAM. The connectors' streaming __iter__ does nothing to help.
• Create-only: ingest() always calls create_index. There is no way to re-sync a source into an existing index (upsert changed rows, add new ones). Users have to delete the index and rebuild it, which leaves search unavailable in the meantime.
• Drift: fixes land in some copies and not others (see Yatharth/samplev2 #2 below).

The S3 connector already shows the right pattern in moss-connector-s3/src/watch.py:

• create_index when the index is missing
• add_docs(..., options=MutationOptions(upsert=True)) for additions and modifications
• delete_docs for removals

This logic is only used by the S3 watcher, and every other connector lacks it.

2. Bug: moss-connector-mysql does not support auto_id

packages/moss-data-connector/README.md documents for all connectors:

Use auto_id=True when your mapper does not have a stable primary key and you want Moss to generate UUID document IDs.

But moss-connector-mysql/src/ingest.py has no auto_id parameter and no _replace_doc_id helper. await ingest(source, ..., auto_id=True) raises TypeError: ingest() got an unexpected keyword argument 'auto_id'. Every other connector supports it (it was added in #172).

Proposed solution

1. Fix MySQL first (small, separate PR): add auto_id: bool = False to match moss-connector-sqlite/src/ingest.py, plus a test in moss-connector-mysql/tests.
2. Batched ingest: consume source in chunks (e.g. batch_size: int = 500):
   • create the index with the first batch
   • send the remaining batches with add_docs(index_name, batch, options=MutationOptions(upsert=True))
   • return an aggregated result (e.g. total docs, list of job IDs)
3. Add a mode parameter:
   • mode="create": current behavior (default, backwards compatible)
   • mode="upsert": if the index exists, upsert into it; otherwise create it
   • optionally mode="replace": upsert, then delete_docs for IDs no longer present in the source. Reuse the diff logic from s3/watch.py.
4. Deduplicate: either move the shared ingest into one small shared module that the connectors depend on, or (if the maintainers prefer each connector to stay standalone) at minimum update _template/src/ingest.py and add a parametrized test that checks every connector's ingest has the same signature, so they can't drift again.
5. Document batch_size / mode in packages/moss-data-connector/README.md and each connector README.

Acceptance criteria

• moss_connector_mysql.ingest(..., auto_id=True) works and is tested.
• Peak memory while ingesting N rows is bounded by batch_size: a test with a generator source of 10k fake docs and a mocked MossClient checks that no single call receives more than batch_size docs.
• mode="upsert" on an existing index calls add_docs(upsert=True) and never create_index / delete_index.
• Default behavior (mode="create", no batch_size) stays compatible for existing callers.
• All connectors expose the same ingest signature.

Overlap check

• [Feat] : Add moss-connector-postgres #168 (postgres), [Feat] : Add moss-connector-snowflake; Snowflake source connector #170 / PR feat: Add moss-connector-snowflake #336 (snowflake), [Feature]: add moss-connector-notion #352 / Connector: Notion #380 / PR feat: add Notion connector #353 (notion) add new connectors. This issue changes the shared ingest path they would copy, so it's worth linking those PRs so they adopt the new _template.
• Good first issue: CSV/JSON bulk-import command for the CLI #379 / PR feat(moss-cli): add mapped CSV/JSON bulk import #392 / PR fix(moss-cli): keep newlines in quoted csv fields on doc import #488 cover bulk import in moss-cli, not the connector packages.
• PR feat: add moss-chunking - pluggable chunking strategies over a shared chunk contract #505 (moss-chunking) splits documents into chunks and doesn't touch ingest or upload.
• Connector: S3 / object storage #381 (S3 connector) is closed. This issue reuses its watch/upsert logic rather than changing it.