pyspark join with CassandraSparkExtensions enabled - Unable to join using a custom generated column via when clauses or fillna

Background Details

Spark - 3.1.2 pyspark - 3.1.2 Cassandra Connector - 3.1

SparkSession = SparkSession.builder \
            .config("spark.sql.crossJoin.enabled", "true") \
            .config("spark.streaming.backpressure.enabled", "true") \
            .config("spark.sql.extensions", "com.datastax.spark.connector.CassandraSparkExtensions") \
            .config("directJoinSetting", "off")

Issue

I have recently enabled CassandraSparkExtensions, which has started to complain about directJoinSetting meta-data, as expected (mismatch types etc). So, I disabled directJoinSetting ('off') and re-executed my spark jobs.

However, I am now getting exceptions when joining 2 data frames, sourced from Cassandra, where any of the columns used in the join involve case statements. It looks like the directJoinStrategy checks are still being executed, even though I have turned it off, but it is ignoring the meta data issues and instead finds the above issue around join columns.

Example

Generating a column temp_etl_hr_history_to_datetime based on a when clause and then using in a join later on.

    org_structure_log = self._spark_client.load('human_resource_org_structure_log', DatabaseConstants.ADA_BUSINESS_KEYSPACE) \
                .withColumn('temp_etl_hr_history_to_datetime',
                            F.when(F.col('etl_hr_history_to_datetime').isNull(), F.lit('N/A')).otherwise(F.col('etl_hr_history_to_datetime')).cast('string'))

    organisation_parent_child_details = org_children_details \
                .join(org_parent_details, primary_columns + ['temp_etl_hr_history_to_datetime'])

Exception

An error occurred while calling o299.cache: scala.MatchError: CASE WHEN isnull(etl_hr_history_to_datetime#5) THEN N/A ELSE cast(etl_hr_history_to_datetime#5 as string) END AS temp_etl_hr_history_to_datetime#18 (of class org.apache.spark.sql.catalyst.expressions.Alias)
    at org.apache.spark.sql.cassandra.execution.CassandraDirectJoinStrategy$.$anonfun$aliasMap$1(CassandraDirectJoinStrategy.scala:316)
    at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
    at scala.collection.immutable.List.foreach(List.scala:392)
    at scala.collection.TraversableLike.map(TraversableLike.scala:238)
    at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
    at scala.collection.immutable.List.map(List.scala:298)
    at org.apache.spark.sql.cassandra.execution.CassandraDirectJoinStrategy$.aliasMap(CassandraDirectJoinStrategy.scala:316)
    at org.apache.spark.sql.cassandra.execution.CassandraDirectJoinStrategy$.allPartitionKeysAreJoined(CassandraDirectJoinStrategy.scala:300)
    at org.apache.spark.sql.cassandra.execution.CassandraDirectJoinStrategy$.validJoinBranch(CassandraDirectJoinStrategy.scala:277)
    at org.apache.spark.sql.cassandra.execution.CassandraDirectJoinStrategy.leftValid(CassandraDirectJoinStrategy.scala:124)
    at org.apache.spark.sql.cassandra.execution.CassandraDirectJoinStrategy.hasValidDirectJoin(CassandraDirectJoinStrategy.scala:82)
    at org.apache.spark.sql.cassandra.execution.CassandraDirectJoinStrategy.apply(CassandraDirectJoinStrategy.scala:29)

Workaround

The only workaround which gets around the above issue is to cache the data frame that generates the temporary column temp_etl_hr_history_to_datetime

See below for adding the cache(). This is not ideal and seems hacky.

    org_structure_log = self._spark_client.load('human_resource_org_structure_log', DatabaseConstants.ADA_BUSINESS_KEYSPACE) \
                .withColumn('temp_etl_hr_history_to_datetime',
                            F.when(F.col('etl_hr_history_to_datetime').isNull(), F.lit('N/A')).otherwise(F.col('etl_hr_history_to_datetime')).cast('string')) \
                .cache()