I have a Delta table that I keep in sync with a relational (SQL Server) table. The inserts and updates are easy but checking for records to delete is prohibitively slow. I am querying the relational table for all primary key values and any primary key values that don't exist in the Delta table get deleted from the Delta table. The following job takes about 10 minutes for the 700 million record table.

pks = spark.read.format("jdbc").option("query": "SELECT pk FROM sql_table_name").load()
delta_table = spark.read.table(delta_table_name)
r = target_table.filter(~col("pk").isin(pks[0]))
display(r)

Is this just something I should expect to take a long time, or is there a meaningfully more efficient way to do this? The Delta table uses liquid clustering, partitioned on lower cardinality columns. I am doing this in a PySpark notebook at the moment.
Thanks!