-- Scan notebook command history for potential paper cut issues
-- https://docs.databricks.com/en/compute/access-mode-limitations.html#compute-access-mode-limitations
-- This query 'overcounts' the paper cuts that might be encountered. There are complex DBR interactions with DBR 11, 12, 13 and 14
with paper_cut_patterns(
select col1 as pattern, col2 as issue from values
    ('hive_metastore.', 'I00'),
    ('spark.catalog.', 'I01.1'),
    ('spark._jsparkSession.catalog','I01.2'),
    ('spark._jspark', 'I02.1'),
    ('spark._jvm','I02.2'),
    ('._jdf', 'I02.3'),
    ('._jcol','I02.4'),
    ('spark.read.format("jdbc")', 'I04'),
    ('dbutils.notebook.entry_point.getDbutils().notebook().getContext().toJson()','I05.1'),
    ('dbutils.notebook.entry_point.getDbutils().notebook().getContext()','I05.2'),
    ('spark.udf.registerJavaFunction','I06'),
    ('boto3', 'I07.1'),
    ('s3fs', 'I07.2'),
    ('from graphframes', 'I08'),
    ('pyspark.ml.', 'I09'),
    ('applyInPandas', 'I10.1'),
    ('mapInPandas', 'I10.2'),
    ('dbutils.fs.', 'I11'),
    ('dbutils.credentials.', 'I12') -- credential passthrough
),
sparkcontext (
    select explode(split("_jvm, _jvm.org.apache.log4j, emptyRDD, range, init_batched_serializer, parallelize, pickleFile, textFile, wholeTextFiles, binaryFiles, binaryRecords, sequenceFile, newAPIHadoopFile, newAPIHadoopRDD, hadoopFile, hadoopRDD, union, runJob, setSystemProperty, uiWebUrl, stop, setJobGroup, setLocalProperty, getConf",', ')) as pattern,
    'I03.1' as issue
    UNION ALL
    select explode(split("from pyspark.sql import SQLContext, import org.apache.spark.sql.SQLContext, spark.sparkContext ", ', ')) as pattern, 'I03.2' as issue
),
streaming (
    select explode(split('.trigger(continuous, kafka.sasl.client.callback.handler.class, kafka.sasl.login.callback.handler.class, kafka.sasl.login.class, kafka.partition.assignment.strategy, kafka.ssl.truststore.location, kafka.ssl.keystore.location, cleanSource, sourceArchiveDir, applyInPandasWithState, .format("socket"), StreamingQueryListener',', ')) pattern,
    'I30' as issue
),
paper_cuts(
    select pattern, issue from paper_cut_patterns
    UNION ALL
    select concat('sc.',pattern) as pattern, issue from sparkcontext
    UNION ALL
    select pattern, issue from streaming
),
iteractive_cluster_commands (
    select 
        a.request_params.notebookId as notebook_id, 
        a.request_params.clusterId as cluster_id, 
        a.user_identity.email,
        a.request_params.commandLanguage,
        a.request_params.commandText 
    from system.access.audit a
    join system.compute.clusters as c
        ON c.cluster_source != 'JOB'
        AND (c.tags.ResourceClass is null OR c.tags.ResourceClass != "SingleNode")
        AND a.action_name = 'runCommand' 
        AND a.request_params.clusterId = c.cluster_id
    WHERE
        a.event_date >= DATE_SUB(CURRENT_DATE(), 90)
),
python_matcher(
    select 
    p.issue, 
    a.notebook_id, 
    a.cluster_id, 
    a.email,
    a.commandLanguage,
    a.commandText 
from iteractive_cluster_commands a
join paper_cuts p
    ON a.commandLanguage = 'python'
    AND contains(a.commandText, p.pattern)
),
scala_matcher(
    select
        'I20 - scala/R' as issue,
        a.notebook_id, 
        a.cluster_id, 
        a.email, 
        a.commandText  
    from iteractive_cluster_commands a
    where a.commandLanguage in ('scala','r')
),
unions(
    SELECT issue, notebook_id, cluster_id, email, commandText from python_matcher
    UNION ALL
    SELECT issue, notebook_id, cluster_id, email, commandText from scala_matcher
)
SELECT issue, 
    count(distinct notebook_id) distinct_notebooks,
    count(distinct cluster_id) distinct_clusters,
    count(distinct email) distinct_users
FROM unions group by 1
order by 1
;