Фильтровать диапазон дат в Dataframe

Question

Я очень плохо знаком с scala, и у меня есть фрейм данных с Id и valid_From и Valid_to Date, как показано ниже:

DataFrame1 (`name in query as tD`): 
id  E_id    start_date  CTC     valid_from  valid_to
z1  it1     2016-01-01  25      2016-02-01  2017-03-01 
z1  it1     2017-01-01  25.75   2017-03-01  null

И у меня есть еще один фрейм данных, как показано ниже:

DataFrame 2 (name in query as finalDF):

id  D_id   Join_date
z1  it1    2016-05-01
z1  it1    2017-06-01

, и я объединяю эти два кадра данных на основе id и D_id и нуждаюсь в записях, где DataFrame 2 Join_Date находится между DataFrame 1 start_date и valid_to. Таким образом, конечный результат должен быть таким:

id  D_id
z1  it1 
z1  it1 

Я делаю это так:

val  joinDF = finalDF.as("DF")
   .join(tD.as("tDF"),($"tDF.id" === $"DF.id" && $"tDF.D_id" === $"DF.D_id"))
   .filter($"Join_date".between($"start_date", $"valid_to") || $"Join_date".between($"start_date", $"valid_to".isNull))

Но это не работает. Пожалуйста, помогите мне.