I have a complex join that I'm trying to optimize df1 has cols id,main_key,col1,col1_isnull,col2,col2_isnull...col30 df2 has cols id,main_key,col1,col2..col_30

I'm trying to run this sql query on Pyspark

select df1.id, df2.id from df1 join df2 on df1.main_key = df2.main_key AND (df1.col1_is_null OR (df1.col1 = df2.col1)) AND (df1.col2_is_null OR (df1.col2 = df2.col2)) ...

This query takes a very lot of time with just a few long running straggler tasks both dataframes are huge, and the join key is skewed

Things I've tried:

1. spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

2. Salting the smaller df, exploding the other

3. broadcasting the smaller df (sometimes the AQE overrides it with a SortMergeJoin(skew=true))

4. Filtering just the top 2 most common main_key value first, then doing all the above

5. Splitting the query to joining on main_key and then filtering using a 2nd query

The tasks execution still is very skewed What more can I do to optimize this further?