# this took 18.9 h
# 1.4h in the suffle, and the rest (17.5h) on the sort, which retried many times
# presumably the sort would only take 7.5h withouth the retries.
# all retries were due to FetchFailedExceptions

spark.sql("""
INSERT INTO xcollazo_iceberg.temporary_wikitext_raw_source_data_from_2023_07

WITH visibility AS (
    SELECT
        rev_id                                        AS v_revision_id,
        wiki_db                                       AS v_wiki_db,
        to_timestamp(rev_timestamp, 'yyyyMMddkkmmss') AS v_revision_timestamp,
        rev_deleted & 1 = 0                           AS v_is_text_visible,
        rev_deleted & 2 = 0                           AS v_is_user_visible,
        rev_deleted & 4 = 0                           AS v_is_comment_visible
    FROM
        wmf_raw.mediawiki_revision
    WHERE
        rev_deleted > 0
        AND snapshot = '2023-07'
)

SELECT /*+ BROADCAST(v) */
    page_id                             AS s_page_id,
    page_namespace                      AS s_page_namespace,
    page_title                          AS s_page_title,
    page_redirect_title                 AS s_page_redirect_title,
    user_id                             AS s_user_id,
    user_text                           AS s_user_text,
    v_is_user_visible                   AS s_user_is_visible,
    revision_id                         AS s_revision_id,
    revision_parent_id                  AS s_revision_parent_id,
    to_timestamp(revision_timestamp)    AS s_revision_timestamp,
    revision_minor_edit                 AS s_revision_is_minor_edit,
    revision_comment                    AS s_revision_comment,
    v_is_comment_visible                AS s_revision_comment_is_visible,
    revision_text_sha1                  AS s_revision_sha1,
    revision_text_bytes                 AS s_revision_size,
    v_is_text_visible                   AS s_revision_content_is_visible,
    MAP(
        'main',
        ( revision_text                 AS content_body,
          revision_content_format       AS content_format,
          revision_content_model        AS content_model,
          revision_text_sha1            AS content_sha1,
          revision_text_bytes           AS content_size
        )
    )                                   AS s_revision_content_slots,
    wiki_db                             AS s_wiki_db
FROM wmf.mediawiki_wikitext_history s
LEFT JOIN visibility v ON s.wiki_db = v_wiki_db AND s.revision_id = v_revision_id
WHERE snapshot = '2023-07'
"""
).collect()

SELECT count(1) FROM xcollazo_iceberg.temporary_wikitext_raw_source_data_from_2023_07
+----------+
|  count(1)|
+----------+
|6252573034|
+----------+

SELECT summary FROM xcollazo_iceberg.temporary_wikitext_raw_source_data_from_2023_07.snapshots
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|summary                                                                                                                                                                                                                                                                                                                                                          |
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|{spark.app.id -> application_1694521537759_11590, changed-partition-count -> 0, total-records -> 0, total-files-size -> 0, total-data-files -> 0, total-delete-files -> 0, total-position-deletes -> 0, total-equality-deletes -> 0}                                                                                                                             |
|{spark.app.id -> application_1694521537759_11590, added-data-files -> 218191, added-records -> 6252573034, added-files-size -> 25552586038578, changed-partition-count -> 176749, total-records -> 6252573034, total-files-size -> 25552586038578, total-data-files -> 218191, total-delete-files -> 0, total-position-deletes -> 0, total-equality-deletes -> 0}|
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

# this run of *all* wikis for the year of 2022 took 2.4hs
# most of that was the final sort (1.3h)
# no evidence of failures.
# this is good evidence that with the intermediary table, we can run the backfill significantly faster

spark.sql("""
MERGE INTO xcollazo_iceberg.wikitext_raw_rc1 t
USING (
    SELECT * FROM (
        SELECT *
        FROM xcollazo_iceberg.temporary_wikitext_raw_source_data_from_2023_07 s
        -- AND wiki_db IN ('simplewiki')
    )
    -- time bound backfill to avoid having a window where we may not have revision visibility data
    WHERE s_revision_timestamp < to_timestamp('2023-07') + INTERVAL 1 MONTH
    AND ( 
s_revision_timestamp >= to_timestamp('2022') AND s_revision_timestamp < to_timestamp('2022') + INTERVAL 1 YEAR )
)

ON  s_wiki_db = t.wiki_db
AND s_revision_id = t.revision_id
-- AND t.wiki_db IN ('simplewiki')
AND ( 
t.revision_timestamp >= to_timestamp('2022') AND t.revision_timestamp < to_timestamp('2022') + INTERVAL 1 YEAR )

WHEN MATCHED AND to_timestamp('2023-07') > t.row_last_update THEN
    UPDATE SET
      t.page_id = s_page_id,
      t.page_namespace = s_page_namespace,
      t.page_title = s_page_title,
      t.page_redirect_title = s_page_redirect_title,
      t.user_id = s_user_id,
      t.user_text = s_user_text,
      t.user_is_visible = COALESCE(s_user_is_visible, TRUE),
      t.revision_id = s_revision_id,
      t.revision_parent_id = s_revision_parent_id,
      t.revision_timestamp = s_revision_timestamp,
      t.revision_is_minor_edit = s_revision_is_minor_edit,
      t.revision_comment = s_revision_comment,
      t.revision_comment_is_visible = COALESCE(s_revision_comment_is_visible, TRUE),
      t.revision_sha1 = s_revision_sha1,                       -- from backfill, revision_sha1 == main slot sha1
      t.revision_size = s_revision_size,                       -- from backfill, revision_size == main slot size
      t.revision_content_slots = s_revision_content_slots,
      t.revision_content_is_visible = COALESCE(s_revision_content_is_visible, TRUE),
      t.wiki_db = s_wiki_db,
      t.row_last_update = to_timestamp('2023-07')
WHEN NOT MATCHED THEN
    INSERT (
        page_id,
        page_namespace,
        page_title,
        page_redirect_title,
        user_id,
        user_text,
        user_is_visible,
        revision_id,
        revision_parent_id,
        revision_timestamp,
        revision_is_minor_edit,
        revision_comment,
        revision_comment_is_visible,
        revision_sha1,
        revision_size,
        revision_content_slots,
        revision_content_is_visible,
        wiki_db,
        row_last_update
    ) VALUES (
      s_page_id,
      s_page_namespace,
      s_page_title,
      s_page_redirect_title,
      s_user_id,
      s_user_text,
      COALESCE(s_user_is_visible, TRUE),
      s_revision_id,
      s_revision_parent_id,
      s_revision_timestamp,
      s_revision_is_minor_edit,
      s_revision_comment,
      COALESCE(s_revision_comment_is_visible, TRUE),
      s_revision_sha1,                                         -- from backfill, revision_sha1 == main slot sha1
      s_revision_size,                                         -- from backfill, revision_size == main slot size
      s_revision_content_slots,
      COALESCE(s_revision_content_is_visible, TRUE),
      s_wiki_db,
      to_timestamp('2023-07')
    )"""
).show(100, truncate=False)

# this seems like a winning config!
spark = wmfdata.spark.create_custom_session(
    master='yarn',
    spark_config={
        "spark.driver.memory": "32g",
        #"spark.dynamicAllocation.maxExecutors": 64,
        #"spark.executor.memory": "26g",
        #"spark.executor.memoryOverhead": "6g",
        "spark.executor.memory": "20g",
        "spark.executor.memoryOverhead": "4g",
        "spark.executor.cores": 2,
        "spark.sql.shuffle.partitions": 131072,
        "spark.shuffle.io.retryWait": "15s",
        "spark.shuffle.io.maxRetries": "15",
        "spark.network.timeout": "600s",

        # maxResultSize default is 1g, and its giviing us problems with MERGE INTO tasks
        "spark.driver.maxResultSize": "8g",

        # extra settings as per https://wikitech.wikimedia.org/wiki/Data_Engineering/Systems/Cluster/Spark#Executors
        "spark.shuffle.file.buffer": "1m",
        "spark.shuffle.service.index.cache.size": "256m",
        "spark.io.compression.lz4.blockSize": "512KB",

        ##
        # extras to make Iceberg work on 3.3.2:
        ##
        "spark.jars.packages": "org.apache.iceberg:iceberg-spark-runtime-3.3_2.12:1.2.1", # bumped to 1.3.1 to test hash
        # we need to clear spark.yarn.archive so that we don't pick up old 3.1.2 one
        # for this I commented out spark.yarn.archive from spark-defaults.conf
        # 3.1 Shuffle service is not happy with Spark 3.3, so have to disable it (and dynamic allocation)
        "spark.shuffle.service.enabled": False,
        "spark.dynamicAllocation.enabled": False,
        "spark.executor.instances": 100,

        # extra settings to mmke adaptive query execution work in our favor
        # "spark.sql.adaptive.coalescePartitions.enabled" : False
    }
)

sudo -u analytics yarn logs -appOwner analytics -applicationId application_1694521537759_58660
...
:::: ERRORS
	SERVER ERROR: Server Error url=https://archiva.wikimedia.org/repository/mirrored/org/apache/iceberg/iceberg-spark-runtime-3.3_2.12/1.2.1/iceberg-spark-runtime-3.3_2.12-1.2.1.pom

	SERVER ERROR: Server Error url=https://archiva.wikimedia.org/repository/mirrored/org/apache/iceberg/iceberg-spark-runtime-3.3_2.12/1.2.1/iceberg-spark-runtime-3.3_2.12-1.2.1.jar
...

√ ~ % curl https://archiva.wikimedia.org/repository/mirrored/org/apache/iceberg/iceberg-spark-runtime-3.3_2.12/1.2.1/iceberg-spark-runtime-3.3_2.12-1.2.1.pom
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=ISO-8859-1"/>
<title>Error 500 Server Error</title>
</head>
<body><h2>HTTP ERROR 500</h2>
<p>Problem accessing /repository/mirrored/org/apache/iceberg/iceberg-spark-runtime-3.3_2.12/1.2.1/iceberg-spark-runtime-3.3_2.12-1.2.1.pom. Reason:
<pre>    Server Error</pre></p><h3>Caused by:</h3><pre>java.lang.RuntimeException: /var/cache/archiva/temp2871579391051196831: No space left on device
	at org.apache.archiva.proxy.DefaultRepositoryProxyConnectors.createWorkingDirectory(DefaultRepositoryProxyConnectors.java:1078)
...

sudo -u analytics yarn logs -appOwner analytics -applicationId application_1694521537759_62173 | grep -v INFO
...
py4j.protocol.Py4JJavaError: An error occurred while calling o79.sql.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 2165 in stage 13.0 failed 4 times, most recent failure: Lost task 2165.3 in stage 13.0 (TID 7534) (analytics1073.eqiad.wmnet executor 45): org.apache.spark.SparkException: 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.
	at org.apache.spark.sql.execution.datasources.v2.MergeRowsExec.processRowWithCardinalityCheck$1(MergeRowsExec.scala:163)
...

MERGE INTO {target_table} t
USING (
    SELECT *
    FROM {intermediate_table}
    -- time bound backfill to avoid having a window where we may not have revision visibility data
    WHERE s_revision_timestamp < to_timestamp('{snapshot}') + INTERVAL 1 MONTH
    AND {generate_timestamp_predicate("s_revision_timestamp", years)}
)

ON  s_wiki_db = t.wiki_db
AND s_revision_id = t.revision_id
AND {generate_timestamp_predicate("t.revision_timestamp", years)}

WHEN MATCHED AND to_timestamp('{snapshot}') > t.row_last_update THEN
...

select s_wiki_db, s_revision_id, count(1) as count
from tmp.tmp_dumps_merge_backfill_to_wikitext_raw_20230701
where s_revision_timestamp >= to_timestamp('2004')
and  s_revision_timestamp < to_timestamp('2005')
group by s_wiki_db, s_revision_id
having count(1) > 1

+---------+-------------+-----+
|s_wiki_db|s_revision_id|count|
+---------+-------------+-----+
|   enwiki|     13123563|    2|
|   enwiki|      6884215|    2|
|   enwiki|      6883838|    2|
|   enwiki|      6884201|    2|
+---------+-------------+-----+

select s_wiki_db, s_revision_id, s_revision_parent_id, s_page_id, s_user_id, s_revision_timestamp, s_revision_sha1
from tmp.tmp_dumps_merge_backfill_to_wikitext_raw_20230701
where s_revision_timestamp >= to_timestamp('2004')
and  s_revision_timestamp < to_timestamp('2005')
and s_wiki_db = 'enwiki'
and s_revision_id = 13123563

+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|s_wiki_db|s_revision_id|s_revision_parent_id|s_page_id|s_user_id|s_revision_timestamp|s_revision_sha1                |
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|enwiki   |13123563     |6884215             |1106422  |13800    |2004-10-26 23:17:08 |kutq8isyomg408cqdygxosjptgv8r00|
|enwiki   |13123563     |6884215             |57286698 |13800    |2004-10-26 23:17:08 |kutq8isyomg408cqdygxosjptgv8r00|
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+



select s_wiki_db, s_revision_id, s_revision_parent_id, s_page_id, s_user_id, s_revision_timestamp, s_revision_sha1
from tmp.tmp_dumps_merge_backfill_to_wikitext_raw_20230701
where s_revision_timestamp >= to_timestamp('2004')
and  s_revision_timestamp < to_timestamp('2005')
and s_wiki_db = 'enwiki'
and s_revision_id = 6884215

+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|s_wiki_db|s_revision_id|s_revision_parent_id|s_page_id|s_user_id|s_revision_timestamp|s_revision_sha1                |
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|enwiki   |6884215      |6884201             |1106422  |13800    |2004-10-26 23:16:23 |0cqof88wc97z2l7nezaar4aw9ypbwst|
|enwiki   |6884215      |6884201             |57286698 |13800    |2004-10-26 23:16:23 |0cqof88wc97z2l7nezaar4aw9ypbwst|
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+



select s_wiki_db, s_revision_id, s_revision_parent_id, s_page_id, s_user_id, s_revision_timestamp, s_revision_sha1
from tmp.tmp_dumps_merge_backfill_to_wikitext_raw_20230701
where s_revision_timestamp >= to_timestamp('2004')
and  s_revision_timestamp < to_timestamp('2005')
and s_wiki_db = 'enwiki'
and s_revision_id = 6883838

+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|s_wiki_db|s_revision_id|s_revision_parent_id|s_page_id|s_user_id|s_revision_timestamp|s_revision_sha1                |
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|enwiki   |6883838      |-1                  |1106422  |-1       |2004-10-26 22:59:23 |fwwkr1vcaz89uqk6bmxsqg80n536sad|
|enwiki   |6883838      |-1                  |57286698 |-1       |2004-10-26 22:59:23 |fwwkr1vcaz89uqk6bmxsqg80n536sad|
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+



select s_wiki_db, s_revision_id, s_revision_parent_id, s_page_id, s_user_id, s_revision_timestamp, s_revision_sha1
from tmp.tmp_dumps_merge_backfill_to_wikitext_raw_20230701
where s_revision_timestamp >= to_timestamp('2004')
and  s_revision_timestamp < to_timestamp('2005')
and s_wiki_db = 'enwiki'
and s_revision_id = 6884201

+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|s_wiki_db|s_revision_id|s_revision_parent_id|s_page_id|s_user_id|s_revision_timestamp|s_revision_sha1                |
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+
|enwiki   |6884201      |6883838             |1106422  |-1       |2004-10-26 23:00:52 |7yytbakswgc26dgvu1dnc63qkozly6q|
|enwiki   |6884201      |6883838             |57286698 |-1       |2004-10-26 23:00:52 |7yytbakswgc26dgvu1dnc63qkozly6q|
+---------+-------------+--------------------+---------+---------+--------------------+-------------------------------+

presto> select rev_id, wiki_db, rev_page, rev_user, rev_sha1 from wmf_raw.mediawiki_revision where snapshot = '2023-08' and wiki_db = 'enwiki' and rev_id IN (6883838, 6884201, 6884215, 13123563);
  rev_id  | wiki_db | rev_page | rev_user |            rev_sha1             
----------+---------+----------+----------+---------------------------------
 13123563 | enwiki  | 57286698 | NULL     | kutq8isyomg408cqdygxosjptgv8r00 
  6883838 | enwiki  | 57286698 | NULL     | fwwkr1vcaz89uqk6bmxsqg80n536sad 
  6884201 | enwiki  | 57286698 | NULL     | 7yytbakswgc26dgvu1dnc63qkozly6q 
  6884215 | enwiki  | 57286698 | NULL     | 0cqof88wc97z2l7nezaar4aw9ypbwst 
(4 rows)

+- == Initial Plan ==
   Sort (46)
   +- Exchange (45)
      +- Project (44)
         +- MergeRows (43)
            +- Sort (42)
               +- SortMergeJoin FullOuter (41)
                  :- Sort (26)
                  :  +- Exchange (25)
                  :     +- Project (24)
                  :        +- Project (23)
                  :           +- BatchScan (22)
                  +- Sort (40)
                     +- Exchange (39)
                        +- Project (38)
                           +- BroadcastHashJoin LeftAnti BuildRight (37)
                              :- Project (27)
                              :  +- BatchScan (7)
                              +- BroadcastExchange (36)
                                 +- Project (35)
                                    +- Filter (34)
                                       +- Window (33)
                                          +- Sort (32)
                                             +- Exchange (31)
                                                +- Project (30)
                                                   +- Filter (29)
                                                      +- BatchScan (28)