Fix the join audit for a Spark job
Your task
An AI agent wrote a Spark job that inner-joins clickstream events to a user table, with a broadcast hint, and writes the output partitioned by date. Before review you want three facts you can defend: how many events the join drops, whether the broadcast is safe, and how many date partitions the output lands in.
An assistant wrote audit_spark_join(event_user_ids, event_dates, dim_user_ids, dim_size_mb, broadcast_limit_mb). It is in the editor. It has a bug. Find it and fix it.
Input and output:
event_user_ids,event_dates: two lists of the same length. Eventibelongs to userevent_user_ids[i]and happened onevent_dates[i].dim_user_ids: the user ids in the user table. Ids may repeat.dim_size_mb,broadcast_limit_mb: the user table's size and the largest table the cluster will broadcast, in MB.- Return exactly three strings, in this order:
"dropped=<n>","broadcast=<safe or unsafe>","partitions=<k>".
Rules the tests check:
- An event survives the join when its user id appears in
dim_user_ids, compared exactly. The empty string is an ordinary id. nis the number of event rows that do not survive. Every row counts, so two events from the same missing user count as 2.- The broadcast is
safewhendim_size_mbis less than or equal tobroadcast_limit_mb, elseunsafe. kis the number of distinct dates among the surviving events only.
Examples
audit_spark_join(event_user_ids: list[str], event_dates: list[str], dim_user_ids: list[str], dim_size_mb: int, broadcast_limit_mb: int) -> list[str]Example 1
Input
audit_spark_join(["u1", "u2", "u3"], ["2026-01-01", "2026-01-01", "2026-01-02"], ["u1", "u2"], 8, 10)Output
["dropped=1", "broadcast=safe", "partitions=1"]Example 2
Input
audit_spark_join([], [], ["u1"], 12, 10)Output
["dropped=0", "broadcast=unsafe", "partitions=0"]Example 3
Input
audit_spark_join(["a"], ["2026-02-01"], [], 0, 10)Output
["dropped=1", "broadcast=safe", "partitions=0"]