From e758e739e8fddf46fb5cf2e73dfe77e88471913e Mon Sep 17 00:00:00 2001 From: Shivam Sharma <192242476+ShivamSharma43@users.noreply.github.com> Date: Tue, 6 Oct 2026 22:17:10 +0530 Subject: [PATCH] fix(es): key existing documents on _id instead of _source.id (#4523) ## Closes https://github.com/elastic/connectors/issues/2776 When the sink loads the documents already in an index, `yield_existing_documents_metadata` keyed each one on `_source.id` and only fell back to `_id` when `id` was missing. Everything that consumes that map treats the key as the Elasticsearch `_id`: `get_docs` and `get_access_control_docs` look up each incoming doc by the `_id` they are about to write, and `enqueue_docs_to_delete` issues bulk deletes by `_id`. The connector itself always writes `id` equal to `_id`, so the two only diverge when something else changes `id`, for example an ingest pipeline. When they do, no incoming doc matches an existing one: every doc is counted as a create, `skip_unchanged_documents` never skips anything, and the deletes at the end of a full sync target `_id` values that don't exist, so docs removed from the source are left in the index. This change keys the map on `_id` and drops `id` from the scan's `_source`, since it is no longer read. The `id` field the connector writes into documents is unchanged. This changes behaviour for indices already written where `id` differs from `_id`. The first full or access control sync after this change will delete documents that were orphaned by the old behaviour, so document counts may drop. Existing docs will be counted as updates rather than creates, and unchanged docs will be skipped when `skip_unchanged_documents` is enabled. Indices where `id` equals `_id`, or where `id` is absent, are unaffected. Tests: `test_management_client.py` gets a case where `_source.id` differs from `_id` and asserts the yielded keys are the `_id`s. `test_sink.py` gets a case that runs `get_docs` through the real `yield_existing_documents_metadata` (with `async_scan` mocked) and asserts the matching doc is queued as an update and the stale doc is deleted by its `_id`. Both fail without the fix. ## Checklists #### Pre-Review Checklist - [x] this PR does NOT contain credentials of any kind, such as API keys or username/passwords (double check `config.yml.example`) - [x] this PR has a meaningful title - [x] this PR links to all relevant github issues that it fixes or partially addresses - [x] Covered the changes with automated tests - [ ] Tested the changes locally (ran `make lint` and `make test`; not tested against a live cluster) - [ ] Added a label for each target release version (example: `v7.13.2`, `v7.14.0`, `v8.0.0`) - [ ] For bugfixes: backport safely to all minor branches still receiving patch releases ## Release Note Fixes syncs failing to delete stale documents, and treating existing documents as new, when an ingest pipeline changes a document's `id` field so that it no longer matches its `_id`. Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 Co-authored-by: Jan-Kazlouski-elastic --- connectors/es/management_client.py | 6 +++--- tests/es/test_management_client.py | 24 ++++++++++++++++++++++ tests/test_sink.py | 32 ++++++++++++++++++++++++++++++ 3 files changed, 59 insertions(+), 3 deletions(-) diff --git a/connectors/es/management_client.py b/connectors/es/management_client.py index 73c45257a..523c487f8 100644 --- a/connectors/es/management_client.py +++ b/connectors/es/management_client.py @@ -283,7 +283,7 @@ async def bulk_insert(self, operations, pipeline): ) async def yield_existing_documents_metadata(self, index): - """Returns an iterator on the `id` and `_timestamp` fields of all documents in an index. + """Returns an iterator on the `_id` and `_timestamp` fields of all documents in an index. WARNING @@ -297,10 +297,10 @@ async def yield_existing_documents_metadata(self, index): return async for doc in async_scan( - client=self.client, index=index, _source=["id", TIMESTAMP_FIELD] + client=self.client, index=index, _source=[TIMESTAMP_FIELD] ): source = doc["_source"] - doc_id = source.get("id", doc["_id"]) + doc_id = doc["_id"] timestamp = source.get(TIMESTAMP_FIELD) yield doc_id, timestamp diff --git a/tests/es/test_management_client.py b/tests/es/test_management_client.py index e42e3e265..34bdd0911 100644 --- a/tests/es/test_management_client.py +++ b/tests/es/test_management_client.py @@ -310,6 +310,30 @@ async def test_yield_existing_documents_metadata_when_index_exists( assert ids == ["1", "2"] + @pytest.mark.asyncio + async def test_yield_existing_documents_metadata_when_source_id_differs_from_id( + self, es_management_client, mock_responses + ): + es_management_client.index_exists = AsyncMock(return_value=True) + + records = [ + {"_id": "1", "_source": {"id": "a", "_timestamp": str(datetime.now())}}, + {"_id": "2", "_source": {"id": "b", "_timestamp": str(datetime.now())}}, + ] + + with mock.patch( + "connectors.es.management_client.async_scan", + return_value=AsyncIterator(records), + ): + ids = [] + async for ( + doc_id, + _, + ) in es_management_client.yield_existing_documents_metadata("something"): + ids.append(doc_id) + + assert ids == ["1", "2"] + @pytest.mark.asyncio async def test_get_connector_secret(self, es_management_client, mock_responses): secret_id = "secret-id" diff --git a/tests/test_sink.py b/tests/test_sink.py index e1c035df5..f69599ddb 100644 --- a/tests/test_sink.py +++ b/tests/test_sink.py @@ -790,6 +790,38 @@ async def test_get_docs( assert queue_called_with_operations(queue, expected_queue_operations) +@mock.patch("connectors.es.management_client.async_scan") +@mock.patch( + "connectors.es.management_client.ESManagementClient.index_exists", + return_value=True, +) +@pytest.mark.asyncio +async def test_get_docs_matches_and_deletes_existing_docs_by_es_id( + index_exists, async_scan +): + # e.g. an ingest pipeline rewrote `id`, so it no longer matches `_id` + async_scan.return_value = AsyncIterator( + [ + {"_id": "1", "_source": {"id": "rewritten-1", "_timestamp": TIMESTAMP}}, + {"_id": "2", "_source": {"id": "rewritten-2", "_timestamp": TIMESTAMP}}, + ] + ) + queue = await queue_mock() + extractor = await setup_extractor(queue) + + await extractor.run( + AsyncIterator([(deepcopy(DOC_ONE), None, "index")]), JobType.FULL + ) + + assert extractor.counters.get(UPDATES_QUEUED) == 1 + assert extractor.counters.get(CREATES_QUEUED) == 0 + assert extractor.counters.get(DELETES_QUEUED) == 1 + assert queue_called_with_operations( + queue, + [index_operation(DOC_ONE), delete_operation(DOC_TWO), end_docs_operation()], + ) + + def test_log_progress_includes_extraction_counters(): logger_mock = Mock() extractor = Extractor(None, Mock(), INDEX, logger_=logger_mock)