In [2]:
artifact_path = "hdfs:///user/dcausse/rdf-spark-tools-0.3.138-SNAPSHOT-spark-free.jar"
base_analysis_path = "hdfs:///user/dcausse/T352538_wdqs_graph_split_eval"
tools_csv_path = base_analysis_path + "/blazegraph_queries_sample_2023_12_21.csv"
In [ ]:
import pandas as pd
import wmfdata
from pyspark.sql.functions import *
from pyspark.sql.types import StructType, StructField, StringType, LongType, Row, ArrayType
from pyspark.sql import Window

import re

# Or get a totally customizable SparkSession using get_custom_session.
spark = wmfdata.spark.create_custom_session(
    master='yarn',
    spark_config={
        # 16g because we have many partition
        "spark.jars": artifact_path,
        "spark.driver.memory": "16g",
        "spark.driver.cores": 1,
        "spark.executor.memory": "16g",
        # XXX: Always set these values to something low: 5 to 10 concurrent queries at most! (beware of timeouts!)
        "spark.executor.cores": 1,
        "spark.dynamicAllocation.maxExecutors": 5,
        "spark.executor.memoryOverhead": "8g",
        "spark.sql.shuffle.partitions": 5,
        'spark.locality.wait': '1s' # test 0
    }
)
spark.sparkContext.setCheckpointDir("hdfs:///user/dcausse/spark_checkpoints/wdqs_query_results_analysis")
In [ ]:
""" regen scholarly items
# extract scholarly article q items
schol = spark.table('discovery.wikibase_rdf_scholarly_split').where('wiki="wikidata" AND `snapshot` = "20231016" AND scope="scholarly_articles"').alias("schol");
(schol
 .select(regexp_extract('context', r'/entity/Q(\d+)>$', 1).cast(LongType()).alias("item"))
 .filter("item is not null")
 .distinct()
 .repartition(10)
 .write
 .mode("overwrite")
 .parquet(base_analysis_path + "/schol_Q_items.parquet"))
"""
In [5]:
schol_Q_items = spark.read.parquet(base_analysis_path + "/schol_Q_items.parquet")
In [6]:
def query_recorder(endpoint, ua_suffix, spark):
    qr = spark.sparkContext._jvm.org.wikidata.query.rdf.spark.metrics.queries.QueryResultRecorder.create(endpoint, ua_suffix)
    from pyspark.sql.column import Column, _to_java_column
    def run(col):
        return Column(qr.apply(_to_java_column(col)))
    return run

run_wdqs_sparql_endpoint_wikidata = query_recorder("http://wdqs1024.eqiad.wmnet/sparql", "WDQS Split Analysis T352538", spark)
run_wdqs_sparql_endpoint_schol = query_recorder("http://wdqs1023.eqiad.wmnet/sparql", "WDQS Split Analysis T352538", spark)
run_wdqs_sparql_endpoint_full = query_recorder("http://wdqs1022.eqiad.wmnet/sparql", "WDQS Split Analysis T352538", spark)

"""from pyspark.sql.types import StructType, StructField, StringType, LongType, Row


inputType = StructType([StructField(name = "id", dataType = StringType()),
                        StructField(name = "query", dataType = StringType()),
                        StructField(name = "prov", dataType = StringType())])
inputDf = spark.createDataFrame([
    Row("id1", "SELECT * { ?s ?p ?o £ } LIMIT 1", "provenance 1"),
    Row("id2", "SELECT * { ?s wdt:P31 wd:Q5 . } LIMIT 5", "provenance 1"),
    Row("id3", "SELECT * { ?s wdt:P31 wd:Q18918145 . } LIMIT 10", "provenance 2"),
    Row("id4", "SELECT * { ?s wdt:P31 wd:Q523 . } LIMIT 12", "provenance 2")
], inputType)
"""

def read_queries(tool):
    return (spark.read
            .option("multiline", True)
            .option("quote", '"')
            .option("escape", '"')
            .csv(tools_csv_path, header=True)
            .filter(col("tool") == tool))

def exec_queries(input_queries, wdqs_endpoint, wdqs_endpoint_name, in_parts=30, out_parts=1, query_col="query"):
    return (input_queries.join(input_queries
                        .select(query_col)
                        .distinct()
                        .repartition(in_parts)
                        .withColumn("results", wdqs_endpoint(input_queries[query_col]))
                        .select("results.*", query_col)
                        .withColumn("endpoint", lit(wdqs_endpoint_name)),
                        query_col))


def save_results(input_queries, wdqs_endpoint, wdqs_endpoint_name, output_path, query_col="query", mode="overwrite", in_parts=30, out_parts=1):
    (exec_queries(input_queries, wdqs_endpoint, wdqs_endpoint_name, in_parts, query_col)
            .repartition(out_parts)
            .write
            .mode(mode)
            .parquet(output_path))
    return spark.read.parquet(output_path)
In [7]:
from IPython.display import display, HTML

def panda_style_context():
    opts = [
        'styler.format.precision', 2,
        'styler.format.thousands', ' ',
        'display.max_colwidth', None
    ]
    return pd.option_context(*opts)


def to_html_table(df, nb_rows=100):
    output = "<table><tr>"
    for f in df.schema.fields:
        output += "<th style=\"text-align: left\">" + f.name + "</th>"
    output += "</tr>"
    for r in df.take(nb_rows):
        output += "<tr>"
        for v in r:
            output += "<td style=\"text-align: left\"><pre>"+v+"</pre></td>"
        output += "</tr>"
    output += "</table>"
    return HTML(output)
In [8]:
# Classification of sparql queries

def extract_qitems(query):
    return [int(q) for q in re.findall(r'wd:Q(\d+)', query)] 
        
def extract_results_qitems(results):
    if results is None:
        return []
    items = set()
    for r in results:
        for k, v in r.items():
            items |= set([int(q) for q in re.findall(r'Q(\d+)', str(v))])
            
    return list(items)

def q_classify(query, resultSize):
    classes = []
    if re.search(r'LIMIT\s+\d+.*OFFSET\s+\d+', query, flags=re.DOTALL|re.IGNORECASE):
        classes.append("limit_offset") # LIMIT X OFFSET Y queries are likely to yield different results
    if re.search(r'SERVICE\s+wikibase\:mwapi.*bd\:serviceParam\s+wikibase\:api\s+"EntitySearch"', query, re.DOTALL):
        classes.append("EntitySearch") # relying on the search API might yield different results
    if re.search(r'bd\:slice.offset\s+\d+', query):
        classes.append("slice_offset") # relying on bd:slice.offset might yield different results
    if re.search(r'GROUP_CONCAT', query, flags=re.IGNORECASE):
        classes.append("group_concat") # group_concat doesn't support ordering https://github.com/w3c/sparql-dev/issues/9
    if re.search(r'SAMPLE\s*\(\s*[?{]', query, flags=re.IGNORECASE):
        classes.append("sample") # sample is not deterministic
    if re.search('SERVICE\s+wikibase:label', query, flags=re.IGNORECASE) and re.search('\?[A-Za-z0-9_]+AltLabel[^\w]+', query):
        classes.append("lbl_srv_alt_label") # the label service has to pickup one alias
    if resultSize is not None:
        top_level_limit = re.search(r'LIMIT\s+(\d+)[^}]*$', query, flags=re.DOTALL|re.IGNORECASE)
        if top_level_limit and int(top_level_limit.group(1)) == resultSize:
            classes.append("top_level_limit")
    return classes

def ua_classify(ua):
    classes = []
    if re.match(r'AuthorBot Pywikibot/.*', ua):
        classes.append("bot:authorbot") # Mike Peel? (https://phabricator.wikimedia.org/T300207)
    if 'User:Research Bot' in ua:
        classes.append("bot:research_bot") # Daniel Mietchen via https://www.wikidata.org/wiki/User:Research_Bot
    if 'User:Ulmus pumila' in ua:
        classes.append("bot:research_bot") # User:Ulmus pumila is unknown searching on wiki, it's always tagged with User:Research Bot 
    return classes

def error_classify(error):
    if error is None:
        return None
    if 'java.util.concurrent.TimeoutException' in error:
        return "timeout"
    if re.search(r'Caused by: java\.lang\.IllegalArgumentException: Buffering capacity \d+ exceeded', error):
        return "response_too_big"
    if re.search(r'java\.io\.IOException: Unkown record type: \d+', error):
        return "timeout" # likely a timeout exception printed in the middle of the binary response
    return "unclassified"


uq_classify = udf(q_classify, ArrayType(StringType()))
ua_classify = udf(ua_classify, ArrayType(StringType()))
uerror_classify = udf(error_classify, StringType())
uextract_qitems = udf(extract_qitems, ArrayType(LongType()))
uextract_results_qitems = udf(extract_results_qitems, ArrayType(LongType()))

def identify_scientific_queries(queries_df, schol_arts_df):
    return queries_df.join(queries_df
     .select("query")
     .dropDuplicates()
     .withColumn("qitem", explode(uextract_qitems(col("query"))))
     .join(schol_arts_df, col("item") == col("qitem"))
     .withColumn("query_with_direct_sa_ref", lit(True))
     .select("query", "query_with_direct_sa_ref")
     .dropDuplicates(), "query", "leftouter").checkpoint(eager=True)

def identify_scientific_results(querie_results_df, schol_arts_df):
    qitem_exploded = (querie_results_df
                  .select("tool", "query", "results")
                  .dropDuplicates(['tool', 'query'])
                  .withColumn("qitem", explode(uextract_results_qitems(col("results"))))
                  .drop("results")).checkpoint(eager=True)
    
    qitems_joined = (qitem_exploded
                  .join(schol_arts_df, col("item") == col("qitem"))
                  .select("tool", "query")
                  .dropDuplicates()
                  .withColumn("results_with_direct_sa_ref", lit(True))).checkpoint(eager=True)
    return (querie_results_df
            .join(qitems_joined, ["tool", "query"], "leftouter")).checkpoint(eager=True)

def classify(r, df_schol_items, with_scientific_results_analysis=False):
    """
    classify rows with:
    - query_with_direct_sa_ref a boolean to true
    r: input DataFrame with queries and results
    df_schol_items:
    Returns: augmented dataframe
    """
    r = identify_scientific_queries(r, df_schol_items)
    if with_scientific_results_analysis:
        r = identify_scientific_results(r, df_schol_items)
    else:
        r = r.withColumn("results_with_direct_sa_ref", lit(False))
    return (r
            .withColumn("error_class", uerror_classify(col("error_msg")))
            .withColumn("query_classes", uq_classify(col("query"), col("resultSize")))
            .withColumn("ua_classes", ua_classify(col("user_agent"))))

def drop_dups(df):
    window = Window.partitionBy("tool", "query").orderBy(col("id").asc())
    return (df
            .withColumn("rank", rank().over(window))
            .filter("rank = 1")
            .drop("rank"))
In [9]:
"""
# Classification is increasidibly slow... only compute it once and save it
wikidata = spark.read.parquet(base_analysis_path + "/tools_wikidata_Pywikibot.parquet")
for tool in ["MixNMatch", "Listeria", "SPARQLWrapper", "WikidataIntegrator"]:
    wikidata = wikidata.union(spark.read.parquet(base_analysis_path + "/tools_wikidata_"+tool+".parquet"))

(classify(wikidata, schol_Q_items, with_scientific_results_analysis=False)
 .repartition(5)
 .write
 .mode("overwrite")
 .parquet(base_analysis_path + "/wikidata_classified.parquet"))
"""
wikidata = spark.read.parquet(base_analysis_path + "/wikidata_classified.parquet")
In [10]:
"""
# Classification is increasidibly slow... only compute it once and save it

full = spark.read.parquet(base_analysis_path + "/tools_full_Pywikibot.parquet")
for tool in ["MixNMatch", "Listeria", "SPARQLWrapper", "WikidataIntegrator"]:
    full = full.union(spark.read.parquet(base_analysis_path + "/tools_full_"+tool+".parquet"))

(classify(full, schol_Q_items, with_scientific_results_analysis=True)
 .repartition(5)
 .write
 .mode("overwrite")
 .parquet(base_analysis_path + "/full_classified.parquet"))
"""
full = spark.read.parquet(base_analysis_path + "/full_classified.parquet")

Basic statistics¶

How the queries were extracted is explained at [link to andrew's notebook].

The table below shows the number of queries in the sample group by *tool*:

  • Total: total number of queries
  • Unique: total number of unique queries
In [32]:
def simple_stats(df):
    return (df
            .groupBy("tool")
            .agg(
                sum(lit(1)).alias("Total"),
                countDistinct("query").alias("Unique"),
            )
            .fillna(0)
            .sort(col("tool").asc()))

with panda_style_context():
    display(simple_stats(full).toPandas())
tool Total Unique
0 Listeria 10000 101
1 MixNMatch 8396 626
2 Pywikibot 10000 5455
3 SPARQLWrapper 10000 9919
4 WikidataIntegrator 10000 2507

Error statistics¶

Here we compute the number of queries that ran successfully:

  • originally: when they hit production WDQS servers did they yield a HTTP 200 response
  • full: when they run on the full graph
  • wikidata: when they run on the wikidata main subbraph

Worth noting that listeria queries did perform very badly originally.

Exploring HTTP status:

  • 200: success
  • 400: likely a syntax error
  • 403: the code returned by WDQS when the user is banned by the throttling mechanism and not honoring the retry-after header
  • 429: too many requests, user is throttled
  • 500: server error, e.g. timeout

We understand that most Listeria queries did fail originally because because the IP was banned.

In [12]:
def error_stats(full, wikidata):
    wikidata_errors = (wikidata
                       .groupBy("tool")
                       .agg(
                           sum(col("success").cast(LongType())).alias("Total successful (wikidata)"),
                       )
                       .fillna(0))

    return (full
     .groupBy("tool")
     .agg(
         sum(lit(1)).alias("Total"),
         sum((col("http_status") == lit(200)).cast(LongType())).alias("Total successful (originally)"),
         sum(col("success").cast(LongType())).alias("Total successful (full)")
     )
     .fillna(0)
     .join(wikidata_errors, "tool")
     .sort(col("tool").asc()))
    
def http_status_stats(df):
    return (df
     .groupBy("tool")
     .pivot("http_status")
     .count()
     .fillna(0)
     .sort(col("tool").asc()))
     
error_pd = error_stats(full, wikidata).toPandas()
for c in ["successful (originally)", "successful (full)", "successful (wikidata)"]: 
    error_pd["% " + c] = error_pd["Total " + c]/error_pd["Total"]*100

with panda_style_context():
    display(error_pd.style.set_caption("Success original vs full vs wikidata"))
    display(http_status_stats(full).toPandas().style.set_caption("Number of queries per original http status"))
Success original vs full vs wikidata
  tool Total Total successful (originally) Total successful (full) Total successful (wikidata) % successful (originally) % successful (full) % successful (wikidata)
0 Listeria 10 000 133 9 466 9 466 1.33 94.66 94.66
1 MixNMatch 8 396 7 477 8 369 8 369 89.05 99.68 99.68
2 Pywikibot 10 000 9 812 9 949 9 951 98.12 99.49 99.51
3 SPARQLWrapper 10 000 9 864 9 907 9 912 98.64 99.07 99.12
4 WikidataIntegrator 10 000 9 997 9 998 9 999 99.97 99.98 99.99
Number of queries per original http status
  tool 200 400 403 429 500
0 Listeria 133 27 9 829 3 8
1 MixNMatch 7 477 0 0 0 919
2 Pywikibot 9 812 13 0 3 172
3 SPARQLWrapper 9 864 3 40 59 34
4 WikidataIntegrator 9 997 0 0 1 2

Query feature analysis¶

Undertanding and tagging some query features is important in the scope of a difference analysis. Some SPARQL query features might be very dependent on the shape of the internal data-structure of blazegraph. Such queries have to be tagged to be treated differently and not count for false positives. The features prone to such problems that we identified are:

  • EntitySearch: the use of the MW API may yield different results depending on when the API request is performed.
  • group_concat: The GROUP_CONCAT SPARQL feature does not provide a stable ordering in how it orders the groups it concats.
  • lbl_srv_alt_label: Asking for aliases with the label service is not deterministic when there are multiple aliases for the entity
  • sample: The SPARQL SAMPLE is not required to be deterministic
  • slice.offset: Blazegraph bd:slice with bd:slice.offset cannot operate similarly between two different graphs
  • top_level_limit: A sparql query using top level LIMIT X and produces X results without deterministic ordering clauses might yield different results

Queries without any of this features are considered/named "deterministic".

In [30]:
def per_tool_query_classes(df):
    return (df.groupBy("tool")
     .agg(
         sum(lit(1)).alias("Total"),
         sum((size(col("query_classes")) == 0).cast(LongType())).alias("Total deterministic"),
     )
     .join(df
           .withColumn("qclass", explode(col("query_classes")))
           .groupBy("tool")
           .pivot("qclass")
           .count(),
           "tool", "outer")
     .sort(col("tool").asc())
     .fillna(0))

with panda_style_context():
    display(per_tool_query_classes(full).toPandas().style.set_caption("Query classes per tool"))
    display(per_tool_query_classes(drop_dups(full)).toPandas().style.set_caption("Query (unique) classes per tool"))
Query classes per tool
  tool Total Total deterministic EntitySearch group_concat lbl_srv_alt_label limit_offset sample slice_offset top_level_limit
0 Listeria 10 000 1 380 0 0 0 5 169 0 1 995 4 080
1 MixNMatch 8 396 659 0 7 737 0 0 0 0 0
2 Pywikibot 10 000 7 367 1 138 0 0 1 028 8 428 530
3 SPARQLWrapper 10 000 9 434 15 359 1 23 14 0 205
4 WikidataIntegrator 10 000 1 765 0 1 8 225 8 0 0 3
Query (unique) classes per tool
  tool Total Total deterministic EntitySearch group_concat lbl_srv_alt_label limit_offset sample slice_offset top_level_limit
0 Listeria 101 14 0 0 0 53 0 20 41
1 MixNMatch 626 313 0 313 0 0 0 0 0
2 Pywikibot 5 455 3 212 863 0 0 939 5 428 327
3 SPARQLWrapper 9 919 9 357 12 359 1 23 14 0 201
4 WikidataIntegrator 2 507 1 559 0 1 938 8 0 0 3

Identify true positives¶

Queries and results have also been scanned to detect the presence of a reference to scientific article either in the query itself or in the results they returned once executed on the full graph. Such queries are known to be true positives.

In [14]:
def scientific_art_analysis(df):
    return (df.groupBy("tool")
     .agg(
         sum(lit(1)).alias("total"),
         sum(col("query_with_direct_sa_ref").cast(LongType())).alias("Scientific article in query"),
         sum(col("results_with_direct_sa_ref").cast(LongType())).alias("Scientific article in results"))
     .sort(col("tool").asc())
     .fillna(0))

with panda_style_context():
    display(scientific_art_analysis(full).toPandas().style.set_caption("Queries with a reference to a scientific article"))
    display(scientific_art_analysis(drop_dups(full)).toPandas().style.set_caption("Queries (unique) with a reference to a scientific article"))
Queries with a reference to a scientific article
  tool total Scientific article in query Scientific article in results
0 Listeria 10 000 0 0
1 MixNMatch 8 396 0 7 150
2 Pywikibot 10 000 1 99
3 SPARQLWrapper 10 000 49 25
4 WikidataIntegrator 10 000 61 81
Queries (unique) with a reference to a scientific article
  tool total Scientific article in query Scientific article in results
0 Listeria 101 0 0
1 MixNMatch 626 0 32
2 Pywikibot 5 455 1 60
3 SPARQLWrapper 9 919 49 25
4 WikidataIntegrator 2 507 61 81
In [15]:
def aggA_B(fun, sourceCol, destCol):
    return [
      fun(col("A." + sourceCol)).alias("A_" + destCol),
      fun(col("B." + sourceCol)).alias("B_" + destCol)
    ]


def join_queries(results_A, results_B):
    return (results_A.alias("A")
      .join(results_B.alias("B"), (results_A["id"] == results_B["id"]))
      .withColumn("no_query_class", size(col("A.query_classes")) == 0)
      .withColumn("both_success", (col("A.success") == col("B.success")) & (col("A.success") == lit(True)))
      .withColumn("both_failed", (col("A.success") == col("B.success")) & (col("A.success") == lit(False)))
      .withColumn("true_positive", (coalesce(col("A.query_with_direct_sa_ref"), lit(False)) == lit(True)) | (coalesce(col("A.results_with_direct_sa_ref"), lit(False)) == lit(True)))
      .withColumn("true_positive_success", (col("true_positive") == lit(True)) & (col("both_success") == lit(True)))
      .withColumn("deterministic_success", (col("no_query_class") == lit(True)) & (col("both_success") == lit(True)))
      .withColumn("deterministic_success_minus_known_true_positive", (col("deterministic_success") == lit(True)) & (col("true_positive") == lit(False)))
      .withColumn("same", (col("both_success") == lit(True)) & (col("A.exactHash") == col("B.exactHash")))
      .withColumn("same_reordered", (col("both_success") == lit(True)) & (col("A.reorderedHash") == col("B.reorderedHash")))
      .withColumn("same_reordered_deterministic", (col("deterministic_success") == lit(True)) & (col("same_reordered") == lit(True)))
      .withColumn("same_reordered_deterministic_minus_known_true_positive", (col("true_positive") == lit(False)) & (col("same_reordered_deterministic") == lit(True)))
      .withColumn("same_result_size", (col("both_success") == lit(True)) & (col("A.resultSize") == col("B.resultSize"))))

def aggregate(joined):
    return (joined.groupBy(joined["A.tool"])
      .agg(
        *[sum(lit(1)).alias("total"),
          sum(col("both_success").cast(LongType())).alias("total_both_success"),
          sum(col("both_failed").cast(LongType())).alias("total_both_failed"),
          sum(col("deterministic_success").cast(LongType())).alias("total_success_deterministic"),
          sum(col("deterministic_success_minus_known_true_positive").cast(LongType())).alias("total_deterministic_success_minus_known_true_positive"),
          sum(col("true_positive").cast(LongType())).alias("total_true_positive"),
          sum(col("true_positive_success").cast(LongType())).alias("total_true_positive_success"),
          sum(col("same").cast(LongType())).alias("total_same"),
          sum(col("same_reordered").cast(LongType())).alias("total_same_reordered"),
          sum(col("same_reordered_deterministic").cast(LongType())).alias("total_same_reordered_deterministic"),
          sum(col("same_reordered_deterministic_minus_known_true_positive").cast(LongType())).alias("total_same_reordered_deterministic_minus_known_true_positive"),
          sum(col("same_result_size").cast(LongType())).alias("same_result_size"),
          *aggA_B(lambda c: sum(c.cast(LongType())), "success", "total_success"),
          *aggA_B(lambda c: sum(c.cast(LongType())), "resultSize", "total_result_size"),
          *aggA_B(lambda c: avg(c.cast(LongType())), "resultSize", "avg_result_size"),
          *aggA_B(lambda c: sum(c.cast(LongType())), "results_with_direct_sa_ref", "results_with_direct_sa_ref"),
          *aggA_B(lambda c: sum(when(c == lit("timeout"), lit(1)).otherwise(lit(0))), "error_class", "total_timeouts")
         ])
      )

def compare(results_A, results_B):
    return (aggregate(join_queries(results_A, results_B))
            .withColumn("same_pct", col("total_same") / col("total_both_success")*100)
            .withColumn("same_reordered_pct", col("total_same_reordered") / col("total_both_success")*100)
            .withColumn("same_reordered_deterministic_pct", col("total_same_reordered_deterministic") / col("total_success_deterministic")*100)
            .withColumn("same_reordered_deterministic_minus_known_true_positive_pct", col("total_same_reordered_deterministic_minus_known_true_positive") / col("total_deterministic_success_minus_known_true_positive")*100))

Basic comparison metrics¶

Here we compare the outcome of running the queries against the full graph A and the wikidata main graph B. We extract the following metrics:

  • total: total number of queries
  • total_both_success: total number of queries that are successful in A and B
  • total_both_failed: total number of queries that failed both in A and B
  • [AB]_total_timeouts: total number of failure because of a timeout in A or B
  • [AB]_total_result_size: total number of results in A or B
  • [AB]_avg_result_size: avg of the result size in A or B
In [16]:
def simple_diff_report(df):
    return (df
            .select("tool", "total", "total_both_success", "total_both_failed",
                    "A_total_timeouts", "B_total_timeouts",
                    "A_total_result_size", "B_total_result_size",
                    "A_avg_result_size", "B_avg_result_size")
            .sort(col("tool").asc()))

with panda_style_context():
    display(simple_diff_report(compare(full, wikidata)).toPandas().style.set_caption("Simple diff report (A = full, B = wikidata)"))
    display(simple_diff_report(compare(drop_dups(full), drop_dups(wikidata))).toPandas().style.set_caption("Unique queries - Simple diff report (A = full, B = wikidata)"))
Simple diff report (A = full, B = wikidata)
  tool total total_both_success total_both_failed A_total_timeouts B_total_timeouts A_total_result_size B_total_result_size A_avg_result_size B_avg_result_size
0 Listeria 10 000 9 466 534 206 206 15 844 788 15 844 462 1 673.86 1 673.83
1 MixNMatch 8 396 8 369 27 0 0 69 292 739 69 182 368 8 279.69 8 266.50
2 Pywikibot 10 000 9 944 44 38 36 550 241 528 239 55.31 53.08
3 SPARQLWrapper 10 000 9 906 87 76 71 675 017 669 164 68.14 67.51
4 WikidataIntegrator 10 000 9 998 1 2 1 73 293 81 820 7.33 8.18
Unique queries - Simple diff report (A = full, B = wikidata)
  tool total total_both_success total_both_failed A_total_timeouts B_total_timeouts A_total_result_size B_total_result_size A_avg_result_size B_avg_result_size
0 Listeria 101 96 5 2 2 162 611 162 614 1 693.86 1 693.90
1 MixNMatch 626 619 7 0 0 4 712 535 4 710 489 7 613.14 7 609.84
2 Pywikibot 5 455 5 409 35 29 26 513 110 515 271 94.79 95.14
3 SPARQLWrapper 9 919 9 831 81 70 65 637 957 632 104 64.89 64.26
4 WikidataIntegrator 2 507 2 505 1 2 1 61 998 70 525 24.75 28.14

Results comparison metrics¶

  • success: total number of queries that are successful in A and B
  • deterministic: total number of deterministic queries
  • true positives: total number of true positive queries (queries where we identified a scientific article in the query or in the results)
  • Same: number of queries that returns exactly the same result set
  • Same (reordered): number of queries that returns the same result set once re-ordered
  • deterministic Same (reordered): number of deterministic queries that return the same result set once re-ordered
  • deterministic - TP Same (reordered): number of deterministic queries (excluding the known true positives) that return the same result set once re-ordered
  • % Same: Percentage of queries that return similar results same / success
  • % Same (reordered): Percentage of queries that return similar results (reordered) same (reordered) / success
  • % deterministic Same (reordered): Percentage of deterministic queries that return similar results (reordered) deterministic same (reordered) / deterministic
  • % deterministic - TP Same (reordered): Percentage of deterministic queries (excl. true positives) that return similar results (reordered)
In [17]:
def results_comparison_report(df):
    return (df
            .select("tool",
                    col("total_both_success").alias("success"),
                    col("total_success_deterministic").alias("deterministic"),
                    col("total_true_positive_success").alias("true positives"),
                    col("total_same").alias("Same"),
                    col("total_same_reordered").alias("Same (reordered)"),
                    col("total_same_reordered_deterministic").alias("deterministic Same (reordered)"),
                    col("total_same_reordered_deterministic_minus_known_true_positive").alias("deterministic - TP Same (reordered)"),
                    col("same_pct").alias("% Same"),
                    col("same_reordered_pct").alias("% Same (reordered)"),
                    col("same_reordered_deterministic_pct").alias("% deterministic Same (reordered)"),
                    col("same_reordered_deterministic_minus_known_true_positive_pct").alias("% deterministic - TP Same (reordered)"),
                   )
            .fillna(0)
            .sort(col("tool").asc()))

with panda_style_context():
    display(results_comparison_report(compare(full, wikidata)).toPandas().style.set_caption("Comparison report (A = full, B = wikidata)"))
    display(results_comparison_report(compare(drop_dups(full), drop_dups(wikidata))).toPandas().style.set_caption("Unique queries - Comparison report (A = full, B = wikidata)"))
Comparison report (A = full, B = wikidata)
  tool success deterministic true positives Same Same (reordered) deterministic Same (reordered) deterministic - TP Same (reordered) % Same % Same (reordered) % deterministic Same (reordered) % deterministic - TP Same (reordered)
0 Listeria 9 466 1 052 0 5 660 6 419 862 862 59.79 67.81 81.94 81.94
1 MixNMatch 8 369 632 7 150 615 1 184 593 593 7.35 14.15 93.83 100.00
2 Pywikibot 9 944 7 341 100 9 122 9 396 7 329 7 328 91.73 94.49 99.84 99.90
3 SPARQLWrapper 9 906 9 344 74 9 152 9 845 9 317 9 268 92.39 99.38 99.71 99.97
4 WikidataIntegrator 9 998 1 765 82 8 271 8 830 1 683 1 683 82.73 88.32 95.35 100.00
Unique queries - Comparison report (A = full, B = wikidata)
  tool success deterministic true positives Same Same (reordered) deterministic Same (reordered) deterministic - TP Same (reordered) % Same % Same (reordered) % deterministic Same (reordered) % deterministic - TP Same (reordered)
0 Listeria 96 11 0 58 66 9 9 60.42 68.75 81.82 81.82
1 MixNMatch 619 306 32 302 567 282 282 48.79 91.60 92.16 100.00
2 Pywikibot 5 409 3 190 61 4 761 4 939 3 178 3 177 88.02 91.31 99.62 99.78
3 SPARQLWrapper 9 831 9 273 74 9 094 9 770 9 246 9 197 92.50 99.38 99.71 99.97
4 WikidataIntegrator 2 505 1 559 82 1 797 2 355 1 477 1 477 71.74 94.01 94.74 100.00

Impacted User-Agents¶

The list of UA sorted by their number of impacted queries (deduplicated). An impacted query is a query that we have returned different results even when reordered and that is deterministic. The total shows how many are differents including deterministic ones.

In [25]:
def impacted_uas(df_full, df_wikidata):
    return (join_queries(drop_dups(df_full), drop_dups(df_wikidata))
     .filter("same_reordered = false AND same = false AND both_success = true")
     .groupBy("A.user_agent")
     .agg(
         sum(lit(1)).alias("total"),
         sum(col("no_query_class").cast(LongType())).alias("impacted"),
         sum(col("true_positive").cast(LongType())).alias("true_positives"))
     .join(join_queries(drop_dups(df_full), drop_dups(df_wikidata))
            .filter("same_reordered = false AND same = false AND both_success = true")
            .withColumn("qclass", explode(col("A.query_classes")))
            .groupBy("A.user_agent")
            .pivot("qclass")
            .count(),
           "user_agent", "outer")
     .sort(col("impacted").desc())
     .withColumn("user_agent", lit("REDACTED (CHECKING PII)"))
     .fillna(0))
     

with panda_style_context():
    display(impacted_uas(full, wikidata).toPandas().style.set_caption("Impacted UA"))
Impacted UA
  user_agent total impacted true_positives EntitySearch group_concat lbl_srv_alt_label limit_offset sample slice_offset top_level_limit
0 REDACTED (CHECKING PII) 64 62 62 0 0 0 2 0 0 2
1 REDACTED (CHECKING PII) 54 27 25 0 4 1 3 2 0 24
2 REDACTED (CHECKING PII) 52 24 32 0 28 0 0 0 0 0
3 REDACTED (CHECKING PII) 18 18 18 0 0 0 0 0 0 0
4 REDACTED (CHECKING PII) 9 7 0 0 0 0 0 1 0 1
5 REDACTED (CHECKING PII) 4 4 4 0 0 0 0 0 0 0
6 REDACTED (CHECKING PII) 2 2 2 0 0 0 0 0 0 0
7 REDACTED (CHECKING PII) 30 2 0 0 0 0 0 0 15 13
8 REDACTED (CHECKING PII) 4 1 1 0 0 0 0 0 3 0
9 REDACTED (CHECKING PII) 7 0 0 0 0 0 7 0 0 7
10 REDACTED (CHECKING PII) 290 0 0 0 0 0 290 0 0 0
11 REDACTED (CHECKING PII) 58 0 50 58 0 0 0 0 0 32
12 REDACTED (CHECKING PII) 14 0 0 0 0 0 14 0 0 0
13 REDACTED (CHECKING PII) 66 0 0 0 0 66 0 0 0 0
14 REDACTED (CHECKING PII) 3 0 0 0 3 0 0 0 0 0
15 REDACTED (CHECKING PII) 47 0 1 0 0 0 47 0 0 0
16 REDACTED (CHECKING PII) 7 0 0 0 0 0 7 0 0 0
17 REDACTED (CHECKING PII) 1 0 0 0 0 0 1 0 0 1
18 REDACTED (CHECKING PII) 4 0 0 0 0 0 4 0 0 4
19 REDACTED (CHECKING PII) 1 0 0 0 0 0 0 0 0 1
20 REDACTED (CHECKING PII) 1 0 1 0 0 0 0 0 0 1
21 REDACTED (CHECKING PII) 4 0 3 0 0 0 4 0 0 4
22 REDACTED (CHECKING PII) 1 0 0 0 0 0 0 0 0 1
23 REDACTED (CHECKING PII) 1 0 0 0 0 0 0 0 0 1
24 REDACTED (CHECKING PII) 21 0 0 0 0 0 21 0 0 0
In [20]:
def results_substract(res_a, res_b, query_id):
    """
    Extract diff between to result sets
    """
    return (res_a.alias("A").filter(col("id") == lit(query_id))
     .select(explode("A.results").alias("ar"))
     .withColumn("ars", col("ar").cast(StringType()))
     .join(res_b.alias("B").filter(col("B.id") == lit(query_id))
           .select(explode("B.results").alias("br"))
           .withColumn("brs", col("br").cast(StringType())), col("ars") == col("brs"), "left_anti")
     .select("ar"))
In [ ]:
# Debug
def deterministic_queries_that_are_different(df1, df2, ua):
    return (join_queries(drop_dups(df1), drop_dups(df2))
                .filter("same_reordered = false AND same = false AND both_success = true AND no_query_class is true AND size(A.ua_classes) = 0 and ")
                .filter(col(A.user_agent) == ua)
                .select("A.query", "A.id", "A.user_agent", "A.resultSize"))

# results_substract(wikidata, full, "???QUERY_ID????").show(10, False)

Pywikibot affected queries¶

These are the queries that return different results but for which no scientific article can be identified neither in the query nor in the results. Quickly glancing over them they seem to be extracting counts.

In [28]:
df = (join_queries(drop_dups(full), drop_dups(wikidata))
     .filter("true_positive == false AND both_success = true AND no_query_class = true AND same_reordered = false AND same = false")
     .filter("A.tool = 'Pywikibot'")
     .drop("A.results", "B.results", "B.query")
     .select("A.query", lit("UA - REDACTED (CHECKING PII)")))

display(to_html_table(df))
queryUA - REDACTED (CHECKING PII)
SELECT (COUNT(*) as ?count) WHERE {
  ?entity wdt:P4101 wd:Q41506
  FILTER(EXISTS {
    ?entity p:P495[]
  })
}

UA - REDACTED (CHECKING PII)
SELECT (COUNT(*) AS ?count) WHERE {
  ?entity wdt:P31/wdt:P279? wd:Q1266946 .
  MINUS { ?entity wdt:P4101 _:b28. }
  FILTER(EXISTS {
    ?entity p:P577[]
  })
}

UA - REDACTED (CHECKING PII)
SELECT ?grouping (COUNT(DISTINCT ?entity) as ?count) WHERE {
  ?entity wdt:P31 wd:Q3331189. ?sitelink schema:about ?entity; schema:isPartOf  .
  ?entity wdt:P407 ?grouping .
  FILTER(EXISTS {
    ?entity p:P577[]
  })
}
GROUP BY ?grouping
HAVING (?count >= 0)
ORDER BY DESC(?count)
LIMIT 1000

UA - REDACTED (CHECKING PII)
SELECT ?grouping (COUNT(DISTINCT ?entity) as ?count) WHERE {
  ?entity wdt:P361 wd:Q18477654 .
  ?entity wdt:P361 ?grouping .
  FILTER(EXISTS {
    ?entity p:P2860[]
  })
}
GROUP BY ?grouping
HAVING (?count >= 0)
ORDER BY DESC(?count)
LIMIT 1000

UA - REDACTED (CHECKING PII)
SELECT (COUNT(*) AS ?count) WHERE {
  ?entity wdt:P31 wd:Q3331189. MINUS {?entity  wdt:P1433 [] } .
  MINUS { ?entity wdt:P407 _:b28. }
  FILTER(EXISTS {
    ?entity p:P291[]
  })
}

UA - REDACTED (CHECKING PII)
SELECT ?grouping (COUNT(DISTINCT ?entity) as ?count) WHERE {
  ?entity wdt:P31/wdt:P279? wd:Q1266946 .
  ?entity wdt:P4101 ?grouping .
  FILTER(EXISTS {
    ?entity p:P577[]
  })
}
GROUP BY ?grouping
HAVING (?count >= 0)
ORDER BY DESC(?count)
LIMIT 1000

UA - REDACTED (CHECKING PII)
SELECT ?grouping (COUNT(DISTINCT ?entity) as ?count) WHERE {
  ?entity wdt:P31 wd:Q3331189. ?sitelink schema:about ?entity; schema:isPartOf  .
  ?entity wdt:P407 ?grouping .
} GROUP BY ?grouping
HAVING (?count >= 0)
ORDER BY DESC(?count)
LIMIT 1000

UA - REDACTED (CHECKING PII)

SPARQLWrapper affected queries¶

These are the queries that return different results but for which no scientific article can be identified neither in the query nor in the results.

These queries do also seem to extract counts but seem to be specifically targetted against scholarly article (with ?s wdt:P31 wd:Q13442814)

In [29]:
df = (join_queries(drop_dups(full), drop_dups(wikidata))
     .filter("true_positive == false AND both_success = true AND no_query_class = true AND same_reordered = false AND same = false")
     .filter("A.tool = 'SPARQLWrapper'")
     .drop("A.results", "B.results", "B.query")
     .select("A.query", lit("UA - REDACTED (CHECKING PII)")))

display(to_html_table(df))
queryUA - REDACTED (CHECKING PII)
SELECT (COUNT(DISTINCT ?o) AS ?count)
                        WHERE{
                          ?o wdt:P106 wd:Q36180.
                          ?s wdt:P3931 ?o
                        }
                    
UA - REDACTED (CHECKING PII)
SELECT (COUNT(DISTINCT ?o) AS ?distinctObjectCount)
                (COUNT(*) AS ?totalFactCount)
                WHERE {
                  ?o wdt:P106 wd:Q33999.
                  ?s wdt:P31 wd:Q13442814.
                  ?s wdt:P50 ?o.
                }
                
UA - REDACTED (CHECKING PII)
SELECT (COUNT(DISTINCT ?o) AS ?distinctObjectCount)
                (COUNT(*) AS ?totalFactCount)
                WHERE {
                  ?o wdt:P31 wd:Q3331189.
                  ?s wdt:P31 wd:Q13442814.
                  ?s wdt:P2860 ?o.
                }
                
UA - REDACTED (CHECKING PII)
In [ ]: