The DAG mw_content_merge_events_to_mw_content_history_daily has failed in the task spark_process_events.
The error log message is:
The ON search condition of the MERGE statement matched a single row from the target table with multiple rows of the source table. This could result in the target row being operated on more than once with an update or delete operation and is not allowed.
This seems to indicate that there are unexpected duplicate rows.
This task is to troubleshoot and fix this.
| Title | Reference | Author | Source Branch | Dest Branch | |
|---|---|---|---|---|---|
| Bump snapshot retention of MWCH to 45 days. | repos/data-engineering/airflow-dags!1829 | xcollazo | keep-more-snapshots | main | |
| Set write.delete.mode=copy-on-write for mediawiki_content_current_v1. | repos/data-engineering/mediawiki-content-pipelines!83 | xcollazo | delete-mode-for-current | main |
Looks like (yet) another instance of T404975: Another instance of duplicate rows on wmf_content.mediawiki_content_history_v1.
Pausing all MW Content pipelines.
Total duplicates:
spark.sql("""
SELECT count(1) as count FROM (
SELECT count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
""").show(300, truncate=False)
[Stage 2:====================================================>(1021 + 3) / 1024]
+-----+
|count|
+-----+
|2457 |
+-----+Distribution:
spark.sql("""
SELECT count(1) as count,
wiki_id
FROM (
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
GROUP BY wiki_id
ORDER BY count DESC
""").show(3000, truncate=False)
+-----+-------------+
|count|wiki_id |
+-----+-------------+
|1759 |enwiki |
|157 |zhwiki |
|150 |viwiki |
|112 |hiwiki |
|27 |fawiki |
|26 |commonswiki |
|21 |eswiki |
|21 |idwiki |
|17 |etwiki |
|15 |hewiki |
|13 |trwiki |
|11 |enwiktionary |
|10 |arwiki |
|9 |metawiki |
|8 |bnwiki |
|8 |azwiki |
|7 |cawiki |
|7 |mediawikiwiki|
|7 |rowiki |
|6 |itwiki |
|5 |pawiki |
|5 |kowiki |
|5 |ptwiki |
|4 |frwiki |
|4 |brwiki |
|3 |dewiki |
|3 |srwiki |
|3 |ruwiki |
|3 |trwikibooks |
|2 |pswiki |
|2 |ukwiki |
|2 |elwiki |
|2 |nlwiki |
|2 |uzwiki |
|2 |plwiki |
|2 |cswiki |
|1 |svwiki |
|1 |nlwikisource |
|1 |cebwiki |
|1 |iswiki |
|1 |skwiki |
|1 |itwikiquote |
|1 |mswiki |
|1 |ruwikisource |
|1 |kuwiki |
|1 |siwiktionary |
|1 |arwikisource |
|1 |simplewiki |
|1 |labswiki |
|1 |itwikibooks |
|1 |kuwiktionary |
|1 |ruwiktionary |
|1 |kowikisource |
+-----+-------------+Total duplicates and distribution are similar, but not same as in T404975. Thus likely same root cause but is not a repeat of the same rows...
Following procedure from T404975#11197939:
Yarn job: https://yarn.wikimedia.org/proxy/application_1758550516505_1543350/
$ hostname -f an-launcher1003.eqiad.wmnet sudo -u analytics bash screen -S xcollazo_mw_content_history_fix_T410431 spark3-sql \ --master yarn \ --conf spark.driver.maxResultSize=8G \ --conf spark.dynamicAllocation.maxExecutors=200 \ --conf spark.sql.shuffle.partitions=2048 \ --conf spark.sql.iceberg.locality.enabled=true \ --conf spark.reducer.maxReqsInFlight=1 \ --conf spark.shuffle.io.retryWait=180s \ --conf spark.shuffle.io.maxRetries=10 \ --executor-cores 2 \ --executor-memory 16G \ --driver-cores 4 \ --driver-memory 16G \ --name xcollazo_mw_content_history_fix_T410431 DELETE FROM wmf_content.mediawiki_content_history_v1 WHERE ( (wiki_id = 'arwiki' AND revision_id = 6834505 AND page_id = 356359 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 8327979 AND page_id = 117901 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 17267940 AND page_id = 2396830 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 49958410 AND page_id = 2432692 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) ... Response code Time taken: 1762.235 seconds
Verification:
# after fix
spark.sql("""
SELECT count(1) as count FROM (
SELECT count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
""").show(300, truncate=False)wmf_content.mediawiki_content_current_v1 also has duplicates:
spark.sql("""
SELECT count(1) as count FROM (
SELECT count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_current_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
""").show(300, truncate=False)
+-----+
|count|
+-----+
|19 |
+-----+Thus, following T404975#11198200, reusing same spark-sql session as above:
# we will run the fix on Iceberg 1.2.1, and that version can't do merge-on-read
ALTER TABLE wmf_content.mediawiki_content_current_v1 SET TBLPROPERTIES (
'write.delete.mode'='copy-on-write'
);
DELETE
FROM wmf_content.mediawiki_content_current_v1
WHERE (
(wiki_id = 'arwiki' AND revision_id = 71353192 AND page_id = 10058034 AND row_update_dt = '2025-07-11 20:54:55.691437' )
OR
(wiki_id = 'arwiki' AND revision_id = 71610582 AND page_id = 10091488 AND row_update_dt = '2025-08-06 02:22:44.092936' )
OR
(wiki_id = 'arwiki' AND revision_id = 71716699 AND page_id = 10112415 AND row_update_dt = '2025-08-13 04:49:20.423400' )
OR
(wiki_id = 'arwiki' AND revision_id = 72125788 AND page_id = 10163173 AND row_update_dt = '2025-09-26 12:00:37.167298' )
OR
(wiki_id = 'arwiki' AND revision_id = 72318222 AND page_id = 10200301 AND row_update_dt = '2025-10-24 20:59:42.681229' )
OR
(wiki_id = 'bnwiki' AND revision_id = 3956745 AND page_id = 633369 AND row_update_dt = '2024-07-01 00:00:00' )
OR
(wiki_id = 'enwiki' AND revision_id = 1053544024 AND page_id = 5486114 AND row_update_dt = '2024-07-01 00:00:00' )
OR
(wiki_id = 'enwiki' AND revision_id = 1306068758 AND page_id = 3734905 AND row_update_dt = '2025-08-15 19:17:08.266261' )
OR
(wiki_id = 'enwiki' AND revision_id = 1317587997 AND page_id = 70091669 AND row_update_dt = '2025-10-18 21:42:00.296047' )
OR
(wiki_id = 'enwiktionary' AND revision_id = 80863995 AND page_id = 10384038 AND row_update_dt = '2024-07-24 13:21:39.713362' )
OR
(wiki_id = 'frwiki' AND revision_id = 229962619 AND page_id = 17091357 AND row_update_dt = '2025-10-22 09:43:00.684007' )
OR
(wiki_id = 'frwiki' AND revision_id = 230247310 AND page_id = 17053203 AND row_update_dt = '2025-10-31 21:27:48.676236' )
OR
(wiki_id = 'kkwiki' AND revision_id = 3505859 AND page_id = 765216 AND row_update_dt = '2025-09-23 18:16:33.160759' )
OR
(wiki_id = 'kowiki' AND revision_id = 40750918 AND page_id = 4096003 AND row_update_dt = '2025-10-16 15:45:44.247884' )
OR
(wiki_id = 'nowiki' AND revision_id = 23863328 AND page_id = 58001 AND row_update_dt = '2024-07-01 00:00:00' )
OR
(wiki_id = 'ptwiki' AND revision_id = 44542675 AND page_id = 4618776 AND row_update_dt = '2024-07-01 00:00:00' )
OR
(wiki_id = 'tawiki' AND revision_id = 4378006 AND page_id = 591429 AND row_update_dt = '2025-10-28 20:16:14.305186' )
OR
(wiki_id = 'zhwiki' AND revision_id = 89607054 AND page_id = 9302997 AND row_update_dt = '2025-10-20 23:54:24.283751' )
OR
(wiki_id = 'zhwiki' AND revision_id = 89741238 AND page_id = 6649301 AND row_update_dt = '2025-10-30 12:40:31.302326' )
)
Response code
Time taken: 91.648 secondsConsidering this issue continues to happen, I will leave 'write.delete.mode'='merge-on-read', at least until we root cause.
xcollazo opened https://gitlab.wikimedia.org/repos/data-engineering/dumps/mediawiki-content-dump/-/merge_requests/83
Set write.delete.mode=copy-on-write for mediawiki_content_current_v1.
xcollazo merged https://gitlab.wikimedia.org/repos/data-engineering/dumps/mediawiki-content-dump/-/merge_requests/83
Set write.delete.mode=copy-on-write for mediawiki_content_current_v1.
xcollazo opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1829
Bump snapshot retention of MWCH to 45 days.
xcollazo merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1829
Bump snapshot retention of MWCH to 45 days.
Current situation after the monthly reconciliation (query executed on 2025-12-08):
spark.sql("""
SELECT count(*) as total_duplicates
FROM (
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id ,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+-----------------+ |total_duplicates | +-----------------+ |2929 | +-----------------+
However, when including the page_id the count of duplicates lowers to 0:
spark.sql("""
SELECT count(*) as total_duplicates
FROM (
SELECT
count(1) as count,
wiki_id,
page_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id, page_id,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+-----------------+ |total_duplicates | +-----------------+ |0 | +-----------------+
Checking the distribution of duplicates by page_change_kind and wiki_id:
spark.sql("""
with duplicated as (
SELECT
wiki_id,
revision_id
FROM (
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
GROUP BY wiki_id, revision_id
),
joined as (
select
distinct
enr.wiki_id,
enr.revision.rev_id as revision,
enr.page_change_kind
from duplicated dup
join event.mediawiki_content_history_reconcile_enriched_v1 enr
on enr.wiki_id = dup.wiki_id
and enr.revision.rev_id = dup.revision_id
)
select wiki_id, page_change_kind, count(revision)
from joined
group by wiki_id, page_change_kind
""").show(200, truncate=False)+-------------+----------------+---------------+ |wiki_id |page_change_kind|count(revision)| +-------------+----------------+---------------+ |enwiki |move |2215 | |zhwiki |move |144 | |arwiki |move |10 | |viwiki |move |203 | |eswiki |move |20 | |pawiki |move |5 | |hiwiki |move |109 | |nlwiki |move |2 | |rowiki |move |7 | |bnwiki |move |6 | |idwiki |move |17 | |commonswiki |move |22 | |mediawikiwiki|move |6 | |itwikiquote |move |1 | |trwikibooks |move |3 | |etwiki |move |18 | |metawiki |move |8 | |trwiki |move |8 | |ruwikisource |move |1 | |dewiki |move |4 | |ruwiki |move |3 | |enwiktionary |move |11 | |frwiki |move |3 | |hewiki |move |16 | |itwiki |move |5 | |uzwiki |move |2 | |ptwiki |move |5 | |fawiki |move |26 | |azwiki |move |9 | |elwiki |move |2 | |itwikibooks |move |1 | |kowiki |move |5 | |srwiki |move |3 | |brwiki |move |3 | |cawiki |move |6 | |commonswiki |edit |1 | |svwiki |move |1 | |mswiki |move |1 | |iswiki |move |1 | |skwiki |move |1 | |ukwiki |move |2 | |kowikisource |move |1 | |nlwikisource |move |1 | |kuwiktionary |move |1 | |cswiki |move |2 | |pswiki |move |1 | |labswiki |move |1 | |ruwiktionary |move |1 | |kuwiki |move |1 | |lawiki |move |1 | |slwiki |move |1 | |cebwiki |move |1 | |plwiki |move |1 | |simplewiki |move |1 | +-------------+----------------+---------------+
We can see that the majority of page_change_kind are move. Let's check first the odd one out for the revision_id 1118294298 in commonswiki corresponding both to an edit and a move:
spark.sql("""select dt,page.page_id,revision.rev_id, page_change_kind, year,month,day,hour from event.mediawiki_content_history_reconcile_enriched_v1
where wiki_id= 'commonswiki' and revision.rev_id='1118294298'""").show(200, truncate=False)+--------------------+---------+----------+----------------+----+-----+---+----+ |dt |page_id |rev_id |page_change_kind|year|month|day|hour| +--------------------+---------+----------+----------------+----+-----+---+----+ |2025-11-19T07:26:54Z|178775087|1118294298|move |2025|12 |2 |13 | |2025-11-19T07:26:54Z|100282687|1118294298|edit |2025|11 |20 |7 | +--------------------+---------+----------+----------------+----+-----+---+----+
spark.sql( "select wiki_id, page_id, revision_id,revision_dt, row_content_update_dt,row_visibility_update_dt, row_move_update_dt from wmf_content.mediawiki_content_history_v1 where wiki_id = 'commonswiki' and revision_id = 1118294298").show(truncate=False)
+-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+ |wiki_id |page_id |revision_id|revision_dt |row_content_update_dt |row_visibility_update_dt |row_move_update_dt | +-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+ |commonswiki|178775087|1118294298 |2025-11-19 07:26:54|2025-12-02 13:18:42.9273 |2025-12-02 13:18:42.9273 |2025-12-02 13:18:42.9273 | |commonswiki|100282687|1118294298 |2025-11-19 07:26:54|2025-11-20 07:23:16.356547|2025-11-20 07:23:16.356547|2025-11-20 07:23:16.356547| +-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+
Is clear that the first event we received for this revision_id was an edit and then a move.
Let's finally check the mwch table for before and after the recent reconciliation:
Before the reconciliation:
spark.sql("""
select wiki_id, page_id, revision_id,revision_dt, row_content_update_dt,row_visibility_update_dt, row_move_update_dt
from wmf_content.mediawiki_content_history_v1 TIMESTAMP AS OF '2025-12-03'
where wiki_id= 'commonswiki'
and revision_id = 1118294298
order by revision_id, row_content_update_dt desc
""").show(3000, truncate=False)+-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+ |wiki_id |page_id |revision_id|revision_dt |row_content_update_dt |row_visibility_update_dt |row_move_update_dt | +-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+ |commonswiki|100282687|1118294298 |2025-11-19 07:26:54|2025-11-20 07:23:16.356547|2025-11-20 07:23:16.356547|2025-11-20 07:23:16.356547| +-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+
After the reconciliation:
spark.sql("""
select wiki_id, page_id, revision_id,revision_dt, row_content_update_dt,row_visibility_update_dt, row_move_update_dt
from wmf_content.mediawiki_content_history_v1 TIMESTAMP AS OF '2025-12-04'
where wiki_id= 'commonswiki'
and revision_id = 1118294298
order by revision_id, row_content_update_dt desc
""").show(3000, truncate=False)+-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+ |wiki_id |page_id |revision_id|revision_dt |row_content_update_dt |row_visibility_update_dt |row_move_update_dt | +-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+ |commonswiki|178775087|1118294298 |2025-11-19 07:26:54|2025-12-02 13:18:42.9273 |2025-12-02 13:18:42.9273 |2025-12-02 13:18:42.9273 | |commonswiki|100282687|1118294298 |2025-11-19 07:26:54|2025-11-20 07:23:16.356547|2025-11-20 07:23:16.356547|2025-11-20 07:23:16.356547| +-----------+---------+-----------+-------------------+--------------------------+--------------------------+--------------------------+
The obtained results of the reconciliation for the given revision_id is to add the row instead of updating the existing one.
I have run the mw_content_merge_events_to_mw_content_history_daily DAG on a set of test table containing the following data:
spark.sql("""insert into apizzata.mediawiki_page_content_change_v1 select *
from event.mediawiki_page_content_change_v1 where wiki_id= 'commonswiki' and year= 2025 and month in (6,7,8,9,10,11,12)""")spark.sql("""insert into apizzata.mediawiki_revision_visibility_change select *
from event.mediawiki_revision_visibility_change where wiki_id= 'commonswiki' and year= 2025 and month in (6,7,8,9,10,11,12)""")spark.sql("""insert into apizzata.mediawiki_content_history_v1 select * from wmf_content.mediawiki_content_history_v1 timestamp as of '2025-12-02' where wiki_id='commonswiki' and revision_dt >='2024-10-01' order by wiki_id, page_id, revision_dt """)By checking the snapshots on the target table apizzata.mediawiki_content_history_v1 we have:
spark.sql("select * from apizzata.mediawiki_content_history_v1.snapshots order by committed_at desc").show()+--------------------+-------------------+-------------------+---------+--------------------+--------------------+
| committed_at| snapshot_id| parent_id|operation| manifest_list| summary|
+--------------------+-------------------+-------------------+---------+--------------------+--------------------+
|2025-12-08 15:00:...|1796286285245367197|5551823393169662060|overwrite|hdfs://analytics-...|{spark.app.id -> ...|
|2025-12-08 14:59:...|5551823393169662060|6321057030902882321|overwrite|hdfs://analytics-...|{spark.app.id -> ...|
|2025-12-08 14:57:...|6321057030902882321|2882563131767895621|overwrite|hdfs://analytics-...|{spark.app.id -> ...|
|2025-12-08 14:53:...|2882563131767895621|1216507440369392920|overwrite|hdfs://analytics-...|{spark.app.id -> ...|
|2025-12-08 14:45:...|1216507440369392920|1879494188702815704|overwrite|hdfs://analytics-...|{spark.app.id -> ...|
|2025-12-08 14:42:...|1879494188702815704|1298415091598223708|overwrite|hdfs://analytics-...|{spark.app.id -> ...|
|2025-12-08 14:41:...|1298415091598223708|4594911566527932515| append|hdfs://analytics-...|{spark.app.id -> ...|
|2025-12-08 14:25:...|4594911566527932515| null| append|hdfs://analytics-...|{spark.app.id -> ...|
+--------------------+-------------------+-------------------+---------+--------------------+--------------------+After analyzing each state of the table it is possible to find the duplication of the revision_id 1118294298 in the snapshot 6321057030902882321:
spark.sql("""select page_id from apizzata.mediawiki_content_history_v1 version as of 6321057030902882321 where wiki_id= 'commonswiki' and revision_id=1118294298""" ).show(truncate=False)+---------+ |page_id | +---------+ |178775087| |100282687| +---------+
Matching the snapshot time with the operations performed by the mw_content_merge_events_to_mw_content_history_daily DAG we can pinpoint the moment the duplication occurred: it is in the process_revisions function for the mediawiki_content_history_reconcile_enriched_v1 table.
Here is the merge query logged by the application application_1764064841637_358710:
WITH deduplicated_mediawiki_page_content_change AS (
-- we want to ingest the latest state of a (wiki_id, rev_id) pair
-- so below we pick latest event based on dt (event timestamp).
-- if there are multiple events with same dt, pick highest changelog_priority
SELECT * FROM (
SELECT
*,
row_number() over ( PARTITION BY wiki_id, revision.rev_id ORDER BY to_timestamp(dt) DESC, changelog_priority ASC ) AS row_num
FROM (
SELECT
*,
CASE
WHEN changelog_kind = 'update' THEN 1
WHEN changelog_kind = 'insert' THEN 2
END AS changelog_priority
FROM apizzata.mediawiki_content_history_reconcile_enriched_v1
WHERE year=2025
AND month=12
AND day=2
-- a move creates a new revision that needs to be ingested
-- applying the move to old revisions will be done separately for perf reasons.
AND page_change_kind IN ('create', 'edit', 'move')
AND wiki_id IN ('commonswiki')
)
)
WHERE row_num = 1
)
MERGE INTO apizzata.mediawiki_content_history_v1 t
USING (
SELECT
page.page_id AS s_page_id,
page.namespace_id AS s_page_namespace_id,
page.page_title AS s_page_title,
page.redirect_page_link.page_title AS s_page_redirect_target,
performer.user_id AS s_user_id,
performer.user_central_id AS s_user_central_id,
performer.user_text AS s_user_text,
revision.is_editor_visible AS s_user_is_visible,
revision.rev_id AS s_revision_id,
revision.rev_parent_id AS s_revision_parent_id,
to_timestamp(revision.rev_dt) AS s_revision_dt,
revision.is_minor_edit AS s_revision_is_minor_edit,
revision.comment AS s_revision_comment,
revision.is_comment_visible AS s_revision_comment_is_visible,
revision.rev_size AS s_revision_size,
transform_values(revision.content_slots,
(k, v) -> (
v.content_body,
v.content_format,
v.content_model,
v.content_sha1,
v.content_size,
v.origin_rev_id
)
) AS s_revision_content_slots,
revision.is_content_visible AS s_revision_content_is_visible,
wiki_id AS s_wiki_id,
-- fields that help us control the ingestion
changelog_kind AS s_changelog_kind,
to_timestamp(meta.dt) AS s_meta_dt,
meta.id AS s_meta_id
FROM deduplicated_mediawiki_page_content_change s
)
ON s_wiki_id = t.wiki_id
AND s_revision_id = t.revision_id
-- pushdown wiki_ids that are changing
-- to limit how much data we effectively read
AND (
(t.wiki_id = 'commonswiki' AND t.page_id IN (24389, ... 156909011,... 178774051, 178775087, 178828381 ... 179299437))
)
WHEN MATCHED AND s_meta_dt >= t.row_content_update_dt THEN
UPDATE SET
t.page_id = s_page_id,
t.page_namespace_id = s_page_namespace_id,
t.page_title = s_page_title,
t.page_redirect_target = s_page_redirect_target,
t.user_id = s_user_id,
t.user_central_id = s_user_central_id,
t.user_text = s_user_text,
t.user_is_visible = s_user_is_visible,
t.revision_id = s_revision_id,
t.revision_parent_id = s_revision_parent_id,
t.revision_dt = s_revision_dt,
t.revision_is_minor_edit = s_revision_is_minor_edit,
t.revision_comment = s_revision_comment,
t.revision_comment_is_visible = s_revision_comment_is_visible,
t.revision_size = s_revision_size,
t.revision_content_slots = s_revision_content_slots,
t.revision_content_is_visible = s_revision_content_is_visible,
t.wiki_id = s_wiki_id,
t.row_content_update_dt = s_meta_dt
WHEN NOT MATCHED THEN
INSERT (
page_id,
page_namespace_id,
page_title,
page_redirect_target,
user_id,
user_central_id,
user_text,
user_is_visible,
revision_id,
revision_parent_id,
revision_dt,
revision_is_minor_edit,
revision_comment,
revision_comment_is_visible,
revision_size,
revision_content_slots,
revision_content_is_visible,
wiki_id,
row_content_update_dt,
row_visibility_update_dt,
row_move_update_dt
) VALUES (
s_page_id,
s_page_namespace_id,
s_page_title,
s_page_redirect_target,
s_user_id,
s_user_central_id,
s_user_text,
s_user_is_visible,
s_revision_id,
s_revision_parent_id,
s_revision_dt,
s_revision_is_minor_edit,
s_revision_comment,
s_revision_comment_is_visible,
s_revision_size,
s_revision_content_slots,
s_revision_content_is_visible,
s_wiki_id,
s_meta_dt,
s_meta_dt,
s_meta_dt
)The optimization_predicates computed with the 'set_of_page_ids' pushdown_strategy lists the page_id 178775087 and since it will not match with the page_id 100282687 already stored in the table for the revision_id 1118294298, the insert clause will be performed.
This causes the duplication of the revision_id 1118294298.
Let's double check with another revision_id: 977913129.
spark.sql("""select page_id from apizzata.mediawiki_content_history_v1 version as of 6321057030902882321 where wiki_id= 'commonswiki' and revision_id=977913129""" ).show(truncate=False)+---------+ |page_id | +---------+ |156909011| |157388601| +---------+
And the situation is confirmed to be the same. The duplicated page_id (156909011) is infact in the optimization_predicates as the previous case.
Same thing goes for all the other revisions_id with duplicates for the commonswiki wiki.
Remaining questions I have that unfortunately I miss the lore are:
The optimization_predicates computed with the 'set_of_page_ids' pushdown_strategy lists the page_id 178775087 and since it will not match with the page_id 100282687 already stored in the table for the revision_id 1118294298, the insert clause will be performed.
This causes the duplication of the revision_id 1118294298.
Awesome finding @APizzata-WMF !!
I understand that the revision_id is unique in the system, but how come we received from our sources different page_id for the same revision_id? Is there any process that can create this situation?
Unfortunately yes. From https://www.mediawiki.org/wiki/Manual:Page_table#page_id:
Uniquely identifying primary key. This value is preserved across edits and renames.
Page IDs do not change when pages are moved, but they may change when pages are deleted and then restored. As of MediaWiki 1.27, the historical page ID persists in the archive table, and restored pages attempt to reclaim their old page ID.
The problem seems to be in the optimization_predicates but I feel is more deeply connected by the join strategy based on the page_id. If we know that the revision_id is unique, why are we joining on the page_id? Are we scared of the volume of records?
Right, we do join primarily on wiki_id and revision_id, but for performance reasons, we also send in optimization predicates:
...
ON s_wiki_id = t.wiki_id
AND s_revision_id = t.revision_id
-- pushdown wiki_ids that are changing
-- to limit how much data we effectively read
AND (
{optimization_predicates}
)
...We do this because otherwise we would be doing a full table scan each time we MERGE, and the size of that table is not pretty:
spark.sql("""
SELECT SUM(file_size_in_bytes)/1024/1024/1024/1024 as file_size_in_teras
FROM wmf_content.mediawiki_content_history_v1.files
""").show(truncate=False)
[Stage 76:===================> (1 + 1) / 3]
+------------------+
|file_size_in_teras|
+------------------+
|2.123862072910015 |
+------------------+Because the table is WRITE ORDERED BY wiki_id, page_id, revision_dt, pushing down the set of page_ids makes us read way, way less data, as we know from the Iceberg manifest whether any file will include a specific page_id or not.
Anyhow, clearly this is not working as intended because we get the perf benefit but we have duplicate rows.
What would be the next steps to fix this issue?
Let's meet and brainstorm. There should be a happy medium here between perf and correctness.
After a discussion with @xcollazo we realised that the problem is connected to a combination of undelete and delete events.
spark.sql("""
select
revision.rev_dt,
day,month,year,
page.page_id,
revision.rev_id,
page_change_kind
from
event.mediawiki_page_content_change_v1
where wiki_id= 'commonswiki'
and page.page_id in(178775087,100282687)
order by revision.rev_dt asc""").show()+--------------------+---+-----+----+---------+----------+----------------+ | rev_dt|day|month|year| page_id| rev_id|page_change_kind| +--------------------+---+-----+----+---------+----------+----------------+ |2025-11-19T07:26:37Z| 20| 11|2025|100282687|1118294200| undelete| |2025-11-19T07:26:37Z| 19| 11|2025|100282687|1118294200| edit| |2025-11-20T07:39:21Z| 20| 11|2025|100282687|1118806092| edit| |2025-11-20T07:39:21Z| 20| 11|2025|100282687|1118806092| delete| |2025-11-20T07:39:21Z| 20| 11|2025|178775087|1118806092| undelete| |2025-11-20T21:16:32Z| 20| 11|2025|100282687|1119085982| move| |2025-11-20T21:17:30Z| 20| 11|2025|100282687|1119086355| edit| |2025-11-20T21:34:41Z| 20| 11|2025|178775087|1119092139| edit| |2025-12-05T08:38:02Z| 5| 12|2025|100282687|1125950709| edit| |2025-12-05T08:38:36Z| 5| 12|2025|178775087|1125950925| edit| +--------------------+---+-----+----+---------+----------+----------------+
The page_id 100282687 is the result of an undelete, same for 178775087.
spark.sql("""
select
revision.rev_dt,
day,month,year,
page.page_id,
revision.rev_id,
page_change_kind
from
event.mediawiki_content_history_reconcile_enriched_v1
where wiki_id= 'commonswiki'
and page.page_id in(178775087,100282687)
and year= 2025 and month in (8,9,10,11,12)
order by revision.rev_dt asc""").show(truncate=False)+--------------------+---+-----+----+---------+----------+----------------+ |rev_dt |day|month|year|page_id |rev_id |page_change_kind| +--------------------+---+-----+----+---------+----------+----------------+ |2025-11-19T07:26:54Z|12 |12 |2025|178775087|1118294298|move | |2025-11-19T07:26:54Z|2 |12 |2025|178775087|1118294298|move | |2025-11-19T07:26:54Z|20 |11 |2025|100282687|1118294298|edit | |2025-11-19T07:26:54Z|11 |12 |2025|178775087|1118294298|move | |2025-11-20T07:39:21Z|21 |11 |2025|178775087|1118806092|edit | +--------------------+---+-----+----+---------+----------+----------------+
The 1118294298 revision that is duplicated comes from the event,mediawiki_content_history_reconcile_enriched_v1 table and the process that generates the duplication is described in the previous comment.
However, the 100282687 should be deleted as for what is stated in the event.mediawiki_page_content_change_v1. Should we implement a table that stores page deletion and use it to check instead of the table that the process_event pipeline is running upon?
Eg. we would create a page_delete table and in the process_page_deletes of the process_event pipeline the source table would be page_delete instead of event.mediawiki_page_content_change_v1 or event.mediawiki_content_history_reconcile_enriched_v1.
In this case knowing that 100282687 is deleted would not generete the duplicate.
If this idea would not work we would have to decide how to deduplicate the records:
Option 1:
We can, at merging time, enable a check that would identify the duplication and delete both records.
This way the table will have no duplicates, but also no record regarding the revision_id.
Then, during reconciliation (both daily or monthly) the process will pull from the MariaDB replica which one is correct.
This option would not write any wrong record, but will have missing data up until the reconciled events are ingested.
Option 2:
At reconciliation time, we would check the MWCH table for duplicate entries. If any are found, we would query MariaDB directly to determine which page_id is connected to the duplicated revision_id. This way we would connect straight to the source of truth and will delete only the wrong record.
This option would write wrong records and then at reconciliation time would delete any wrong one.
Finally, there is also a more radical option:
Restructuring the pipeline to take in account daily copies of the interested MariaDB tables and eventually replacing the monthly Sqoops. This way every day we would have correct informations of what MediaWiki can see and we should be protected from dupliaction or other edge cases.
Then, during reconciliation (both daily or monthly) the process will pull from the MariaDB replica which one is correct.
I was discussing this issue with @JAllemandou, and he mentioned that these rows could very well be coming from the logging table. See discussion in an MR comment thread here.
Run a deduplication like what was done in https://phabricator.wikimedia.org/T404975#11197939 :
duplicate situation for mediawiki_content_history_v1:
spark.sql("""
SELECT count(*) as total_duplicate
FROM (
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id ,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+---------------+ |total_duplicate| +---------------+ |2934 | +---------------+
ran:
hostname -f an-launcher1003.eqiad.wmnet sudo -u analytics bash screen -S apizzata-deduplication spark3-sql \ --master yarn \ --conf spark.driver.maxResultSize=8G \ --conf spark.dynamicAllocation.maxExecutors=200 \ --conf spark.sql.shuffle.partitions=2048 \ --conf spark.sql.iceberg.locality.enabled=true \ --conf spark.reducer.maxReqsInFlight=1 \ --conf spark.shuffle.io.retryWait=180s \ --conf spark.shuffle.io.maxRetries=10 \ --executor-cores 2 \ --executor-memory 16G \ --driver-cores 4 \ --driver-memory 16G \ --name apizzata-deduplication DELETE FROM wmf_content.mediawiki_content_history_v1 WHERE ( (wiki_id = 'arwiki' AND revision_id = 1831134 AND page_id = 356359 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 8298356 AND page_id = 117901 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 17267938 AND page_id = 2396830 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 45598368 AND page_id = 2432692 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR ... (wiki_id = 'zhwiki' AND revision_id = 89527393 AND page_id = 5982005 AND row_content_update_dt = '2025-10-14 15:44:00.685239' AND row_visibility_update_dt = '2025-10-14 15:44:00.685239' AND row_move_update_dt = '2025-10-15 01:58:08.099248' ) ) Response code Time taken: 1661.816 seconds
and the results are:
+---------------+ |total_duplicate| +---------------+ |0 | +---------------+
For mediawiki_content_current_v1 with the same spark session and screen:
spark.sql("""
SELECT count(*) as total_duplicate
FROM (
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_current_v1
GROUP BY wiki_id ,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+---------------+ |total_duplicate| +---------------+ |4 | +---------------+
ran:
DELETE FROM wmf_content.mediawiki_content_current_v1 WHERE ( (wiki_id = 'enwiki' AND revision_id = 1286074323 AND page_id = 72163467 AND row_update_dt = '2025-04-17 15:39:57.286525' ) OR (wiki_id = 'enwiki' AND revision_id = 1322644114 AND page_id = 36120546 AND row_update_dt = '2025-11-17 09:57:58.250384' ) OR (wiki_id = 'enwiki' AND revision_id = 1325682183 AND page_id = 54326451 AND row_update_dt = '2025-12-04 15:15:00.691865' ) OR (wiki_id = 'etwiki' AND revision_id = 6946668 AND page_id = 698990 AND row_update_dt = '2025-08-06 09:44:20.186341' ) ); Response code Time taken: 74.323 seconds
and the results are:
+---------------+ |total_duplicate| +---------------+ |0 | +---------------+
For mediawiki_revision_history_v1 with the same spark session and screen:
spark.sql("""
SELECT count(*) as total_duplicate
FROM (
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_revision_history_v1
GROUP BY wiki_id ,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+---------------+ |total_duplicate| +---------------+ |2917 | +---------------+
ran:
DELETE FROM wmf_content.mediawiki_revision_history_v1 WHERE ( (wiki_id = 'arwiki' AND revision_id = 1831134 AND page_id = 356359 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 8298356 AND page_id = 117901 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR (wiki_id = 'arwiki' AND revision_id = 17267938 AND page_id = 2396830 AND row_content_update_dt = '2024-07-01 00:00:00' AND row_visibility_update_dt = '2024-07-01 00:00:00' AND row_move_update_dt IS NULL ) OR ... (wiki_id = 'zhwiki' AND revision_id = 89527393 AND page_id = 5982005 AND row_content_update_dt = '2025-10-14 15:44:00.685239' AND row_visibility_update_dt = '2025-10-14 15:44:00.685239' AND row_move_update_dt = '2025-10-15 01:58:08.099248' ) ); Response code Time taken: 512.137 seconds
and the results are:
+---------------+ |total_duplicate| +---------------+ |0 | +---------------+
Regarding this:
I was discussing this issue with @JAllemandou, and he mentioned that these rows could very well be coming from the logging table. See discussion in an MR comment thread here.
isn't it related to what I was saying regarding the deletes? I feel situation for the pages (100282687, 178775087) is due to a concurrent arrival of edit and delete events.
@MGerlach identified another repro of duplicates, but this time found via page_title:
I am observing that some page titles appear multiple times in wmf_content.mediawiki_content_current_v1 . Sharing here to check if this is a bug or there might be an explanation.
Query
df_n = ( spark.read.table("wmf_content.mediawiki_content_current_v1") .where(F.col("page_namespace_id") == 0) .where(F.col("wiki_id") == "enwiki") .withColumn("wikitext", F.col("revision_content_slots")["main"]["content_body"]) #filter out redirects https://phabricator.wikimedia.org/T400632 .filter(~F.col("wikitext").contains("#REDIRECT")) .filter(F.col("page_redirect_target").isNull() | (F.col("page_redirect_target") == "")) # page title formatting inconsistency https://phabricator.wikimedia.org/T410405 .withColumn("page_title",F.regexp_replace(F.col("page_title")," ","_")) # count how often each page title occurs .withColumn("n",F.count(F.col('page_title')).over(w)) .where(F.col("n")>1).orderBy(["n","page_title"],ascending=False) )This affects 934 different articles in enwiki.
df_n.select("page_title").distinct().count() Inspecting some examples, it seems that some of these articles have been deleted and cant be accessed anymore on enwiki (e.g. VTrader) or have been proposed for deletion (Naufal Ahmad). +-------+-----------------+--------+--------------------+--------------------+-----------+-------------------+--------------------+ |wiki_id|page_namespace_id| page_id| page_title|page_redirect_target|revision_id| revision_dt| wikitext| +-------+-----------------+--------+--------------------+--------------------+-----------+-------------------+--------------------+ | enwiki| 0|80423618| Naufal_Ahmad| null| 1300148517|2025-07-12 14:46:24|'''Naufal Ahmad''...| | enwiki| 0|80195213| Naufal_Ahmad| null| 1295567855|2025-06-14 15:13:04|{{db-repost|xfd=W...| | enwiki| 0|81810943| Naufal_Ahmad| null| 1328454423|2025-12-19 23:57:34|{{Short descripti...| | enwiki| 0|80245788| Naufal_Ahmad| null| 1296405319|2025-06-19 20:35:03| [[File:9aufal ah...| | enwiki| 0|80773406|Gurdwara_Bibi_Vee...| null| 1306534708|2025-08-18 07:48:51|{{Short descripti...| | enwiki| 0|80343653|Gurdwara_Bibi_Vee...| null| 1298579466|2025-07-03 10:24:14|{{one source|date...| | enwiki| 0|79501358|Gurdwara_Bibi_Vee...| null| 1281406600|2025-03-20 05:01:51|{{short descripti...| | enwiki| 0|80043147|Gurdwara_Bibi_Vee...| null| 1293253911|2025-05-31 17:10:40|{{db-banned|user=...| | enwiki| 0|79854181| Don't_Click_Play| null| 1327737045|2025-12-15 22:44:53|{{Use mdy dates|d...| | enwiki| 0|80089150| Don't_Click_Play| null| 1294430095|2025-06-07 17:39:00|{{Db-afc-move|Dra...| | enwiki| 0|80164169| Don't_Click_Play| null| 1294531254|2025-06-08 07:47:36|#redirect [[Ava M...| | enwiki| 0|80009105| Don't_Click_Play| null| 1291718103|2025-05-22 23:29:52|{{Infobox album |...| | enwiki| 0|79883247|Zubair_Ahmad_Qura...| null| 1289958392|2025-05-11 22:42:28|{{db-person|help=...| | enwiki| 0|80315813|Zubair_Ahmad_Qura...| null| 1298002105|2025-06-29 22:44:43|'''Zubair Ahmad Q...| | enwiki| 0|79974862|Zubair_Ahmad_Qura...| null| 1296728064|2025-06-21 21:34:30|{{Short descripti...| | enwiki| 0|81058129| VTrader| null| 1310920245|2025-09-12 09:18:37|{{In use|time=09:...| | enwiki| 0|81188664| VTrader| null| 1313484016|2025-09-26 13:57:01|{{db-spam|help=of...| | enwiki| 0|81188750| VTrader| null| 1313487301|2025-09-26 14:23:36|{{db-spam}} {{Inf...| | enwiki| 0|81765922| Renderforest| null| 1325804882|2025-12-05 07:33:39|<!-- Please do no...| | enwiki| 0|81509529| Renderforest| null| 1321326660|2025-11-09 23:05:10|{{db-corp|help=of...| +-------+-----------------+--------+--------------------+--------------------+-----------+-------------------+--------------------+ only showing top 20 rows
Regarding T410431#11492207 after a call with @xcollazo we determined that this duplication in the current table is due to another possible bug. Created: T413888
Run a deduplication like what was done in https://phabricator.wikimedia.org/T404975#11197939 :
duplicates situation for mediawiki_content_history_v1:
spark.sql("""
SELECT count(*) as total_duplicate
FROM (
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id ,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+---------------+ |total_duplicate| +---------------+ |7 | +---------------+
Ran:
hostname -f an-launcher1003.eqiad.wmnet sudo -u analytics bash screen -S apizzata-deduplication spark3-sql \ --master yarn \ --conf spark.driver.maxResultSize=8G \ --conf spark.dynamicAllocation.maxExecutors=200 \ --conf spark.sql.shuffle.partitions=2048 \ --conf spark.sql.iceberg.locality.enabled=true \ --conf spark.reducer.maxReqsInFlight=1 \ --conf spark.shuffle.io.retryWait=180s \ --conf spark.shuffle.io.maxRetries=10 \ --executor-cores 2 \ --executor-memory 16G \ --driver-cores 4 \ --driver-memory 16G \ --name apizzata-deduplication DELETE FROM wmf_content.mediawiki_content_history_v1 where (wiki_id = 'enwiki' AND revision_id = 1330296521 AND page_id = 81987627 AND row_content_update_dt = '2026-01-10 03:27:44.932011' AND row_visibility_update_dt = '2025-12-30 16:26:48.691881' AND row_move_update_dt = '2025-12-30 16:26:48.694604' ) OR (wiki_id = 'enwiki' AND revision_id = 1331165397 AND page_id = 75767596 AND row_content_update_dt = '2026-01-04 23:05:54.655830' AND row_visibility_update_dt = '2026-01-04 23:05:54.655830' AND row_move_update_dt = '2026-01-04 23:05:54.655830' ) OR (wiki_id = 'fawiki' AND revision_id = 43192476 AND page_id = 6371151 AND row_content_update_dt = '2025-12-16 20:36:54.683322' AND row_visibility_update_dt = '2025-12-16 20:36:54.683322' AND row_move_update_dt = '2025-12-16 20:36:54.683322' ) OR (wiki_id = 'ruwiki' AND revision_id = 150652639 AND page_id = 11297541 AND row_content_update_dt = '2025-12-14 21:01:48.685095' AND row_visibility_update_dt = '2025-12-14 21:01:48.685095' AND row_move_update_dt = '2025-12-14 21:25:30.693726' ) OR (wiki_id = 'srwiki' AND revision_id = 30632336 AND page_id = 4636176 AND row_content_update_dt = '2026-01-03 17:54:24.671965' AND row_visibility_update_dt = '2026-01-03 17:54:24.671965' AND row_move_update_dt = '2026-01-03 17:54:24.671965' ) OR (wiki_id = 'ukwiki' AND revision_id = 46954248 AND page_id = 5780335 AND row_content_update_dt = '2025-12-14 22:07:42.681037' AND row_visibility_update_dt = '2025-12-14 22:07:42.681037' AND row_move_update_dt = '2025-12-14 22:07:42.681037' ) OR (wiki_id = 'zhwiki' AND revision_id = 90803970 AND page_id = 6683401 AND row_content_update_dt = '2026-01-10 06:42:35.437477' AND row_visibility_update_dt = '2025-12-23 08:40:52.649861' AND row_move_update_dt = '2025-12-23 08:40:52.649861' ); Response code Time taken: 106.825 seconds
and the results are:
+---------------+ |total_duplicate| +---------------+ |0 | +---------------+
duplicates situation for mediawiki_content_current_v1:
spark.sql("""
SELECT count(*) as total_duplicate
FROM(
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_current_v1
GROUP BY wiki_id ,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+---------------+ |total_duplicate| +---------------+ |14 | +---------------+
Ran:
hostname -f an-launcher1003.eqiad.wmnet sudo -u analytics bash screen -S apizzata-deduplication spark3-sql \ --master yarn \ --conf spark.driver.maxResultSize=8G \ --conf spark.dynamicAllocation.maxExecutors=200 \ --conf spark.sql.shuffle.partitions=2048 \ --conf spark.sql.iceberg.locality.enabled=true \ --conf spark.reducer.maxReqsInFlight=1 \ --conf spark.shuffle.io.retryWait=180s \ --conf spark.shuffle.io.maxRetries=10 \ --executor-cores 2 \ --executor-memory 16G \ --driver-cores 4 \ --driver-memory 16G \ --name apizzata-deduplication DELETE FROM wmf_content.mediawiki_content_current_v1 where (wiki_id = 'arwiki' AND revision_id = 72995661 AND page_id = 10231859 AND row_update_dt = '2025-12-31 22:26:22.656955' ) OR (wiki_id = 'enwiki' AND revision_id = 1253108411 AND page_id = 53942450 AND row_update_dt = '2024-10-24 11:37:18.914003' ) OR (wiki_id = 'enwiki' AND revision_id = 1303762563 AND page_id = 78660731 AND row_update_dt = '2025-08-01 23:05:09.315595' ) OR (wiki_id = 'enwiki' AND revision_id = 1323542006 AND page_id = 46959903 AND row_update_dt = '2025-11-22 10:54:41.922455' ) OR (wiki_id = 'itwiki' AND revision_id = 148046821 AND page_id = 10680837 AND row_update_dt = '2025-11-18 11:11:16.250401' ) OR (wiki_id = 'plwikisource' AND revision_id = 784361 AND page_id = 13434 AND row_update_dt = '2024-07-01 00:00:00' ) OR (wiki_id = 'ruwiki' AND revision_id = 151066738 AND page_id = 11367754 AND row_update_dt = '2026-01-08 21:36:24.658308' ) OR (wiki_id = 'zhwiki' AND revision_id = 50621706 AND page_id = 6238101 AND row_update_dt = '2024-07-01 00:00:00' ) OR (wiki_id = 'zhwiki' AND revision_id = 61029002 AND page_id = 6239848 AND row_update_dt = '2024-07-01 00:00:00' ) OR (wiki_id = 'zhwiki' AND revision_id = 76875414 AND page_id = 8354720 AND row_update_dt = '2024-07-01 00:00:00' ) OR (wiki_id = 'zhwiki' AND revision_id = 88279594 AND page_id = 9127190 AND row_update_dt = '2025-07-15 02:49:45.688872' ) OR (wiki_id = 'zhwiki' AND revision_id = 88334334 AND page_id = 9097160 AND row_update_dt = '2025-07-18 08:49:25.688488' ) OR (wiki_id = 'zhwiki' AND revision_id = 88721122 AND page_id = 9165475 AND row_update_dt = '2025-08-14 00:06:58.424710' ) OR (wiki_id = 'zhwiki' AND revision_id = 88831918 AND page_id = 9091999 AND row_update_dt = '2025-08-21 16:28:36.425405' ); Response code Time taken: 102.375 seconds
and the results are:
+---------------+ |total_duplicate| +---------------+ |0 | +---------------+
Finally, duplicates situation for mediawiki_revision_history_v1:
spark.sql("""
SELECT count(*) as total_duplicate
FROM(
SELECT
count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_revision_history_v1
GROUP BY wiki_id ,revision_id
HAVING count > 1
)
""").show(3000, truncate=False)+---------------+ |total_duplicate| +---------------+ |8 | +---------------+
Ran:
hostname -f an-launcher1003.eqiad.wmnet sudo -u analytics bash screen -S apizzata-deduplication spark3-sql \ --master yarn \ --conf spark.driver.maxResultSize=8G \ --conf spark.dynamicAllocation.maxExecutors=200 \ --conf spark.sql.shuffle.partitions=2048 \ --conf spark.sql.iceberg.locality.enabled=true \ --conf spark.reducer.maxReqsInFlight=1 \ --conf spark.shuffle.io.retryWait=180s \ --conf spark.shuffle.io.maxRetries=10 \ --executor-cores 2 \ --executor-memory 16G \ --driver-cores 4 \ --driver-memory 16G \ --name apizzata-deduplication DELETE FROM wmf_content.mediawiki_revision_history_v1 WHERE (wiki_id = 'enwiki' AND revision_id = 1330296521 AND page_id = 81987627 AND row_content_update_dt = '2026-01-10 03:27:44.932011' AND row_visibility_update_dt = '2025-12-30 16:26:48.691881' AND row_move_update_dt = '2025-12-30 16:26:48.694604' ) OR (wiki_id = 'enwiki' AND revision_id = 1331165397 AND page_id = 75767596 AND row_content_update_dt = '2026-01-04 23:05:54.655830' AND row_visibility_update_dt = '2026-01-04 23:05:54.655830' AND row_move_update_dt = '2026-01-04 23:05:54.655830' ) OR (wiki_id = 'enwiki' AND revision_id = 1331423578 AND page_id = 1292737 AND row_content_update_dt = '2026-01-06 06:15:24.653218' AND row_visibility_update_dt = '2026-01-06 06:15:24.653218' AND row_move_update_dt = '2026-01-06 06:15:24.653218' ) OR (wiki_id = 'fawiki' AND revision_id = 43192476 AND page_id = 6371151 AND row_content_update_dt = '2025-12-16 20:36:54.683322' AND row_visibility_update_dt = '2025-12-16 20:36:54.683322' AND row_move_update_dt = '2025-12-16 20:36:54.683322' ) OR (wiki_id = 'ruwiki' AND revision_id = 150652639 AND page_id = 11297541 AND row_content_update_dt = '2025-12-14 21:01:48.685095' AND row_visibility_update_dt = '2025-12-14 21:01:48.685095' AND row_move_update_dt = '2025-12-14 21:25:30.693726' ) OR (wiki_id = 'srwiki' AND revision_id = 30632336 AND page_id = 4636176 AND row_content_update_dt = '2026-01-03 17:54:24.671965' AND row_visibility_update_dt = '2026-01-03 17:54:24.671965' AND row_move_update_dt = '2026-01-03 17:54:24.671965' ) OR (wiki_id = 'ukwiki' AND revision_id = 46954248 AND page_id = 5780335 AND row_content_update_dt = '2025-12-14 22:07:42.681037' AND row_visibility_update_dt = '2025-12-14 22:07:42.681037' AND row_move_update_dt = '2025-12-14 22:07:42.681037' ) OR (wiki_id = 'zhwiki' AND revision_id = 90803970 AND page_id = 6683401 AND row_content_update_dt = '2026-01-10 06:42:35.437477' AND row_visibility_update_dt = '2025-12-23 08:40:52.649861' AND row_move_update_dt = '2025-12-23 08:40:52.649861' ); Response code Time taken: 292.142 seconds
and the results are:
+---------------+ |total_duplicate| +---------------+ |0 | +---------------+
After today's meeting with @xcollazo and @JAllemandou we realised the following.
Using the query in T410431#11465081:
spark.sql("""
select
meta.dt,
revision.rev_dt,
day,month,year,
page.page_id,
page.page_title,
revision.rev_id,
page_change_kind
from
event.mediawiki_page_content_change_v1
where wiki_id= 'commonswiki'
and page.page_id in(178775087,100282687)
order by meta.dt, page.page_id asc
""").show(truncate=False)+---------------------------+--------------------+---+-----+----+---------+----------------------------+----------+----------------+ |dt |rev_dt |day|month|year|page_id |page_title |rev_id |page_change_kind| +---------------------------+--------------------+---+-----+----+---------+----------------------------+----------+----------------+ |2025-11-19T07:26:41.89519Z |2025-11-19T07:26:37Z|19 |11 |2025|100282687|File:Flag_of_Ulleung.svg |1118294200|edit | |2025-11-20T07:39:25.814734Z|2025-11-20T07:39:21Z|20 |11 |2025|100282687|File:Flag_of_Ulleung.svg |1118806092|edit | |2025-11-20T21:16:29.814428Z|2025-11-20T07:39:21Z|20 |11 |2025|100282687|File:Flag_of_Ulleung.svg |1118806092|delete | |2025-11-20T21:16:35.813491Z|2025-11-19T07:26:37Z|20 |11 |2025|100282687|File:Flag_of_Ulleung.svg |1118294200|undelete | |2025-11-20T21:16:37.824006Z|2025-11-20T21:16:32Z|20 |11 |2025|100282687|File:Flag_of_Ulleung_(2).svg|1119085982|move | |2025-11-20T21:17:31.75636Z |2025-11-20T21:17:30Z|20 |11 |2025|100282687|File:Flag_of_Ulleung_(2).svg|1119086355|edit | |2025-11-20T21:34:28.240769Z|2025-11-20T07:39:21Z|20 |11 |2025|178775087|File:Flag_of_Ulleung.svg |1118806092|undelete | |2025-11-20T21:34:44.24225Z |2025-11-20T21:34:41Z|20 |11 |2025|178775087|File:Flag_of_Ulleung.svg |1119092139|edit | |2025-12-05T08:38:05.952091Z|2025-12-05T08:38:02Z|5 |12 |2025|100282687|File:Flag_of_Ulleung_(2).svg|1125950709|edit | |2025-12-05T08:38:40.678265Z|2025-12-05T08:38:36Z|5 |12 |2025|178775087|File:Flag_of_Ulleung.svg |1125950925|edit | +---------------------------+--------------------+---+-----+----+---------+----------------------------+----------+----------------+
But ordering the result by meta.dt we can infer that the undelete with rev_id 1118294200 of the page_id 1118294200 was a way to reverse the edit of edit 1118806092.
The move 1119085982 is part of the moved page File:Flag of Ulleung.svg to File:Flag of Ulleung (2).svg without leaving a redirect: -- using SplitFileHistory.js procedure that is applied consciously by users. This, allows the users to move the page to a new one without leaving a redirect.
Finally the duplication of the rev_id 1118294298 comes from the delete 1118806092 of the page_id 100282687 and due to the optimization logic we apply when merging data ('set_of_page_ids' pushdown_strategy) that avoids the full scan of the table.
In order to avoid this duplication with @xcollazo we were thinking of two possible solutions:
Additionally, even though this is not a duplication issue but I feel is a factor that brings to duplication, the delete and undelete operation that regularly happen bring incomplete data in the table.
Example being the page 100282687 that has a history of 10 revisions, while in the mediawiki_content_history_v1 table has only 1.
spark.sql("""
select count(*) from wmf_content.mediawiki_content_history_v1
where page_id=100282687 and wiki_id= 'commonswiki'""").show()+--------+ |count(1)| +--------+ | 1| +--------+
Additionally, the incorrect list of the revisions mines the completeness of the table.
While exploring the data and validating the solution in T414779, found the following curious example:
page_id= 69510715 and wiki_id = 'enwiki'
This page on the MariaDB replica presents 51 revisions:
analytics-mysql enwiki
select count(*) from revision where rev_page = 69510715; +----------+ | count(*) | +----------+ | 51 | +----------+
While in all the sources we have:
spark.sql(f"""
SELECT
distinct wiki_id, page['page_id'] AS page_id,
revision['rev_id'] AS revision_id, '1. event' AS source
FROM
{source_mw_event_page_change_table}
WHERE
page['page_id']= 69510715 and wiki_id = 'enwiki'
--AND page_change_kind IN ('create', 'edit', 'move') --removed for the sake of the topic, but in the regular runs we only take these
UNION ALL
SELECT
distinct database, page_id,rev_id, '2. visibility_change' AS source
FROM
{source_mw_revision_visibility_change_table}
WHERE
page_id= 69510715 and database = 'enwiki'
UNION ALL
SELECT
distinct wiki_id, page['page_id'] AS page_id,
revision['rev_id'] AS revision_id, '3. reconcile_enriched' AS source
FROM
{source_mw_content_history_reconcile_enriched_table}
WHERE
page['page_id']= 69510715 and wiki_id = 'enwiki'
--AND page_change_kind IN ('create', 'edit', 'move') --removed for the sake of the topic, but in the regular runs we only take these
UNION ALL
SELECT
distinct wiki_id, page_id, revision_id, '4. inconsistent' AS source
FROM
{source_mw_inconsistent_rows_table}
WHERE page_id= 69510715 and wiki_id = 'enwiki'""").show(truncate=False)returns:
+-------+--------+-----------+--------+ |wiki_id|page_id |revision_id|source | +-------+--------+-----------+--------+ |enwiki |69510715|1314189039 |1. event| +-------+--------+-----------+--------+
Deep diving in the "1. event" category:
spark.sql(f"""
SELECT
wiki_id, page['page_id'] AS page_id,
revision['rev_id'] AS revision_id, page_change_kind
FROM
{source_mw_event_page_change_table}
WHERE
page['page_id']= 69510715 and wiki_id = 'enwiki'""").show()+-------+--------+-----------+----------------+ |wiki_id| page_id|revision_id|page_change_kind| +-------+--------+-----------+----------------+ | enwiki|69510715| 1314189039| delete| | enwiki|69510715| 1314189039| undelete| +-------+--------+-----------+----------------+
I have yet to understand what's happening behind the scenes, but maybe the presence of the delete does not allow for a correct reconciliation?
What's your thoughts @xcollazo @JAllemandou
I have yet to understand what's happening behind the scenes, but maybe the presence of the delete does not allow for a correct reconciliation?
What's your thoughts @xcollazo @JAllemandou
Ok after checking the query used in the reconciliation logic, we don't emit any reconciliation data because the mwch table does not have any record for the page 69510715:
wiki_id= 'enwiki'
target_table= 'wmf_content.mediawiki_content_history_v1'
spark.sql(f"""
SELECT page_id,
revision_id,
revision_dt,
revision_content_is_visible,
user_id,
user_text,
user_is_visible,
revision_comment_is_visible,
revision_size
FROM {target_table}
WHERE wiki_id = '{wiki_id}'
and page_id = 69510715 """).count()Returns 0.
We have changed the pushdown_strategy to earliest_revision_dt and this should avoid the duplication. AFAICS from the 2026-01-11 (day that we changed) we are not having duplicates anymore.
We will continue monitoring the situation for the next 2 weeks (up to 2026-02-11) and if no duplicate shows up we can consider the situation fixed.
cc @xcollazo
spark.sql("""
SELECT count(1) as count FROM (
SELECT count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_history_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
""").show(300, truncate=False)returned
+-----+ |count| +-----+ |0 | +-----+
As already stated in T410431#11566608 we now consider the issue solved. Will push relevant changes to the repo and close the ticket.
@APizzata-WMF can you please also check mediawiki_content_current_v1 and mediawiki_revision_history_v1?
spark.sql("""
SELECT count(1) as count FROM (
SELECT count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_content_current_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
""").show(300, truncate=False)+-----+ |count| +-----+ |0 | +-----+
spark.sql("""
SELECT count(1) as count FROM (
SELECT count(1) as count,
wiki_id,
revision_id
FROM wmf_content.mediawiki_revision_history_v1
GROUP BY wiki_id, revision_id
HAVING count > 1
)
""").show(300, truncate=False)+-----+ |count| +-----+ |0 | +-----+
Both of them are free of duplicates!
Just checked again all wmf_content tables for duplicates since the monthly reconcile is now done. All returned empty.
Phew, I think we are actually done with this particular saga.