[
https://issues.apache.org/jira/browse/SPARK-20073?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15939844#comment-15939844
]
Everett Anderson commented on SPARK-20073:
------------------------------------------
[~maropu]
Hi! Thanks for taking a look. What's the Scala equivalent of the aliasing
necessary?
I've tried calling .toDF(column names) on both tables and assigning table
aliases like this --
{noformat}
val replacedPeople = people.toDF(people.columns:_*).alias("replacedPeople")
val replacedVariantCounts =
variantCounts.toDF(variantCounts.columns:_*).alias("replacedVariantCounts")
replacedVariantCounts.join(replacedPeople,
replacedVariantCounts("name")<=>(replacedPeople("name"))).show
{noformat}
but that doesn't seem to work work.
> Unexpected Cartesian product when using eqNullSafe in join with a derived
> table
> -------------------------------------------------------------------------------
>
> Key: SPARK-20073
> URL: https://issues.apache.org/jira/browse/SPARK-20073
> Project: Spark
> Issue Type: Bug
> Components: Optimizer
> Affects Versions: 2.0.2, 2.1.0
> Reporter: Everett Anderson
> Labels: correctness
>
> It appears that if you try to join tables A and B when B is derived from A
> and you use the eqNullSafe / <=> operator for the join condition, Spark
> performs a Cartesian product.
> However, if you perform the join on tables of the same data when they don't
> have a relationship, the expected non-Cartesian product join occurs.
> {noformat}
> // Create some fake data.
> import org.apache.spark.sql.Row
> import org.apache.spark.sql.Dataset
> import org.apache.spark.sql.types._
> import org.apache.spark.sql.functions
> val peopleRowsRDD = sc.parallelize(Seq(
> Row("Fred", 8, 1),
> Row("Fred", 8, 2),
> Row(null, 10, 3),
> Row(null, 10, 4),
> Row("Amy", 12, 5),
> Row("Amy", 12, 6)))
>
> val peopleSchema = StructType(Seq(
> StructField("name", StringType, nullable = true),
> StructField("group", IntegerType, nullable = true),
> StructField("data", IntegerType, nullable = true)))
>
> val people = spark.createDataFrame(peopleRowsRDD, peopleSchema)
> people.createOrReplaceTempView("people")
> scala> people.show
> +----+-----+----+
> |name|group|data|
> +----+-----+----+
> |Fred| 8| 1|
> |Fred| 8| 2|
> |null| 10| 3|
> |null| 10| 4|
> | Amy| 12| 5|
> | Amy| 12| 6|
> +----+-----+----+
> // Now create a derived table from that table. It doesn't matter much what.
> val variantCounts = spark.sql("select name, count(distinct(name, group,
> data)) as variant_count from people group by name having variant_count > 1")
> variantCounts.show
> +----+-------------+
>
> |name|variant_count|
> +----+-------------+
> |Fred| 2|
> |null| 2|
> | Amy| 2|
> +----+-------------+
> // Now try an inner join using the regular equalTo that drops nulls. This
> works fine.
> val innerJoinEqualTo = variantCounts.join(people,
> variantCounts("name").equalTo(people("name")))
> innerJoinEqualTo.show
> +----+-------------+----+-----+----+
>
> |name|variant_count|name|group|data|
> +----+-------------+----+-----+----+
> |Fred| 2|Fred| 8| 1|
> |Fred| 2|Fred| 8| 2|
> | Amy| 2| Amy| 12| 5|
> | Amy| 2| Amy| 12| 6|
> +----+-------------+----+-----+----+
> // Okay now lets switch to the <=> operator
> //
> // If you haven't set spark.sql.crossJoin.enabled=true, you'll get an error
> like
> // "Cartesian joins could be prohibitively expensive and are disabled by
> default. To explicitly enable them, please set spark.sql.crossJoin.enabled =
> true;"
> //
> // if you have enabled them, you'll get the table below.
> //
> // However, we really don't want or expect a Cartesian product!
> val innerJoinSqlNullSafeEqOp = variantCounts.join(people,
> variantCounts("name")<=>(people("name")))
> innerJoinSqlNullSafeEqOp.show
> +----+-------------+----+-----+----+
>
> |name|variant_count|name|group|data|
> +----+-------------+----+-----+----+
> |Fred| 2|Fred| 8| 1|
> |Fred| 2|Fred| 8| 2|
> |Fred| 2|null| 10| 3|
> |Fred| 2|null| 10| 4|
> |Fred| 2| Amy| 12| 5|
> |Fred| 2| Amy| 12| 6|
> |null| 2|Fred| 8| 1|
> |null| 2|Fred| 8| 2|
> |null| 2|null| 10| 3|
> |null| 2|null| 10| 4|
> |null| 2| Amy| 12| 5|
> |null| 2| Amy| 12| 6|
> | Amy| 2|Fred| 8| 1|
> | Amy| 2|Fred| 8| 2|
> | Amy| 2|null| 10| 3|
> | Amy| 2|null| 10| 4|
> | Amy| 2| Amy| 12| 5|
> | Amy| 2| Amy| 12| 6|
> +----+-------------+----+-----+----+
> // Okay, let's try to construct the exact same variantCount table manually
> // so it has no relationship to the original.
> val variantCountRowsRDD = sc.parallelize(Seq(
> Row("Fred", 2),
> Row(null, 2),
> Row("Amy", 2)))
>
> val variantCountSchema = StructType(Seq(
> StructField("name", StringType, nullable = true),
> StructField("variant_count", IntegerType, nullable = true)))
>
> val manualVariantCounts = spark.createDataFrame(variantCountRowsRDD,
> variantCountSchema)
> // Now perform the same join with the null-safe equals operator. This works
> and gives us the expected non-Cartesian product result.
> val manualVarCountsInnerJoinSqlNullSafeEqOp =
> manualVariantCounts.join(people,
> manualVariantCounts("name")<=>(people("name")))
> manualVarCountsInnerJoinSqlNullSafeEqOp.show
> +----+-------------+----+-----+----+
> |name|variant_count|name|group|data|
> +----+-------------+----+-----+----+
> |Fred| 2|Fred| 8| 1|
> |Fred| 2|Fred| 8| 2|
> | Amy| 2| Amy| 12| 5|
> | Amy| 2| Amy| 12| 6|
> |null| 2|null| 10| 3|
> |null| 2|null| 10| 4|
> +----+-------------+----+-----+----+
> {noformat}
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
