@Direo Direo​ :

The difference between the two scenarios lies in the execution plan generated by Spark. When Spark optimizes and executes a query, it generates a physical plan that includes all the operations that need to be performed to produce the final result. The physical plan is generated by applying a series of optimizations to the logical plan of the query, such as predicate pushdown, projection pruning, and join reordering.

In the first scenario, the filter operation is applied directly to the base dataframe, which only contains the columns "a", "b", "e", and "g". Therefore, when Spark generates the physical plan, it does not include the "c" column, as it is not needed for the filter operation.

In the second scenario, the filter operation is applied to the base_join dataframe, which is the result of a join operation that includes the "a", "b", "e", and "g" columns. However, the df1 dataframe, which contains the "c" column, is also included in the join operation. Therefore, when Spark generates the physical plan, it includes the "c" column in the join operation, even though it is not used in the final filter operation.

To fix the error in the second scenario, you can add a projection operation to

base_join to remove the "c" column before applying the filter operation:

base_join = base.join(df4, "a").join(df5, "a", "left").select("a", "b", "e", "g")
 
base_join_projected = base_join.drop("c")
 
final = base_join_projected.filter(col("c") == 1)

This will remove the "c" column from the physical plan, and the filter operation should work as expected.