title: Spark detected cartesian product despite the join condition is non-trivial
body: <p>I am using Spark 2.3.0 and I have two data frames.</p>
  
  <p>The first one, df1, has the schema:</p>
  
  <pre><code>root
   |-- time: long (nullable = true)
   |-- channel: string (nullable = false)
  </code></pre>
  
  <p>The second one, df2, has the schema:</p>
  
  <pre><code>root
   |-- pprChannel: string (nullable = true)
   |-- ppr: integer (nullable = false)
  </code></pre>
  
  <p>I now try to do:</p>
  
  <pre><code>spark.sql("select a.channel as channel, a.time as time, b.ppr as ppr from df1 a inner join df2 b on a.channel = b.pprChannel")
  </code></pre>
  
  <p>But I get <code>Detected cartesian product for INNER join between logical plans</code>.</p>
  
  <p>When I try to recreate both on a Spark-Shell with <code>sc.parallelize</code> and simple Seqs, it works.</p>
  
  <p>What might be wrong here?</p>
  
  <h1>Followup</h1>
  
  <p>Here is what I get when I use <code>df1.join(df2, 'channel === 'pprChannel, "inner").explain(true)</code>:</p>
  
  <pre><code>== Parsed Logical Plan ==
  Join Inner, (channel#124 = pprChannel#136)
  :- Project [time#113L AS time#127L, channel#124]
  :  +- Project [time#113L, unnamed AS channel#124]
  :     +- Project [time#113L]
  :        +- Project [channel#23, time#113L]
  :           +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98, clipDT#105L, if ((isnull(t0#93L) || isnull(t1#29L))) null else UDF(t0#93L, t1#29L) AS time#113L]
  :              +- Filter (clipDT#105L &gt;= cast(50000000 as bigint))
  :                 +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98, (t1#29L - t0#93L) AS clipDT#105L]
  :                    +- Filter (((t0#93L &gt;= cast(0 as bigint)) &amp;&amp; (pt0#98 = 1)) &amp;&amp; (pt1#82 = 2))
  :                       +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98]
  :                          +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98, pt0#98]
  :                             +- Window [lag(pt1#82, 1, 0) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS pt0#98], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                +- Project [channel#23, t1#29L, pt1#82, t0#93L]
  :                                   +- Project [channel#23, t1#29L, pt1#82, t0#93L]
  :                                      +- Project [channel#23, t1#29L, pt1#82, t0#93L, t0#93L]
  :                                         +- Window [lag(t1#29L, 1, -1) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#93L], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                            +- Project [channel#23, t1#29L, pt1#82]
  :                                               +- Project [channel#23, t1#29L, pt1#82]
  :                                                  +- Filter pt1#82 IN (1,2)
  :                                                     +- Project [channel#23, t1#29L, dv1#58, t0#70L, dv0#75, if ((isnull(dv0#75) || isnull(dv1#58))) null else UDF(dv0#75, dv1#58) AS pt1#82]
  :                                                        +- Filter ((t0#70L &gt;= cast(0 as bigint)) &amp;&amp; NOT isnan(dv0#75))
  :                                                           +- Project [channel#23, t1#29L, dv1#58, t0#70L, dv0#75]
  :                                                              +- Project [channel#23, t1#29L, dv1#58, t0#70L, dv0#75, dv0#75]
  :                                                                 +- Window [lag(dv1#58, 1, NaN) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS dv0#75], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                    +- Project [channel#23, t1#29L, dv1#58, t0#70L]
  :                                                                       +- Project [channel#23, t1#29L, dv1#58, t0#70L]
  :                                                                          +- Project [channel#23, t1#29L, dv1#58, t0#70L, t0#70L]
  :                                                                             +- Window [lag(t1#29L, 1, -1) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#70L], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                                +- Project [channel#23, t1#29L, dv1#58]
  :                                                                                   +- Project [channel#23, t1#29L, dv1#58]
  :                                                                                      +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, v0#49, abs(if ((isnull(v0#49) || isnull(v1#35))) null else UDF(v0#49, v1#35)) AS dv1#58]
  :                                                                                         +- Filter ((t0#42L &gt;= cast(0 as bigint)) &amp;&amp; NOT isnan(v0#49))
  :                                                                                            +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, v0#49]
  :                                                                                               +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, v0#49, v0#49]
  :                                                                                                  +- Window [lag(v1#35, 1, NaN) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS v0#49], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                                                     +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L]
  :                                                                                                        +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L]
  :                                                                                                           +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, t0#42L]
  :                                                                                                              +- Window [lag(t1#29L, 1, -1) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#42L], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                                                                 +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23]
  :                                                                                                                    +- Filter ((NOT isnull(t1#29L) &amp;&amp; NOT isnull(v1#35)) &amp;&amp; ((t1#29L &gt;= cast(0 as bigint)) &amp;&amp; NOT isnan(v1#35)))
  :                                                                                                                       +- Project [_c0#10, _c1#11, t1#29L, value#18 AS v1#35, channel#23]
  :                                                                                                                          +- Project [_c0#10, _c1#11, time#14L AS t1#29L, value#18, channel#23]
  :                                                                                                                             +- Project [_c0#10, _c1#11, time#14L, value#18, unnamed AS channel#23]
  :                                                                                                                                +- Project [_c0#10, _c1#11, time#14L, UDF(_c1#11) AS value#18]
  :                                                                                                                                   +- Project [_c0#10, _c1#11, UDF(_c0#10) AS time#14L]
  :                                                                                                                                      +- Relation[_c0#10,_c1#11] csv
  +- Project [_1#133 AS pprChannel#136, _2#134 AS ppr#137]
     +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._1, true, false) AS _1#133, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._2 AS _2#134]
        +- ExternalRDD [obj#132]
  
  == Analyzed Logical Plan ==
  time: bigint, channel: string, pprChannel: string, ppr: int
  Join Inner, (channel#124 = pprChannel#136)
  :- Project [time#113L AS time#127L, channel#124]
  :  +- Project [time#113L, unnamed AS channel#124]
  :     +- Project [time#113L]
  :        +- Project [channel#23, time#113L]
  :           +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98, clipDT#105L, if ((isnull(t0#93L) || isnull(t1#29L))) null else if ((isnull(t0#93L) || isnull(t1#29L))) null else UDF(t0#93L, t1#29L) AS time#113L]
  :              +- Filter (clipDT#105L &gt;= cast(50000000 as bigint))
  :                 +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98, (t1#29L - t0#93L) AS clipDT#105L]
  :                    +- Filter (((t0#93L &gt;= cast(0 as bigint)) &amp;&amp; (pt0#98 = 1)) &amp;&amp; (pt1#82 = 2))
  :                       +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98]
  :                          +- Project [channel#23, t1#29L, pt1#82, t0#93L, pt0#98, pt0#98]
  :                             +- Window [lag(pt1#82, 1, 0) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS pt0#98], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                +- Project [channel#23, t1#29L, pt1#82, t0#93L]
  :                                   +- Project [channel#23, t1#29L, pt1#82, t0#93L]
  :                                      +- Project [channel#23, t1#29L, pt1#82, t0#93L, t0#93L]
  :                                         +- Window [lag(t1#29L, 1, -1) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#93L], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                            +- Project [channel#23, t1#29L, pt1#82]
  :                                               +- Project [channel#23, t1#29L, pt1#82]
  :                                                  +- Filter pt1#82 IN (1,2)
  :                                                     +- Project [channel#23, t1#29L, dv1#58, t0#70L, dv0#75, if ((isnull(dv0#75) || isnull(dv1#58))) null else if ((isnull(dv0#75) || isnull(dv1#58))) null else UDF(dv0#75, dv1#58) AS pt1#82]
  :                                                        +- Filter ((t0#70L &gt;= cast(0 as bigint)) &amp;&amp; NOT isnan(dv0#75))
  :                                                           +- Project [channel#23, t1#29L, dv1#58, t0#70L, dv0#75]
  :                                                              +- Project [channel#23, t1#29L, dv1#58, t0#70L, dv0#75, dv0#75]
  :                                                                 +- Window [lag(dv1#58, 1, NaN) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS dv0#75], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                    +- Project [channel#23, t1#29L, dv1#58, t0#70L]
  :                                                                       +- Project [channel#23, t1#29L, dv1#58, t0#70L]
  :                                                                          +- Project [channel#23, t1#29L, dv1#58, t0#70L, t0#70L]
  :                                                                             +- Window [lag(t1#29L, 1, -1) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#70L], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                                +- Project [channel#23, t1#29L, dv1#58]
  :                                                                                   +- Project [channel#23, t1#29L, dv1#58]
  :                                                                                      +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, v0#49, abs(if ((isnull(v0#49) || isnull(v1#35))) null else if ((isnull(v0#49) || isnull(v1#35))) null else UDF(v0#49, v1#35)) AS dv1#58]
  :                                                                                         +- Filter ((t0#42L &gt;= cast(0 as bigint)) &amp;&amp; NOT isnan(v0#49))
  :                                                                                            +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, v0#49]
  :                                                                                               +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, v0#49, v0#49]
  :                                                                                                  +- Window [lag(v1#35, 1, NaN) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS v0#49], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                                                     +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L]
  :                                                                                                        +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L]
  :                                                                                                           +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23, t0#42L, t0#42L]
  :                                                                                                              +- Window [lag(t1#29L, 1, -1) windowspecdefinition(channel#23, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#42L], [channel#23], [t1#29L ASC NULLS FIRST]
  :                                                                                                                 +- Project [_c0#10, _c1#11, t1#29L, v1#35, channel#23]
  :                                                                                                                    +- Filter ((NOT isnull(t1#29L) &amp;&amp; NOT isnull(v1#35)) &amp;&amp; ((t1#29L &gt;= cast(0 as bigint)) &amp;&amp; NOT isnan(v1#35)))
  :                                                                                                                       +- Project [_c0#10, _c1#11, t1#29L, value#18 AS v1#35, channel#23]
  :                                                                                                                          +- Project [_c0#10, _c1#11, time#14L AS t1#29L, value#18, channel#23]
  :                                                                                                                             +- Project [_c0#10, _c1#11, time#14L, value#18, unnamed AS channel#23]
  :                                                                                                                                +- Project [_c0#10, _c1#11, time#14L, UDF(_c1#11) AS value#18]
  :                                                                                                                                   +- Project [_c0#10, _c1#11, UDF(_c0#10) AS time#14L]
  :                                                                                                                                      +- Relation[_c0#10,_c1#11] csv
  +- Project [_1#133 AS pprChannel#136, _2#134 AS ppr#137]
     +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._1, true, false) AS _1#133, assertnotnull(assertnotnull(input[0, scala.Tuple2, true]))._2 AS _2#134]
        +- ExternalRDD [obj#132]
  
  == Optimized Logical Plan ==
  org.apache.spark.sql.AnalysisException: Detected cartesian product for INNER join between logical plans
  Project [UDF(t0#93L, t1#29L) AS time#127L, unnamed AS channel#124]
  +- Filter ((isnotnull(pt0#98) &amp;&amp; isnotnull(pt1#82)) &amp;&amp; ((((t0#93L &gt;= 0) &amp;&amp; (pt0#98 = 1)) &amp;&amp; (pt1#82 = 2)) &amp;&amp; ((t1#29L - t0#93L) &gt;= 50000000)))
     +- Window [lag(t1#29L, 1, -1) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#93L, lag(pt1#82, 1, 0) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS pt0#98], [unnamed], [t1#29L ASC NULLS FIRST]
        +- Project [t1#29L, if ((isnull(dv0#75) || isnull(dv1#58))) null else if ((isnull(dv0#75) || isnull(dv1#58))) null else UDF(dv0#75, dv1#58) AS pt1#82]
           +- Filter (((t0#70L &gt;= 0) &amp;&amp; NOT isnan(dv0#75)) &amp;&amp; if ((isnull(dv0#75) || isnull(dv1#58))) null else if ((isnull(dv0#75) || isnull(dv1#58))) null else UDF(dv0#75, dv1#58) IN (1,2))
              +- Window [lag(t1#29L, 1, -1) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#70L, lag(dv1#58, 1, NaN) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS dv0#75], [unnamed], [t1#29L ASC NULLS FIRST]
                 +- Project [t1#29L, abs(UDF(v0#49, v1#35)) AS dv1#58]
                    +- Filter ((t0#42L &gt;= 0) &amp;&amp; NOT isnan(v0#49))
                       +- Window [lag(t1#29L, 1, -1) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#42L, lag(v1#35, 1, NaN) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS v0#49], [unnamed], [t1#29L ASC NULLS FIRST]
                          +- Project [UDF(_c0#10) AS t1#29L, UDF(_c1#11) AS v1#35]
                             +- Filter ((UDF(_c0#10) &gt;= 0) &amp;&amp; NOT isnan(UDF(_c1#11)))
                                +- Relation[_c0#10,_c1#11] csv
  and
  Project [_1#133 AS pprChannel#136, _2#134 AS ppr#137]
  +- Filter (isnotnull(_1#133) &amp;&amp; (unnamed = _1#133))
     +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple2, true])._1, true, false) AS _1#133, assertnotnull(input[0, scala.Tuple2, true])._2 AS _2#134]
        +- ExternalRDD [obj#132]
  Join condition is missing or trivial.
  Use the CROSS JOIN syntax to allow cartesian products between these relations.;
  == Physical Plan ==
  org.apache.spark.sql.AnalysisException: Detected cartesian product for INNER join between logical plans
  Project [UDF(t0#93L, t1#29L) AS time#127L, unnamed AS channel#124]
  +- Filter ((isnotnull(pt0#98) &amp;&amp; isnotnull(pt1#82)) &amp;&amp; ((((t0#93L &gt;= 0) &amp;&amp; (pt0#98 = 1)) &amp;&amp; (pt1#82 = 2)) &amp;&amp; ((t1#29L - t0#93L) &gt;= 50000000)))
     +- Window [lag(t1#29L, 1, -1) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#93L, lag(pt1#82, 1, 0) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS pt0#98], [unnamed], [t1#29L ASC NULLS FIRST]
        +- Project [t1#29L, if ((isnull(dv0#75) || isnull(dv1#58))) null else if ((isnull(dv0#75) || isnull(dv1#58))) null else UDF(dv0#75, dv1#58) AS pt1#82]
           +- Filter (((t0#70L &gt;= 0) &amp;&amp; NOT isnan(dv0#75)) &amp;&amp; if ((isnull(dv0#75) || isnull(dv1#58))) null else if ((isnull(dv0#75) || isnull(dv1#58))) null else UDF(dv0#75, dv1#58) IN (1,2))
              +- Window [lag(t1#29L, 1, -1) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#70L, lag(dv1#58, 1, NaN) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS dv0#75], [unnamed], [t1#29L ASC NULLS FIRST]
                 +- Project [t1#29L, abs(UDF(v0#49, v1#35)) AS dv1#58]
                    +- Filter ((t0#42L &gt;= 0) &amp;&amp; NOT isnan(v0#49))
                       +- Window [lag(t1#29L, 1, -1) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS t0#42L, lag(v1#35, 1, NaN) windowspecdefinition(unnamed, t1#29L ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS v0#49], [unnamed], [t1#29L ASC NULLS FIRST]
                          +- Project [UDF(_c0#10) AS t1#29L, UDF(_c1#11) AS v1#35]
                             +- Filter ((UDF(_c0#10) &gt;= 0) &amp;&amp; NOT isnan(UDF(_c1#11)))
                                +- Relation[_c0#10,_c1#11] csv
  and
  Project [_1#133 AS pprChannel#136, _2#134 AS ppr#137]
  +- Filter (isnotnull(_1#133) &amp;&amp; (unnamed = _1#133))
     +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple2, true])._1, true, false) AS _1#133, assertnotnull(input[0, scala.Tuple2, true])._2 AS _2#134]
        +- ExternalRDD [obj#132]
  Join condition is missing or trivial.
  Use the CROSS JOIN syntax to allow cartesian products between these relations.;
  </code></pre>
  
  <p>Yes, <code>df1</code> is a result of a fairly complex computation, that's why it is so big. <code>df2</code> is a very small DF which always comes from a <code>Map</code> with at most about 50 to 100 entries brought to Spark with <code>sc.parallelize</code>. So I could use <code>crossJoin</code> and a <code>where</code> as a workaround. But I want to understand why Spark thinks it is a cartesian product.</p>
  
  <h1>Followup 2</h1>
  
  <p>I am now using a different approach. Since the first DF is this huge one which is the result of a complex calculation, and the second one is always originating from a small map, I changed my algorithm to use ordinary <code>map</code> operations to achieve it:</p>
  
  <pre><code>val bDF2Data = sc.broadcast(df2Data)
  val res =
    df1.
      as[(Long, String)].
      mapPartitions { iter =&gt;
        val df2Data = bDF2Data.value
        iter.
          flatMap {
            case (time, channel) =&gt;
              df2Data.get(channel).map(ppr =&gt; (time, channel, ppr))
          }
      }.
      toDF("time", "channel", "ppr").
      // More operations ...
  </code></pre>