hive (wmf_raw)> describe mediawiki_slots;
OK
col_name	data_type	comment
slot_revision_id    	bigint              	Reference to revision.rev_id or archive.ar_rev_id. slot_revision_id and slot_role_id together comprise the primary key with wiki_db.
slot_role_id        	int                 	Reference to slot_roles.role_id
slot_content_id     	bigint              	Reference to content.content_id
slot_origin         	bigint              	The revision.rev_id of the revision that originated the slot's content. To find revisions that changed slots, look for slot_origin = slot_revision_id.
snapshot            	string              	Versioning information to keep multiple datasets (YYYY-MM for regular labs imports)
wiki_db             	string              	The wiki_db project 
	 	 
# Partition Information	 	 
# col_name            	data_type           	comment             
	 	 
snapshot            	string              	Versioning information to keep multiple datasets (YYYY-MM for regular labs imports)
wiki_db             	string              	The wiki_db project 
Time taken: 3.26 seconds, Fetched: 12 row(s)

hostname -f
an-launcher1002.eqiad.wmnet

sudo -u analytics bash

screen -S xcollazo_cp_mediawiki_content_current_v2

kerberos-run-command analytics spark3-sql \
 --driver-cores 4 \
 --master yarn \
 --conf spark.driver.maxResultSize=8G \
 --conf spark.dynamicAllocation.maxExecutors=48 \
 --conf spark.sql.shuffle.partitions=1024 \
 --conf spark.sql.iceberg.locality.enabled=false \
 --executor-cores 2 \
 --executor-memory 32G \
 --driver-memory 16G \
 --name xcollazo_cp_mediawiki_content_current_v2


CREATE TABLE wmf_content.mediawiki_content_current_v2 (
    page_id                     BIGINT    COMMENT 'The (database) page ID of the page.',
    page_namespace_id           INT       COMMENT 'The id of the namespace this page belongs to.',
    page_title                  STRING    COMMENT 'The normalized title of the page. If page_namespace_id = 0, then this is the non namespaced title. If page_namespace_id != 0, then the title is prepended with the localized namespace. Examples for "enwiki": "Main_Page" and "Talk:Main_Page".',
    page_redirect_target        STRING    COMMENT 'title of the redirected-to page, if any. Same rules as page_title.',
    user_id                     BIGINT    COMMENT 'id of the user that made the revision; null if anonymous, zero if old system user, and -1 when deleted or malformed XML was imported',
    user_text                   STRING    COMMENT 'text of the user that made the revision (either username or IP)',
    user_is_visible             BOOLEAN   COMMENT 'Whether the user that made the revision is visible. If this is false, then the user should be redacted when shown publicly. See RevisionRecord->DELETED_USER.',
    revision_id                 BIGINT    COMMENT 'The (database) revision ID.',
    revision_parent_id          BIGINT    COMMENT 'The (database) revision ID. of the parent revision',
    revision_dt                 TIMESTAMP COMMENT 'The (database) time this revision was created. This is rev_timestamp in the MediaWiki database.',
    revision_is_minor_edit      BOOLEAN   COMMENT 'True if the editor marked this revision as a minor edit.',
    revision_comment            STRING    COMMENT 'The comment left by the user when this revision was made.',
    revision_comment_is_visible BOOLEAN   COMMENT 'Whether the comment of the revision is visible. If this is false, then the comment should be redacted when shown publicly. See RevisionRecord->DELETED_COMMENT.',
    revision_sha1               STRING    COMMENT 'Nested SHA1 hash of hashes of all content slots. See https://www.mediawiki.org/wiki/Manual:Revision_table#rev_sha1',
    revision_size               BIGINT    COMMENT 'the sum of the content_size of all content slots',
    revision_content_slots      MAP<
                                    STRING,
                                    STRUCT<content_body:   STRING,
                                           content_format: STRING,
                                           content_model:  STRING,
                                           content_sha1:   STRING,
                                           content_size:   BIGINT,
                                           origin_rev_id:  BIGINT
                                    >
                                >         COMMENT 'a MAP containing all the content slots associated to this revision. Typically just the "main" slot, but also "mediainfo" for commonswiki.',
    revision_content_is_visible BOOLEAN   COMMENT 'Whether revision_content_slots is visible. If this is false, then any content should be redacted when shown publicly. See RevisionRecord->DELETED_TEXT.',
    wiki_id                     STRING    COMMENT 'The wiki ID, which is usually the same as the MediaWiki database name. E.g. enwiki, metawiki, etc.',
    row_update_dt               TIMESTAMP COMMENT 'Control column to efficiently update this table from source at wmf_content.mediawiki_content_history_v1.'
)
USING ICEBERG
PARTITIONED BY (wiki_id)                             -- wiki_id partitioning is familiar to users
TBLPROPERTIES (
    'format-version' = '2',                                -- allow merge-on-read
    'write.merge.mode' = 'merge-on-read',                  -- enable merge-on-read when doing MERGE INTO
    'write.update.mode' = 'merge-on-read',                 -- enable merge-on-read when doing UPDATE
    'write.delete.mode' = 'merge-on-read',                 -- enable merge-on-read when doing DELETE
    'write.format.default' = 'parquet',                    -- parquet is currently the only format with min/max stats
    'write.target-file-size-bytes' = '134217728',          -- cap files at 128MB files so executors with 1 core, 16GB RAM can read the entire table
    'write.metadata.delete-after-commit.enabled' = 'true'  -- don't hoard old metadata files. T393012#10781930.
)
COMMENT 'Contains the latest revision for all pages for all wikis. Updated on a daily basis.'
LOCATION 'hdfs://analytics-hadoop/wmf/data/wmf_content/mediawiki_content_current_v1_1'




INSERT INTO wmf_content.mediawiki_content_current_v2
SELECT
    page_id,
    page_namespace_id,
    page_title,
    page_redirect_target,
    user_id,
    user_text,
    user_is_visible,
    revision_id,
    revision_parent_id,
    revision_dt,
    revision_is_minor_edit,
    revision_comment,
    revision_comment_is_visible,
    revision_sha1,
    revision_size,
    transform_values(revision_content_slots,
             (k, v) -> (
                v.content_body,
                v.content_format,
                v.content_model,
                v.content_sha1,
                v.content_size,
                revision_id as origin_rev_id
             )
    ) as revision_content_slots,
    revision_content_is_visible,
    wiki_id,
    row_update_dt
FROM wmf_content.mediawiki_content_current_v1
ORDER BY wiki_id;

Antoine Quhen:
First try failed: application_1758550516505_495012
Second try: application_1758550516505_497903

Antoine Quhen
  Job complete at 2nd try.
25/10/09 01:12:36 INFO MediawikiContentHistoryMigration: === MIGRATION VERIFICATION ===
25/10/09 01:12:36 INFO MediawikiContentHistoryMigration: Source table: 1017 wikis, 7429585869 total rows
25/10/09 01:12:36 INFO MediawikiContentHistoryMigration: Target table: 1017 wikis, 7429585869 total rows
25/10/09 01:12:36 INFO MediawikiContentHistoryMigration: ✓ All wiki row counts match between source and target tables
Now switching tables.

Antoine Quhen
  wmf_content.mediawiki_content_history_v1
now points to hdfs://analytics-hadoop/wmf/data/wmf_content/mediawiki_content_history_v1_1
and has the new field origin_rev_id

Antoine Quhen
  Same with mediawiki_content_current_v1 .

Antoine Quhen
  Old tables are still here: mediawiki_content_history_v1_old and mediawiki_content_current_v1_old

Antoine Quhen
  With data in: /wmf/data/wmf_content/mediawiki_content_history_v1 and /wmf/data/wmf_content/mediawiki_content_current_v1

Antoine Quhen
  Basic request on the new tables performed :white_check_mark:

Antoine Quhen
  Dags tagged mw_content resumed. (ex: mw_content_xml_export_current_mid_month @xcollazo not sure it was activated)

Antoine Quhen
  Pausing them all again. mediawiki-content-dump repo needs a new release + airflow-dags needs to point to it.

spark.sql("""
SELECT count(1) as count,
       wiki_id
FROM wmf_content.mediawiki_content_current_v1
GROUP BY wiki_id
ORDER BY count DESC
  """).show()

+---------+------------+
|    count|     wiki_id|
+---------+------------+
|166168117| commonswiki|
|124617943|wikidatawiki|
| 64373818|      enwiki|
| 19529122|      viwiki|
| 13989995|  ruwikinews|
| 13695549|      frwiki|
| 12831912|    metawiki|
| 11232295|     cebwiki|
| 10432572|enwiktionary|
|  8949355|      arwiki|
|  8523058|      eswiki|
|  8435288|      ruwiki|
|  8415394|      itwiki|
|  8411916|      dewiki|
|  8207016|      zhwiki|
|  7583869|frwiktionary|
|  6342310|      svwiki|
|  5995765|      ptwiki|
|  5983749|      fawiki|
|  5831919|mgwiktionary|
+---------+------------+
only showing top 20 rows


spark.sql("""
SELECT count(1) as count,
       wiki_id
FROM wmf_content.mediawiki_content_current_v1_old
GROUP BY wiki_id
ORDER BY count DESC
  """).show()
[Stage 33:=================================================>(11405 + 2) / 11407]
+---------+------------+
|    count|     wiki_id|
+---------+------------+
|166168117| commonswiki|
|124617943|wikidatawiki|
| 64373818|      enwiki|
| 19529122|      viwiki|
| 13989995|  ruwikinews|
| 13695549|      frwiki|
| 12831912|    metawiki|
| 11232295|     cebwiki|
| 10432572|enwiktionary|
|  8949355|      arwiki|
|  8523058|      eswiki|
|  8435288|      ruwiki|
|  8415394|      itwiki|
|  8411916|      dewiki|
|  8207016|      zhwiki|
|  7583869|frwiktionary|
|  6342310|      svwiki|
|  5995765|      ptwiki|
|  5983749|      fawiki|
|  5831919|mgwiktionary|
+---------+------------+
only showing top 20 rows

MERGE INTO wmf_content.mediawiki_content_history_v1 t
USING (
    SELECT slot_revision_id, slot_role_id, slot_origin
    FROM wmf_raw.mediawiki_slots
    WHERE snapshot = '2025-09'
      AND wiki_db = 'commonswiki'
      AND slot_revision_id != slot_origin
      AND slot_role_id = 1 -- 1 == main
) s

ON  t.wiki_id = 'commonswiki'
AND t.revision_id = s.slot_revision_id
AND t.revision_content_slots['main'].origin_rev_id != s.slot_origin

WHEN MATCHED THEN
  UPDATE SET
    t.revision_content_slots =
      transform_values(
        revision_content_slots,
        (k, v) -> CASE
                    WHEN k = 'main'
                    THEN (
                          v.content_body,
                          v.content_format,
                          v.content_model,
                          v.content_sha1,
                          v.content_size,
                          s.slot_origin AS origin_rev_id
                    )
                    ELSE v
                  END
      )

hostname -f
an-launcher1002.eqiad.wmnet

sudo -u analytics bash

screen -S xcollazo_fix_origin_rev_id_for_commonswiki

kerberos-run-command analytics spark3-sql \
 --driver-cores 4 \
 --master yarn \
 --conf spark.driver.maxResultSize=8G \
 --conf spark.dynamicAllocation.maxExecutors=48 \
 --conf spark.sql.shuffle.partitions=1024 \
 --conf spark.sql.iceberg.locality.enabled=false \
 --executor-cores 2 \
 --executor-memory 32G \
 --driver-memory 16G \
 --name xcollazo_fix_origin_rev_id_for_commonswiki
 
 spark-sql (default)>  MERGE INTO wmf_content.mediawiki_content_history_v1 t
                    >  USING (
                    >      SELECT slot_revision_id, slot_role_id, slot_origin
                    >      FROM wmf_raw.mediawiki_slots
                    >      WHERE snapshot = '2025-09'
                    >        AND wiki_db = 'commonswiki'
                    >        AND slot_revision_id != slot_origin
                    >        AND slot_role_id = 1 -- 1 == main
                    >  ) s
                    > 
                    >  ON  t.wiki_id = 'commonswiki'
                    >  AND t.revision_id = s.slot_revision_id
                    >  AND t.revision_content_slots['main'].origin_rev_id != s.slot_origin
                    > 
                    >  WHEN MATCHED THEN
                    >    UPDATE SET
                    >      t.revision_content_slots =
                    >        transform_values(
                    >          revision_content_slots,
                    >          (k, v) -> CASE
                    >                      WHEN k = 'main'
                    >                      THEN (
                    >                            v.content_body,
                    >                            v.content_format,
                    >                            v.content_model,
                    >                            v.content_sha1,
                    >                            v.content_size,
                    >                            s.slot_origin AS origin_rev_id
                    >                      )
                    >                      ELSE v
                    >                    END
                    >        );
 Response code
 Time taken: 2094.345 seconds

spark-sql (default)>  MERGE INTO wmf_content.mediawiki_content_history_v1 t
                   >  USING (
                   >      SELECT slot_revision_id, slot_role_id, slot_origin
                   >      FROM wmf_raw.mediawiki_slots
                   >      WHERE snapshot = '2025-09'
                   >        AND wiki_db = 'commonswiki'
                   >        AND slot_revision_id != slot_origin
                   >        AND slot_role_id = 2 -- 2 == mediainfo
                   >  ) s
                   > 
                   >  ON  t.wiki_id = 'commonswiki'
                   >  AND t.revision_id = s.slot_revision_id
                   >  AND t.revision_content_slots['mediainfo'].origin_rev_id != s.slot_origin
                   > 
                   >  WHEN MATCHED THEN
                   >    UPDATE SET
                   >      t.revision_content_slots =
                   >        transform_values(
                   >          revision_content_slots,
                   >          (k, v) -> CASE
                   >                      WHEN k = 'mediainfo'
                   >                      THEN (
                   >                            v.content_body,
                   >                            v.content_format,
                   >                            v.content_model,
                   >                            v.content_sha1,
                   >                            v.content_size,
                   >                            s.slot_origin AS origin_rev_id
                   >                      )
                   >                      ELSE v
                   >                    END
                   >        );
Response code
Time taken: 2470.445 seconds

MERGE INTO wmf_content.mediawiki_content_current_v1 t
USING (
    SELECT slot_revision_id, slot_role_id, slot_origin
    FROM wmf_raw.mediawiki_slots
    WHERE snapshot = '2025-09'
      AND wiki_db = 'commonswiki'
      AND slot_revision_id != slot_origin
      AND slot_role_id = 1 -- 1 == main
) s

ON  t.wiki_id = 'commonswiki'
AND t.revision_id = s.slot_revision_id
AND t.revision_content_slots['main'].origin_rev_id != s.slot_origin

WHEN MATCHED THEN
  UPDATE SET
    t.revision_content_slots =
      transform_values(
        revision_content_slots,
        (k, v) -> CASE
                    WHEN k = 'main'
                    THEN (
                          v.content_body,
                          v.content_format,
                          v.content_model,
                          v.content_sha1,
                          v.content_size,
                          s.slot_origin AS origin_rev_id
                    )
                    ELSE v
                  END
      )

MERGE INTO wmf_content.mediawiki_content_current_v1 t
USING (
    SELECT slot_revision_id, slot_role_id, slot_origin
    FROM wmf_raw.mediawiki_slots
    WHERE snapshot = '2025-09'
      AND wiki_db = 'commonswiki'
      AND slot_revision_id != slot_origin
      AND slot_role_id = 2 -- 2 == mediainfo
) s

ON  t.wiki_id = 'commonswiki'
AND t.revision_id = s.slot_revision_id
AND t.revision_content_slots['mediainfo'].origin_rev_id != s.slot_origin

WHEN MATCHED THEN
  UPDATE SET
    t.revision_content_slots =
      transform_values(
        revision_content_slots,
        (k, v) -> CASE
                    WHEN k = 'mediainfo'
                    THEN (
                          v.content_body,
                          v.content_format,
                          v.content_model,
                          v.content_sha1,
                          v.content_size,
                          s.slot_origin AS origin_rev_id
                    )
                    ELSE v
                  END
      )

select count(1)
from wmf_content.mediawiki_content_current_v1
WHERE wiki_id = 'commonswiki' ;
-- 166 490 143

select count(1)
from wmf_content.mediawiki_content_current_v1
WHERE wiki_id = 'commonswiki'
    and revision_content_slots['main'].origin_rev_id <> revision_id;
-- 89 990 930

select count(1)
from wmf_content.mediawiki_content_current_v1
WHERE wiki_id = 'commonswiki'
  and revision_content_slots['mediainfo'].origin_rev_id <> revision_id;
-- 28 808 901