SELECT c.wiki_id,
       c.revision.rev_id,
       c.changelog_kind,
       v.prior_state.visibility as prior_visibility,
       v.visibility,
       c.meta.dt as content_meta_dt,
       v.meta.dt as visibility_meta_dt,
       to_timestamp(c.meta.dt) < to_timestamp(v.meta.dt) as v_after_c
FROM event.mediawiki_page_content_change_v1 c
INNER JOIN event.mediawiki_revision_visibility_change v ON c.wiki_id = v.database AND c.revision.rev_id = v.rev_id
WHERE c.year = 2023
  AND c.month = 10
  AND c.day = 16
  AND c.hour = 0
  
  AND v.year = 2023
  AND v.month = 10
  AND v.day = 16
  AND v.hour = 0

+-------+----------+--------------+------------------+--------------------+---------------------------+--------------------+---------+
|wiki_id|rev_id    |changelog_kind|prior_visibility  |visibility          |content_meta_dt            |visibility_meta_dt  |v_after_c|
+-------+----------+--------------+------------------+--------------------+---------------------------+--------------------+---------+
|enwiki |1180332207|update        |{true, true, true}|{false, false, true}|2023-10-16T00:00:07.573767Z|2023-10-16T00:01:34Z|true     |
|bewiki |4605967   |update        |{true, true, true}|{false, true, false}|2023-10-16T00:16:43.691925Z|2023-10-16T00:16:42Z|false    |
|ruwiki |133622800 |update        |{true, true, true}|{false, true, true} |2023-10-16T00:05:29.567523Z|2023-10-16T00:05:26Z|false    |
|ruwiki |133622800 |update        |{true, true, true}|{false, true, true} |2023-10-16T00:04:56.575687Z|2023-10-16T00:05:26Z|true     |
|bewiki |4605967   |update        |{true, true, true}|{false, true, false}|2023-10-16T00:06:50.596076Z|2023-10-16T00:16:42Z|true     |
+-------+----------+--------------+------------------+--------------------+---------------------------+--------------------+---------+

errors    ARRAY<
             STRUCT<error_timestamp: TIMESTAMP,
                    error_producer: STRING,
                    error_message: STRING,
                    error_severity: STRING
             >
          >        COMMENT 'an ARRAY, typically empty, that will contain any detected ingestions errors. Consumers may want to skip rows with errors depending on their severity.'

...
WHEN MATCHED AND s.changelog_kind = 'insert'
  UPDATE SET
    t.page_id = s_page_id,
    ...
    t.errors = concat(t.errors, array(named_struct(
                                         'error_timestamp', NOW(),
                                         'error_producer', 'events_merge_into.py',
                                         'error_message', '(wiki_db, revision_id) existed, but got a changelog_kind = insert',
                                         'error_severity', 'WARN'
                                         )
                                 )
                     ),
    ...
    t.row_last_update = s_meta_dt

SELECT *
FROM wmf_dumps.wkitext_raw_rc2
WHERE ....
AND cardinality(errors) = 0

SELECT *
FROM wmf_dumps.wkitext_raw_rc2
WHERE ....
AND cardinality(errors) > 0

xcollazo@stat1007:~$ hdfs dfs -count /user/hive/warehouse/wmf_dumps.db/wikitext_raw_rc1/data
Picked up JAVA_TOOL_OPTIONS: -Dfile.encoding=UTF-8
     3569878      5056902     44264942677765 /user/hive/warehouse/wmf_dumps.db/wikitext_raw_rc1/data

SELECT count(1)*8 as timestamp_total_bytes, sum(revision_content_slots['main'].content_size) as content_total_bytes FROM wmf_dumps.wikitext_raw_rc1 WHERE wiki_db = 'eswiki';
timestamp_total_bytes	content_total_bytes
1141134984	3197064784033

ssh an-coord1001.eqiad.wmnet
sudo -u analytics bash

kerberos-run-command analytics spark3-sql

CREATE TABLE wmf_dumps.wikitext_raw_rc2 (
    page_id                     BIGINT    COMMENT 'id of the page',
    page_namespace              INT       COMMENT 'namespace of the page',
    page_title                  STRING    COMMENT 'title of the page ',
    page_redirect_title         STRING    COMMENT 'title of the redirected-to page',
    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 details have been suppressed or not',
    revision_id                 BIGINT    COMMENT 'id of the revision',
    revision_parent_id          BIGINT    COMMENT 'id of the parent revision',
    revision_timestamp          TIMESTAMP COMMENT 'timestamp of the revision',
    revision_is_minor_edit      BOOLEAN   COMMENT 'whether this revision is a minor edit or not',
    revision_comment            STRING    COMMENT 'comment made with revision',
    revision_comment_is_visible BOOLEAN   COMMENT 'whether the comment has been suppressed or not',
    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
                                    >
                                >         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 the content in the slots has been suppressed or not',
    wiki_db                     STRING    COMMENT 'the wiki project',
    errors                      ARRAY<
                                    STRUCT<error_upstream_id:           STRING,     -- the id of the event that caused the error, typically meta.id.
                                           error_upstream_timestamp:    TIMESTAMP,  -- the timestamp of the event that caused the error, typically meta.dt.
                                           error_producer:              STRING,
                                           error_message:               STRING,
                                           error_severity:              STRING
                                    >
                                >         COMMENT 'an ARRAY, typically empty, that will contain any detected ingestion errors. Consumers may want to skip rows with errors depending on their severity.',
    row_last_update             TIMESTAMP COMMENT 'the timestamp of the last content event or backfill that updated this row',
    row_visibility_last_update  TIMESTAMP COMMENT 'the timestamp of the last visibility event or backfill that updated this row'
)
USING ICEBERG
PARTITIONED BY (wiki_db, months(revision_timestamp)) -- wiki_db partitioning helps the backfill process,
                                                     -- while months(revision_timestamp) helps the stream intake
TBLPROPERTIES (
    'format-version' = '2',                                -- allow merge-on-read if needed
    'write.format.default' = 'avro',                       -- avoid OOMs due to revision_content_slots
    'write.metadata.previous-versions-max' = '10',
    'write.metadata.delete-after-commit.enabled' = 'true'
)
COMMENT 'Base to create MediaWiki full revision dumps from.'
LOCATION '/wmf/data/wmf_dumps/wikitext_raw_rc2';

Response code
Time taken: 3.388 seconds

$ hdfs dfs -count /wmf/data/wmf_dumps/wikitext_raw_rc2/data
      179324       220730     26078477110787 /wmf/data/wmf_dumps/wikitext_raw_rc2/data

$ hdfs dfs -count /user/hive/warehouse/wmf_dumps.db/wikitext_raw_rc1/data
     3574753      5127198     48903124525321 /user/hive/warehouse/wmf_dumps.db/wikitext_raw_rc1/data

# Are parquet files as we expect them to be on 2023-10 ?
spark.sql(
"""
SELECT
file_format,
partition.revision_timestamp_month,
readable_metrics['revision_timestamp']['lower_bound'] as lower_bound,
readable_metrics['revision_timestamp']['upper_bound'] as upper_bound
FROM wmf_dumps.wikitext_raw_rc2.data_files
WHERE partition.wiki_db = 'enwiki'
  AND partition.revision_timestamp_month = 645
ORDER BY lower_bound ASC
""").show(100, truncate=False)


+-----------+------------------------+-------------------+-------------------+
|file_format|revision_timestamp_month|lower_bound        |upper_bound        |
+-----------+------------------------+-------------------+-------------------+
|PARQUET    |645                     |2023-10-01 00:00:00|2023-10-31 23:59:37|
|PARQUET    |645                     |2023-10-01 00:00:01|2023-10-31 23:59:16|
|PARQUET    |645                     |2023-10-01 00:00:01|2023-10-31 23:59:31|
|PARQUET    |645                     |2023-10-01 00:00:01|2023-10-31 23:59:49|
|PARQUET    |645                     |2023-10-01 00:00:01|2023-10-31 23:59:57|
|PARQUET    |645                     |2023-10-01 00:00:01|2023-10-31 23:59:57|
|PARQUET    |645                     |2023-10-01 00:00:02|2023-10-31 23:59:54|
|PARQUET    |645                     |2023-10-01 00:00:02|2023-10-31 23:59:54|
|PARQUET    |645                     |2023-10-01 00:00:03|2023-10-31 23:59:52|
|PARQUET    |645                     |2023-10-01 00:00:03|2023-10-31 23:59:50|
|PARQUET    |645                     |2023-10-01 00:00:04|2023-10-31 23:59:07|
|PARQUET    |645                     |2023-10-01 00:00:05|2023-10-31 23:59:24|
|PARQUET    |645                     |2023-10-01 00:00:06|2023-10-31 23:59:45|
|PARQUET    |645                     |2023-10-01 00:00:06|2023-10-31 23:59:33|
|PARQUET    |645                     |2023-10-01 00:00:06|2023-10-31 23:58:35|
|PARQUET    |645                     |2023-10-01 00:00:07|2023-10-31 23:59:42|
|PARQUET    |645                     |2023-10-01 00:00:07|2023-10-31 23:59:11|
...

ALTER TABLE wmf_dumps.wikitext_raw_rc2 SET TBLPROPERTIES (
    'write.distribution-mode'='range'
)

spark-sql (default)> DELETE FROM wmf_dumps.wikitext_raw_rc2;
Response code
Time taken: 8.904 seconds

Table Properties	[current-snapshot-id=7905477679598267939,format=iceberg/parquet,write.distribution-mode=range,write.format.default=parquet,write.merge.mode=copy-on-write,write.metadata.delete-after-commit.enabled=true,write.metadata.previous-versions-max=10]

spark-sql (default)> DELETE FROM wmf_dumps.wikitext_raw_rc2;
Response code
Time taken: 9.864 seconds

spark-sql (default)> ALTER TABLE wmf_dumps.wikitext_raw_rc2 SET TBLPROPERTIES (
                   >     'write.target-file-size-bytes' = '134217728'
                   > );
23/12/11 20:45:27 WARN BaseTransaction: Failed to load metadata for a committed snapshot, skipping clean-up
Response code
Time taken: 1.177 seconds

spark-sql (default)> ALTER TABLE wmf_dumps.wikitext_raw_rc2 WRITE ORDERED BY wiki_db, revision_timestamp;
23/12/11 20:46:26 WARN BaseTransaction: Failed to load metadata for a committed snapshot, skipping clean-up
Response code
Time taken: 0.737 seconds

Table Properties	[current-snapshot-id=7198391926576962044,format=iceberg/parquet,sort-order=wiki_db ASC NULLS FIRST, revision_timestamp ASC NULLS FIRST,write.distribution-mode=range,write.format.default=parquet,write.merge.distribution-mode=range,write.merge.mode=copy-on-write,write.metadata.delete-after-commit.enabled=true,write.metadata.previous-versions-max=10,write.target-file-size-bytes=134217728]

(3) Exchange
Input [21]: [page_id#105L, page_namespace#106, page_title#107, page_redirect_title#108, user_id#109L, user_text#110, user_is_visible#111, revision_id#112L, revision_parent_id#113L, revision_timestamp#114, revision_is_minor_edit#115, revision_comment#116, revision_comment_is_visible#117, revision_sha1#118, revision_size#119L, revision_content_slots#120, revision_content_is_visible#121, wiki_db#122, errors#123, row_last_update#124, row_visibility_last_update#125]
Arguments: rangepartitioning(wiki_db#122 ASC NULLS FIRST, icebergmonthtransform(revision_timestamp#114) ASC NULLS FIRST, 200), REPARTITION_BY_NUM, [plan_id=65]

ALTER TABLE wmf_dumps.wikitext_raw_rc2 WRITE ORDERED BY wiki_db, revision_timestamp;

(3) Exchange
Input [21]: [page_id#1144L, page_namespace#1145, page_title#1146, page_redirect_title#1147, user_id#1148L, user_text#1149, user_is_visible#1150, revision_id#1151L, revision_parent_id#1152L, revision_timestamp#1153, revision_is_minor_edit#1154, revision_comment#1155, revision_comment_is_visible#1156, revision_sha1#1157, revision_size#1158L, revision_content_slots#1159, revision_content_is_visible#1160, wiki_db#1161, errors#1162, row_last_update#1163, row_visibility_last_update#1164]
Arguments: rangepartitioning(wiki_db#1161 ASC NULLS FIRST, revision_timestamp#1153 ASC NULLS FIRST, 200), REPARTITION_BY_NUM, [plan_id=514]

# how does data look?
spark.sql(
"""
SELECT
file_format,
partition.revision_timestamp_month,
readable_metrics['revision_timestamp']['lower_bound'] as lower_bound,
readable_metrics['revision_timestamp']['upper_bound'] as upper_bound
FROM xcollazo_iceberg.wikitext_raw_rc2_sorted.data_files
WHERE partition.wiki_db = 'enwiki'
  AND partition.revision_timestamp_month = 512
ORDER BY lower_bound ASC
""").show(500, truncate=False)

+-----------+------------------------+-------------------+-------------------+
|file_format|revision_timestamp_month|lower_bound        |upper_bound        |
+-----------+------------------------+-------------------+-------------------+
|PARQUET    |512                     |2012-09-01 00:00:01|2012-09-01 02:59:04|
|PARQUET    |512                     |2012-09-01 02:59:05|2012-09-01 03:54:51|
|PARQUET    |512                     |2012-09-01 03:54:52|2012-09-01 08:16:20|
|PARQUET    |512                     |2012-09-01 08:16:20|2012-09-01 08:38:24|
|PARQUET    |512                     |2012-09-01 08:38:25|2012-09-01 12:46:49|
|PARQUET    |512                     |2012-09-01 12:46:49|2012-09-01 12:51:56|
|PARQUET    |512                     |2012-09-01 12:51:57|2012-09-01 15:57:00|
|PARQUET    |512                     |2012-09-01 15:57:02|2012-09-01 16:31:00|
|PARQUET    |512                     |2012-09-01 16:31:01|2012-09-01 19:09:12|
|PARQUET    |512                     |2012-09-01 19:09:13|2012-09-01 19:49:26|
|PARQUET    |512                     |2012-09-01 19:49:27|2012-09-01 22:15:28|
|PARQUET    |512                     |2012-09-01 22:15:28|2012-09-01 23:38:38|
|PARQUET    |512                     |2012-09-01 23:38:39|2012-09-02 03:01:36|
...

# average file size per year
spark.sql(
"""
SELECT
avg(ceil(file_size_in_bytes/1024/1024)) as file_size_in_megas,
count(1) as count,
year(readable_metrics['revision_timestamp']['lower_bound']) as year
FROM wmf_dumps.wikitext_raw_rc2.data_files
WHERE partition.wiki_db = 'enwiki'
GROUP BY year
ORDER BY year ASC
""").show(500, truncate=False)

+------------------+-----+----+
|file_size_in_megas|count|year|
+------------------+-----+----+
|1.0               |54490|2001|
|1.0002437166793603|98475|2002|
|1.0               |64442|2003|
|1.0029809937911784|52667|2004|
|2.348815797206496 |51047|2005|
|6.299488914483156 |55568|2006|
|8.777622351184288 |52901|2007|
|9.898963090976423 |48606|2008|
|10.824583353005051|42362|2009|
|11.284475703324809|39100|2010|
|11.68925300927259 |34726|2011|
|12.751881866811877|31219|2012|
|17.596499373229953|21539|2013|
|17.657454858949183|20879|2014|
|19.38580321382039 |20723|2015|
|21.85241357669578 |19121|2016|
|25.849336362549323|16726|2017|
|28.03143227478937 |15430|2018|
|32.842958459979734|13818|2019|
|39.90892011723153 |13307|2020|
|34.52726004818055 |15774|2021|
|34.276945630696666|15689|2022|
|92.35712816292944 |4861 |2023|   <<<< Only year run with 32,768 spark partitions
+------------------+-----+----+

# average file size per year for all wikis
spark.sql(
"""
SELECT
ceil(avg(file_size_in_bytes/1024/1024)) as avg_file_size_in_megas,
count(1) as count,
year(readable_metrics['revision_timestamp']['lower_bound']) as year
FROM wmf_dumps.wikitext_raw_rc2.data_files
GROUP BY year
ORDER BY year ASC
""").show(500, truncate=False)

+----------------------+-----+----+
|avg_file_size_in_megas|count|year|
+----------------------+-----+----+
|1                     |4949 |2001|
|1                     |5604 |2002|
|1                     |6317 |2003|
|5                     |9366 |2004|
|17                    |11697|2005|
|37                    |15684|2006|
|47                    |17576|2007|
|50                    |18283|2008|
|50                    |19199|2009|
|51                    |19549|2010|
|50                    |19576|2011|
|51                    |19690|2012|
|51                    |19768|2013|
|48                    |19609|2014|
|52                    |20836|2015|
|58                    |20917|2016|
|62                    |22014|2017|
|70                    |24853|2018|
|74                    |26436|2019|
|73                    |27177|2020|
|73                    |26477|2021|
|78                    |29566|2022|
|75                    |24634|2023|
+----------------------+-----+----+

# total files for the current snapshot
spark.sql(
"""
SELECT
count(1) as count
FROM wmf_dumps.wikitext_raw_rc2.data_files 
""").show(500, truncate=False)

+------+
|count |
+------+
|429777|
+------+

spark = wmfdata.spark.create_custom_session(
    master='yarn',
    spark_config={
        "spark.shuffle.service.name": 'spark_shuffle_3_3',
        "spark.shuffle.service.port": '7338',
        "spark.yarn.archive": "hdfs:///user/spark/share/lib/spark-3.3.2-assembly.zip",
        "spark.dynamicAllocation.maxExecutors": 128,
        "spark.driver.memory": "16g",
        "spark.executor.memory": "16g",
        ##
        # 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",
        "spark.jars.ivySettings": "/etc/maven/ivysettings.xml",  # fix jar pulling
    }
)

# let's go all in: force read all content from all wikis from all time!
# .. and do it on 1 core, 16G executors!
spark.sql(
"""
SELECT count(1)
FROM (
    SELECT length(revision_content_slots['main'].content_body) as actual_length,
           revision_content_slots['main'].content_size as advertised_length
    FROM wmf_dumps.wikitext_raw_rc2
)
WHERE actual_length != advertised_length
""").show(100, truncate=False)

+----------+
|count(1)  |
+----------+
|5530001645|
+----------+