Page MenuHomePhabricator
Paste P92610

spike_T426469_revert_window.md
ActivePublic
Actions

Authored by xcollazo on May 19 2026, 3:38 PM.
Project Tags
None
Referenced Files
F82665474: spike_T426469_revert_window.md
May 19 2026, 3:38 PM
Subscribers
None
```python
import wmfdata
spark = wmfdata.spark.create_custom_session(
master="yarn",
spark_config={
"spark.dynamicAllocation.maxExecutors": 256,
"spark.executor.memory": "16g",
"spark.driver.memory": "8g",
"spark.driver.cores": "2",
"spark.executor.cores": "2",
"spark.sql.shuffle.partitions": 512,
"spark.driver.maxResultSize": "2G",
"spark.sql.legacy.timeParserPolicy": "LEGACY"
}
)
print(f"Spark {spark.version} ready")
```
SPARK_HOME: /srv/home/xcollazo/.conda/envs/2026-03-23T17.32.05_xcollazo/lib/python3.10/site-packages/pyspark
Using Hadoop client lib jars at 3.2.0, provided by Spark.
PYSPARK_PYTHON=/opt/conda-analytics/bin/python3
Spark 3.1.2 ready
```python
#
# Run cells top-to-bottom. Adjust the two parameters in the Setup cell.
# Source table: wmf.mediawiki_history (monthly batch, snapshot-partitioned).
# The incremental table uses wiki_id; this table uses wiki_db — noted per query.
# ── Setup ─────────────────────────────────────────────────────────────────────
```
```python
SNAPSHOT = "2026-03" # latest wmf.mediawiki_history snapshot, e.g. "2024-11"
MEDIUM_WIKI = "frwiki" # one medium-sized wiki to profile alongside enwiki
```
```python
# ── Q1: Revert-window histogram (all wikis) ───────────────────────────────────
# How much of the reverted-revision signal falls outside the current 48h window?
# Source: wmf.mediawiki_history, all wikis, single snapshot.
```
```python
q1 = spark.sql(f"""
SELECT
bucket,
reverted_revisions,
pct_of_total,
ROUND(100.0 * SUM(reverted_revisions) OVER (
ORDER BY bucket
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) / SUM(reverted_revisions) OVER (), 2) AS cumulative_pct
FROM (
SELECT
CASE
WHEN revision_seconds_to_identity_revert <= 48 * 3600 THEN '1_0h_to_48h'
WHEN revision_seconds_to_identity_revert <= 7 * 24 * 3600 THEN '2_48h_to_7d'
WHEN revision_seconds_to_identity_revert <= 30 * 24 * 3600 THEN '3_7d_to_30d'
WHEN revision_seconds_to_identity_revert <= 90 * 24 * 3600 THEN '4_30d_to_90d'
WHEN revision_seconds_to_identity_revert <=365 * 24 * 3600 THEN '5_90d_to_1y'
ELSE '6_over_1y'
END AS bucket,
COUNT(*) AS reverted_revisions,
ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS pct_of_total
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
GROUP BY 1
) t
ORDER BY bucket
""")
print("=== Q1: Revert-window histogram (all wikis) ===")
q1.show(truncate=False)
```
=== Q1: Revert-window histogram (all wikis) ===
26/05/19 00:22:17 WARN SessionState: METASTORE_FILTER_HOOK will be ignored, since hive.security.authorization.manager is set to instance of HiveAuthorizerFactory.
26/05/19 00:22:20 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
26/05/19 00:22:20 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
26/05/19 00:22:20 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
[Stage 1:======================================================>(511 + 1) / 512]
+------------+------------------+------------+--------------+
|bucket |reverted_revisions|pct_of_total|cumulative_pct|
+------------+------------------+------------+--------------+
|1_0h_to_48h |257574702 |55.13 |55.13 |
|2_48h_to_7d |30786999 |6.59 |61.72 |
|3_7d_to_30d |39269860 |8.41 |70.13 |
|4_30d_to_90d|32347677 |6.92 |77.05 |
|5_90d_to_1y |44515342 |9.53 |86.58 |
|6_over_1y |62687592 |13.42 |100.00 |
+------------+------------------+------------+--------------+
```python
# ── Q2: Cross-month revert frequency ─────────────────────────────────────────
# How often does the reverting edit land in month M+1 (or later) relative to
# the original revision? These are the rows the daily writer would need to
# back-patch on source='snapshot' rows.
# %%
# Part A: summary by same-month vs cross-month
q2a = spark.sql(f"""
SELECT
CASE
WHEN SUBSTRING(event_timestamp, 1, 7)
= SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
THEN 'same_month'
ELSE 'cross_month'
END AS revert_timing,
COUNT(*) AS cnt,
ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS pct
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
GROUP BY 1
ORDER BY 1
""")
print("=== Q2a: Same-month vs cross-month revert summary ===")
q2a.show(truncate=False)
# Part B: break down cross-month reverts by how many months later
q2b = spark.sql(f"""
SELECT
months_later,
COUNT(*) AS reverted_revisions,
ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2) AS pct_of_cross_month
FROM (
SELECT
FLOOR(
DATEDIFF(
from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert),
event_timestamp
) / 30
) AS months_later
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
AND SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
) x
GROUP BY 1
ORDER BY 1
""")
print("=== Q2b: Cross-month breakdown — months until revert ===")
q2b.show(30, truncate=False)
# Part C: by wiki_db — identify which wikis have highest cross-month revert rates
q2c = spark.sql(f"""
SELECT
wiki_db,
COUNT(*) AS total_reverted,
SUM(CASE
WHEN SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
THEN 1 ELSE 0
END) AS cross_month_reverted,
ROUND(100.0 * SUM(CASE
WHEN SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
THEN 1 ELSE 0
END) / COUNT(*), 2) AS cross_month_pct
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
GROUP BY 1
HAVING COUNT(*) >= 1000
ORDER BY cross_month_pct DESC
LIMIT 30
""")
print("=== Q2c: Cross-month revert rate by wiki (top 30, min 1000 reverts) ===")
q2c.show(30, truncate=False)
```
=== Q2a: Same-month vs cross-month revert summary ===
26/05/19 00:42:16 WARN SessionState: METASTORE_FILTER_HOOK will be ignored, since hive.security.authorization.manager is set to instance of HiveAuthorizerFactory.
26/05/19 00:42:19 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
+-------------+---------+-----+
|revert_timing|cnt |pct |
+-------------+---------+-----+
|cross_month |165371876|35.40|
|same_month |301810296|64.60|
+-------------+---------+-----+
=== Q2b: Cross-month breakdown — months until revert ===
26/05/19 00:42:53 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
+------------+------------------+------------------+
|months_later|reverted_revisions|pct_of_cross_month|
+------------+------------------+------------------+
|0 |25296118 |15.30 |
|1 |20465743 |12.38 |
|2 |12232654 |7.40 |
|3 |8916686 |5.39 |
|4 |6867516 |4.15 |
|5 |5604036 |3.39 |
|6 |4928183 |2.98 |
|7 |4296809 |2.60 |
|8 |3828752 |2.32 |
|9 |3712153 |2.24 |
|10 |3131675 |1.89 |
|11 |2905418 |1.76 |
|12 |2562404 |1.55 |
|13 |2391821 |1.45 |
|14 |2105121 |1.27 |
|15 |1966547 |1.19 |
|16 |1860446 |1.13 |
|17 |1754318 |1.06 |
|18 |1724715 |1.04 |
|19 |1970755 |1.19 |
|20 |1666240 |1.01 |
|21 |1621938 |0.98 |
|22 |1635197 |0.99 |
|23 |1324197 |0.80 |
|24 |1269642 |0.77 |
|25 |1155987 |0.70 |
|26 |1288207 |0.78 |
|27 |1077585 |0.65 |
|28 |1003225 |0.61 |
|29 |942287 |0.57 |
+------------+------------------+------------------+
only showing top 30 rows
=== Q2c: Cross-month revert rate by wiki (top 30, min 1000 reverts) ===
[Stage 7:======================================================>(506 + 6) / 512]
+--------------------+--------------+--------------------+---------------+
|wiki_db |total_reverted|cross_month_reverted|cross_month_pct|
+--------------------+--------------+--------------------+---------------+
|nnwiktionary |29490 |29378 |99.62 |
|lowiktionary |247497 |246243 |99.49 |
|liwiktionary |195231 |194165 |99.45 |
|zh_min_nanwiktionary|71698 |71204 |99.31 |
|etwiktionary |358500 |355991 |99.30 |
|kkwiktionary |64250 |63746 |99.22 |
|chrwiktionary |339956 |337056 |99.15 |
|euwiktionary |56893 |56281 |98.92 |
|cywiktionary |35441 |35007 |98.78 |
|knwiktionary |136612 |134790 |98.67 |
|iowiktionary |266943 |263173 |98.59 |
|pnbwiktionary |2266 |2227 |98.28 |
|vowiktionary |20769 |20382 |98.14 |
|tgwiktionary |27549 |27011 |98.05 |
|tkwiktionary |6455 |6324 |97.97 |
|fywiktionary |23229 |22746 |97.92 |
|hrwiktionary |72468 |70893 |97.83 |
|gawiktionary |14073 |13767 |97.83 |
|cowiktionary |13366 |13044 |97.59 |
|bswiktionary |11491 |11199 |97.46 |
|wowiktionary |7192 |6980 |97.05 |
|mtwiktionary |3899 |3782 |97.00 |
|lvwiktionary |18214 |17649 |96.90 |
|slwiktionary |20155 |19524 |96.87 |
|mywiktionary |76045 |73447 |96.58 |
|zhwiktionary |695244 |670819 |96.49 |
|huwiktionary |523419 |504054 |96.30 |
|kawiktionary |15217 |14625 |96.11 |
|lbwiktionary |19548 |18786 |96.10 |
|sqwiktionary |22972 |22057 |96.02 |
+--------------------+--------------+--------------------+---------------+
```python
# ── Q3: Window size vs. scan cost ────────────────────────────────────────────
# Simulate the revert_seed CTE at 6 candidate window sizes.
# We use wmf.mediawiki_history as the data-volume proxy.
# NOTE: wmf.mediawiki_history is snapshot-partitioned (not date-partitioned), so
# the event_timestamp filter does NOT prune Hive partitions — every run scans the
# full snapshot. The Iceberg incremental table IS date-partitioned; pushdown is
# verified separately in Part B (EXPLAIN).
#
# SIM_DAY must fall within the chosen SNAPSHOT month.
# %%
import time
SIM_DAY = f"{SNAPSHOT}-15" # mid-month of the chosen snapshot, e.g. "2024-11-15"
WINDOWS = {
"48h" : 48,
"7d" : 7 * 24,
"30d" : 30 * 24,
"90d" : 90 * 24,
"1y" : 365 * 24,
"unbounded": None,
}
results_q3 = []
for label, window_hours in WINDOWS.items():
for wiki in ["enwiki", MEDIUM_WIKI]:
if window_hours is not None:
seed_filter = (
f"AND event_timestamp >= TIMESTAMP '{SIM_DAY} 00:00:00'"
f" - INTERVAL {window_hours} HOURS\n"
f" AND event_timestamp < TIMESTAMP '{SIM_DAY} 00:00:00'"
)
else:
seed_filter = f"AND event_timestamp < TIMESTAMP '{SIM_DAY} 00:00:00'"
sql = f"""
WITH revert_seed AS (
SELECT wiki_db, page_id, revision_id, revision_text_sha1, event_timestamp,
FALSE AS is_incoming
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND revision_text_sha1 IS NOT NULL
AND wiki_db = '{wiki}'
AND snapshot = '{SNAPSHOT}'
{seed_filter}
),
incoming AS (
SELECT wiki_db, page_id, revision_id, revision_text_sha1, event_timestamp,
TRUE AS is_incoming
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND revision_text_sha1 IS NOT NULL
AND wiki_db = '{wiki}'
AND snapshot = '{SNAPSHOT}'
AND event_timestamp >= TIMESTAMP '{SIM_DAY} 00:00:00'
AND event_timestamp < TIMESTAMP '{SIM_DAY} 00:00:00' + INTERVAL 24 HOURS
),
revert_candidates AS (
SELECT * FROM revert_seed
UNION ALL
SELECT * FROM incoming
),
sha1_ranked AS (
SELECT *,
row_number() OVER (
PARTITION BY wiki_db, page_id, revision_text_sha1
ORDER BY event_timestamp, revision_id
) AS sha1_rank,
MIN(event_timestamp) OVER (
PARTITION BY wiki_db, page_id, revision_text_sha1
) AS sha1_first_ts
FROM revert_candidates
),
first_revert AS (
SELECT wiki_db, page_id, revision_text_sha1,
revision_id AS first_reverting_rev_id,
event_timestamp AS first_revert_ts
FROM sha1_ranked WHERE sha1_rank = 2
),
revert_annotations AS (
SELECT
r.wiki_db,
r.revision_id,
(r.sha1_rank > 1
AND r.event_timestamp <= r.sha1_first_ts
+ INTERVAL {window_hours if window_hours else 876000} HOURS) AS is_revert,
(r.sha1_rank = 1
AND fr.first_revert_ts IS NOT NULL
AND fr.first_revert_ts <= r.event_timestamp
+ INTERVAL {window_hours if window_hours else 876000} HOURS) AS is_reverted
FROM sha1_ranked r
LEFT JOIN first_revert fr
ON r.wiki_db = fr.wiki_db
AND r.page_id = fr.page_id
AND r.revision_text_sha1 = fr.revision_text_sha1
WHERE r.is_incoming
)
SELECT
COUNT(*) AS incoming_revisions,
SUM(CASE WHEN is_revert THEN 1 ELSE 0 END) AS annotated_reverts,
SUM(CASE WHEN is_reverted THEN 1 ELSE 0 END) AS annotated_reverted
FROM revert_annotations
"""
# Count seed rows (data volume proxy — not timed, fast aggregation)
seed_count_sql = f"""
SELECT COUNT(*) AS seed_rows
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND revision_text_sha1 IS NOT NULL
AND wiki_db = '{wiki}'
AND snapshot = '{SNAPSHOT}'
{seed_filter}
"""
seed_rows = spark.sql(seed_count_sql).collect()[0]["seed_rows"]
t0 = time.time()
row = spark.sql(sql).collect()[0]
elapsed = round(time.time() - t0, 1)
results_q3.append({
"wiki" : wiki,
"window" : label,
"seed_rows" : seed_rows,
"incoming_revisions" : row["incoming_revisions"],
"annotated_reverts" : row["annotated_reverts"],
"annotated_reverted" : row["annotated_reverted"],
"elapsed_s" : elapsed,
})
print(f" wiki={wiki:8s} window={label:10s} seed={seed_rows:>12,} "
f"time={elapsed}s")
from pyspark.sql import Row
q3_df = spark.createDataFrame([Row(**r) for r in results_q3])
print("\n=== Q3: Window size vs. scan/join cost ===")
q3_df.show(truncate=False)
```
wiki=enwiki window=48h seed= 331,366 time=60.0s
wiki=frwiki window=48h seed= 55,979 time=49.7s
wiki=enwiki window=7d seed= 1,283,487 time=57.5s
wiki=frwiki window=7d seed= 203,402 time=53.3s
wiki=enwiki window=30d seed= 5,482,142 time=52.8s
wiki=frwiki window=30d seed= 834,869 time=49.1s
wiki=enwiki window=90d seed= 15,964,768 time=53.7s
wiki=frwiki window=90d seed= 2,621,783 time=49.8s
wiki=enwiki window=1y seed= 62,930,410 time=56.2s
wiki=frwiki window=1y seed= 10,184,608 time=56.1s
wiki=enwiki window=unbounded seed=1,336,444,674 time=88.5s
wiki=frwiki window=unbounded seed= 232,734,771 time=61.5s
=== Q3: Window size vs. scan/join cost ===
[Stage 85:=================================================> (364 + 23) / 387]
+------+---------+----------+------------------+-----------------+------------------+---------+
|wiki |window |seed_rows |incoming_revisions|annotated_reverts|annotated_reverted|elapsed_s|
+------+---------+----------+------------------+-----------------+------------------+---------+
|enwiki|48h |331366 |174012 |3491 |1806 |60.0 |
|frwiki|48h |55979 |28632 |438 |268 |49.7 |
|enwiki|7d |1283487 |174012 |4468 |1727 |57.5 |
|frwiki|7d |203402 |28632 |542 |255 |53.3 |
|enwiki|30d |5482142 |174012 |6069 |1596 |52.8 |
|frwiki|30d |834869 |28632 |710 |240 |49.1 |
|enwiki|90d |15964768 |174012 |6998 |1506 |53.7 |
|frwiki|90d |2621783 |28632 |878 |222 |49.8 |
|enwiki|1y |62930410 |174012 |7994 |1432 |56.2 |
|frwiki|1y |10184608 |28632 |1018 |212 |56.1 |
|enwiki|unbounded|1336444674|174012 |8931 |1396 |88.5 |
|frwiki|unbounded|232734771 |28632 |1208 |212 |61.5 |
+------+---------+----------+------------------+-----------------+------------------+---------+
```python
# ── Q3 Part B: EXPLAIN FORMATTED for Iceberg predicate pushdown ───────────────
# Substitute the actual Iceberg target table name before running.
# Look for "IcebergTableScan" with "Pushed Filters" including event_timestamp
# predicates to confirm partition/file-level pruning is active.
# %%
INCREMENTAL_TABLE = "xcollazo.mediawiki_history_incremental_v1" # adjust as needed
EXPLAIN_WINDOW_HOURS = 7 * 24 # use 7d as the candidate window
explain_sql = f"""
SELECT wiki_db, page_id, revision_id, revision_text_sha1, event_timestamp
FROM {INCREMENTAL_TABLE}
WHERE revision_text_sha1 IS NOT NULL
AND event_timestamp >= TIMESTAMP '{SIM_DAY} 00:00:00' - INTERVAL {EXPLAIN_WINDOW_HOURS} HOURS
AND event_timestamp < TIMESTAMP '{SIM_DAY} 00:00:00'
"""
print(f"=== Q3b: EXPLAIN FORMATTED for revert_seed scan at {EXPLAIN_WINDOW_HOURS}h window ===")
print("(Run this only when the Iceberg incremental table exists and is populated)")
spark.sql(f"EXPLAIN FORMATTED {explain_sql}").show(200, truncate=False)
```
=== Q3b: EXPLAIN FORMATTED for revert_seed scan at 168h window ===
(Run this only when the Iceberg incremental table exists and is populated)
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|plan |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|== Physical Plan ==
* Project (4)
+- * Filter (3)
+- * ColumnarToRow (2)
+- BatchScan (1)
(1) BatchScan
Output [5]: [wiki_db#6564, event_timestamp#6567, page_id#6575L, revision_id#6578L, revision_text_sha1#6583]
spark_catalog.xcollazo.mediawiki_history_incremental_v1 [filters=event_timestamp IS NOT NULL, revision_text_sha1 IS NOT NULL, event_timestamp >= 1772928000000000, event_timestamp < 1773532800000000]
(2) ColumnarToRow [codegen id : 1]
Input [5]: [wiki_db#6564, event_timestamp#6567, page_id#6575L, revision_id#6578L, revision_text_sha1#6583]
(3) Filter [codegen id : 1]
Input [5]: [wiki_db#6564, event_timestamp#6567, page_id#6575L, revision_id#6578L, revision_text_sha1#6583]
Condition : (((isnotnull(event_timestamp#6567) AND isnotnull(revision_text_sha1#6583)) AND (event_timestamp#6567 >= 1772928000000000)) AND (event_timestamp#6567 < 1773532800000000))
(4) Project [codegen id : 1]
Output [5]: [wiki_db#6564, page_id#6575L, revision_id#6578L, revision_text_sha1#6583, event_timestamp#6567]
Input [5]: [wiki_db#6564, event_timestamp#6567, page_id#6575L, revision_id#6578L, revision_text_sha1#6583]
|
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
```python
# ── Q4: MERGE key change feasibility ─────────────────────────────────────────
# Current MERGE key: (source='events', wiki_db, revision_id)
# Proposed key: (wiki_db, revision_id) — drops the source filter so we
# can patch revision_is_identity_reverted on source='snapshot' rows.
#
# We proxy "snapshot" rows as revisions older than the last full month, and
# "events" rows as revisions from the last month. This lets us estimate:
# a) the ratio of snapshot to events rows (write amplification factor)
# b) how many distinct snapshot "date partitions" would be touched per day
# %%
PROXY_EVENTS_CUTOFF = f"{SNAPSHOT}-01" # revisions on/after this date = "events"
# Part A: row-count split by proxy source
q4a = spark.sql(f"""
SELECT
CASE
WHEN event_timestamp >= TIMESTAMP '{PROXY_EVENTS_CUTOFF}'
THEN 'events_proxy'
ELSE 'snapshot_proxy'
END AS source_proxy,
COUNT(*) AS row_count
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
GROUP BY 1
""")
print("=== Q4a: Row-count split — snapshot proxy vs events proxy ===")
q4a.show(truncate=False)
# Part B: for enwiki, how many distinct event_timestamp days exist in snapshot rows?
# This is the maximum number of Iceberg data-file partitions a MERGE could touch
# if it scans the whole snapshot range.
q4b = spark.sql(f"""
SELECT
COUNT(DISTINCT DATE(event_timestamp)) AS distinct_days_in_snapshot,
MIN(DATE(event_timestamp)) AS earliest_day,
MAX(DATE(event_timestamp)) AS latest_day,
COUNT(*) AS snapshot_row_count
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND wiki_db = 'enwiki'
AND event_timestamp < TIMESTAMP '{PROXY_EVENTS_CUTOFF}'
""")
print("=== Q4b: enwiki snapshot date-partition span (write-amplification proxy) ===")
q4b.show(truncate=False)
# Part C: estimate incremental write amplification
# If the MERGE must touch only the cross-month reverted rows (not all snapshot rows),
# how many distinct days do those cross-month reverted revisions span?
q4c = spark.sql(f"""
SELECT
COUNT(*) AS cross_month_reverted_in_snapshot,
COUNT(DISTINCT DATE(event_timestamp)) AS distinct_days_affected,
MIN(DATE(event_timestamp)) AS earliest,
MAX(DATE(event_timestamp)) AS latest
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND wiki_db = 'enwiki'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
AND event_timestamp < TIMESTAMP '{PROXY_EVENTS_CUTOFF}'
AND SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
""")
print("=== Q4c: enwiki — snapshot rows that need back-patch (cross-month reverts) ===")
q4c.show(truncate=False)
```
=== Q4a: Row-count split — snapshot proxy vs events proxy ===
+--------------+----------+
|source_proxy |row_count |
+--------------+----------+
|events_proxy |41425636 |
|snapshot_proxy|8016940250|
+--------------+----------+
=== Q4b: enwiki snapshot date-partition span (write-amplification proxy) ===
+-------------------------+------------+----------+------------------+
|distinct_days_in_snapshot|earliest_day|latest_day|snapshot_row_count|
+-------------------------+------------+----------+------------------+
|9174 |2001-01-15 |2026-02-28|1336245571 |
+-------------------------+------------+----------+------------------+
=== Q4c: enwiki — snapshot rows that need back-patch (cross-month reverts) ===
[Stage 100:==============> (142 + 370) / 512]
+--------------------------------+----------------------+----------+----------+
|cross_month_reverted_in_snapshot|distinct_days_affected|earliest |latest |
+--------------------------------+----------------------+----------+----------+
|68199295 |9044 |2001-01-17|2026-02-28|
+--------------------------------+----------------------+----------+----------+
```python
# ── Q5: Retroactive UPDATE volume per daily run ───────────────────────────────
# For each calendar day D, count revisions written in a prior month whose
# reverting edit lands on D. This is the number of source='snapshot' MERGE
# UPDATEs the daily writer would issue on that day, per window size.
#
# We report this for the full dataset (all wikis) and for enwiki alone,
# at each candidate window size.
# %%
# Part A: per-day UPDATE volume across all wikis (uses the actual revert data)
q5a = spark.sql(f"""
SELECT
DATE(from_unixtime(
unix_timestamp(event_timestamp) + revision_seconds_to_identity_revert
)) AS revert_date,
COUNT(*) AS cross_month_updates_all_wikis
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
AND SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
GROUP BY 1
ORDER BY 1
""")
print("=== Q5a: Per-day UPDATE volume — all wikis ===")
q5a.show(50, truncate=False)
# Part B: summary statistics (median, p95, max) across days
q5b = spark.sql(f"""
SELECT
ROUND(percentile_approx(cross_month_updates_all_wikis, 0.50)) AS median_per_day,
ROUND(percentile_approx(cross_month_updates_all_wikis, 0.95)) AS p95_per_day,
MAX(cross_month_updates_all_wikis) AS max_per_day,
ROUND(AVG(cross_month_updates_all_wikis)) AS avg_per_day
FROM (
SELECT
DATE(from_unixtime(
unix_timestamp(event_timestamp) + revision_seconds_to_identity_revert
)) AS revert_date,
COUNT(*) AS cross_month_updates_all_wikis
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
AND SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
GROUP BY 1
)
""")
print("=== Q5b: UPDATE volume distribution (all wikis) ===")
q5b.show(truncate=False)
# Part C: per-day UPDATE volume for enwiki only — useful for profiling the
# largest single-wiki MERGE cost
q5c = spark.sql(f"""
SELECT
DATE(from_unixtime(
unix_timestamp(event_timestamp) + revision_seconds_to_identity_revert
)) AS revert_date,
COUNT(*) AS cross_month_updates_enwiki
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND wiki_db = 'enwiki'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
AND SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
GROUP BY 1
ORDER BY cross_month_updates_enwiki DESC
LIMIT 30
""")
print("=== Q5c: Top-30 days by UPDATE volume — enwiki only ===")
q5c.show(30, truncate=False)
# Part D: UPDATE volume by window size
# For each candidate window, how many reverted revisions fall within that window?
# Rows exceeding the window would be missed by the daily writer → still stale.
q5d = spark.sql(f"""
SELECT
wiki_db,
SUM(CASE WHEN revision_seconds_to_identity_revert <= 48 * 3600
THEN 1 ELSE 0 END) AS updates_within_48h,
SUM(CASE WHEN revision_seconds_to_identity_revert <= 7 * 86400
THEN 1 ELSE 0 END) AS updates_within_7d,
SUM(CASE WHEN revision_seconds_to_identity_revert <= 30 * 86400
THEN 1 ELSE 0 END) AS updates_within_30d,
SUM(CASE WHEN revision_seconds_to_identity_revert <= 90 * 86400
THEN 1 ELSE 0 END) AS updates_within_90d,
SUM(CASE WHEN revision_seconds_to_identity_revert <=365 * 86400
THEN 1 ELSE 0 END) AS updates_within_1y,
COUNT(*) AS total_cross_month_reverted
FROM wmf.mediawiki_history
WHERE event_entity = 'revision'
AND snapshot = '{SNAPSHOT}'
AND revision_is_identity_reverted = TRUE
AND revision_seconds_to_identity_revert IS NOT NULL
AND event_timestamp IS NOT NULL
AND SUBSTRING(event_timestamp, 1, 7)
< SUBSTRING(from_unixtime(unix_timestamp(event_timestamp)
+ revision_seconds_to_identity_revert), 1, 7)
GROUP BY 1
ORDER BY total_cross_month_reverted DESC
LIMIT 20
""")
print("=== Q5d: Cross-month reverted rows captured at each window size — top 20 wikis ===")
q5d.show(20, truncate=False)
```
=== Q5a: Per-day UPDATE volume — all wikis ===
+-----------+-----------------------------+
|revert_date|cross_month_updates_all_wikis|
+-----------+-----------------------------+
|2001-02-12 |2 |
|2001-03-01 |1 |
|2001-03-06 |5 |
|2001-04-06 |1 |
|2001-05-12 |1 |
|2001-06-05 |1 |
|2001-06-23 |62 |
|2001-06-29 |1 |
|2001-06-30 |2 |
|2001-07-13 |1 |
|2001-08-01 |1 |
|2001-08-10 |3 |
|2001-10-05 |1 |
|2001-10-09 |1 |
|2001-10-19 |2 |
|2001-11-01 |1 |
|2001-11-02 |2 |
|2001-11-10 |2 |
|2001-11-12 |2 |
|2001-11-23 |1 |
|2001-12-01 |10 |
|2001-12-09 |2 |
|2001-12-10 |1 |
|2001-12-16 |1 |
|2001-12-17 |4 |
|2001-12-24 |3 |
|2002-01-01 |1 |
|2002-01-05 |1 |
|2002-02-01 |8 |
|2002-02-05 |2 |
|2002-02-06 |1 |
|2002-02-22 |1 |
|2002-02-24 |1 |
|2002-02-25 |39 |
|2002-03-08 |2 |
|2002-03-09 |1 |
|2002-03-10 |1 |
|2002-03-13 |6 |
|2002-03-14 |1 |
|2002-03-17 |1 |
|2002-03-24 |2 |
|2002-03-28 |6 |
|2002-04-03 |16 |
|2002-04-04 |3 |
|2002-04-05 |2 |
|2002-04-06 |2 |
|2002-04-07 |2 |
|2002-04-10 |1 |
|2002-04-13 |1 |
|2002-04-15 |3 |
+-----------+-----------------------------+
only showing top 50 rows
=== Q5b: UPDATE volume distribution (all wikis) ===
+--------------+-----------+-----------+-----------+
|median_per_day|p95_per_day|max_per_day|avg_per_day|
+--------------+-----------+-----------+-----------+
|16022 |40783 |1927251 |19039.0 |
+--------------+-----------+-----------+-----------+
=== Q5c: Top-30 days by UPDATE volume — enwiki only ===
+-----------+--------------------------+
|revert_date|cross_month_updates_enwiki|
+-----------+--------------------------+
|2026-04-01 |1018511 |
|2011-02-14 |321248 |
|2015-04-18 |194740 |
|2009-09-05 |153317 |
|2010-10-26 |113788 |
|2010-02-26 |87470 |
|2011-02-15 |79870 |
|2011-06-27 |79619 |
|2017-06-30 |77767 |
|2011-10-03 |75897 |
|2021-10-16 |75883 |
|2022-08-18 |72611 |
|2014-11-27 |71785 |
|2020-06-25 |71134 |
|2023-12-31 |68392 |
|2015-07-12 |66854 |
|2014-07-02 |66446 |
|2009-04-01 |65848 |
|2011-10-13 |62611 |
|2021-02-02 |61901 |
|2020-08-29 |60924 |
|2009-09-08 |55475 |
|2010-09-12 |54287 |
|2007-05-01 |52871 |
|2010-12-30 |52716 |
|2009-01-21 |51739 |
|2013-08-09 |51348 |
|2009-02-04 |49458 |
|2012-04-11 |49457 |
|2007-07-28 |48011 |
+-----------+--------------------------+
=== Q5d: Cross-month reverted rows captured at each window size — top 20 wikis ===
[Stage 110:=================================================>(8178 + 14) / 8192]
+------------+------------------+-----------------+------------------+------------------+-----------------+--------------------------+
|wiki_db |updates_within_48h|updates_within_7d|updates_within_30d|updates_within_90d|updates_within_1y|total_cross_month_reverted|
+------------+------------------+-----------------+------------------+------------------+-----------------+--------------------------+
|enwiki |642108 |2102085 |10079876 |22779833 |42015033 |68245760 |
|dewiki |100572 |373979 |1755067 |3907835 |6771608 |10268169 |
|wikidatawiki|99336 |415083 |2123328 |5569722 |7585863 |9404955 |
|ruwiki |50474 |163911 |924639 |2353989 |4480657 |6465358 |
|frwiki |53843 |181194 |891671 |2020769 |3646531 |5868650 |
|commonswiki |72338 |304849 |1273556 |2482710 |3914606 |5853243 |
|eswiki |89212 |260348 |1107094 |2373221 |3967876 |5545866 |
|itwiki |48019 |141709 |617158 |1446793 |2800593 |4976284 |
|zhwiki |27078 |91509 |491892 |1207419 |2196050 |3550638 |
|jawiki |27240 |91895 |410564 |883262 |1496288 |2572126 |
|hewiki |16285 |68706 |343310 |820986 |1591650 |2482329 |
|plwiki |20060 |70398 |347756 |795702 |1400276 |2351318 |
|nlwiki |16622 |53727 |297166 |748179 |1392537 |2319956 |
|ptwiki |32280 |96684 |383626 |809142 |1356170 |1964540 |
|arwiki |24678 |81662 |387366 |888440 |1496195 |1852063 |
|viwiki |6503 |20201 |108541 |272097 |512761 |1448113 |
|enwiktionary|4399 |12637 |93310 |294902 |713644 |1429330 |
|trwiki |15953 |61759 |278118 |610558 |1061521 |1390276 |
|metawiki |8533 |32712 |193926 |493487 |877144 |1331952 |
|cywiki |175234 |488358 |919052 |1108958 |1176779 |1235001 |
+------------+------------------+-----------------+------------------+------------------+-----------------+--------------------------+
```python
```
Content licensed under Creative Commons Attribution-ShareAlike (CC BY-SA) 4.0 unless otherwise noted; code licensed under GNU General Public License (GPL) 2.0 or later and other open source licenses. By using this site, you agree to the Terms of Use, Privacy Policy, and Code of Conduct. · Wikimedia Foundation · Privacy Policy · Code of Conduct · Terms of Use · Disclaimer · CC-BY-SA · GPL · Credits