You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ryan-johnson-databricks (via GitHub)" <gi...@apache.org> on 2023/03/08 17:46:03 UTC
[GitHub] [spark] ryan-johnson-databricks commented on a diff in pull request #40300: [SPARK-42683] Automatically rename conflicting metadata columns
ryan-johnson-databricks commented on code in PR #40300:
URL: https://github.com/apache/spark/pull/40300#discussion_r1127830262
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala:
##########
@@ -42,6 +42,24 @@ abstract class LogicalPlan
*/
def metadataOutput: Seq[Attribute] = children.flatMap(_.metadataOutput)
+ /**
+ * Finds a metadata attribute of this node by its logical name. This search will work even if the
+ * metadata attribute was renamed because of a conflicting name in the data schema.
+ */
+ def getMetadataAttributeByName(name: String): Attribute = {
+ // NOTE: An already-referenced column might appear in `output` instead of `metadataOutput`.
Review Comment:
I originally added it because I was also hitting cases where `metadataOutput` did not contain the metadata column.
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2714,6 +2726,17 @@ class Dataset[T] private[sql](
*/
def withColumn(colName: String, col: Column): DataFrame = withColumns(Seq(colName), Seq(col))
+ /**
+ * Returns a new Dataset by selecting a metadata column with the given logical name.
+ *
+ * A metadata column can be accessed this way even if the underlying data source defines a data
+ * column with a conflicting name.
+ *
+ * @group untypedrel
+ * @since 4.0.0
+ */
+ def withMetadataColumn(colName: String): DataFrame = withColumn(colName, metadataColumn(colName))
Review Comment:
> people can just do `df.withColumn(name, df.metadataColumn(colName))`, right?
Yes. Fair point.
It does lead to possible weirdness, in case the column obtained by `df.metadataColumn` is added to some other DF, e.g.
```scala
df.select(...).as("view").withColumn("m", df.metadataColumn("_metadata"))
```
...but the same hazard exists for normal columns already, so it's probably fine?
```scala
df.select(...).as("view").withColumn("m", df.col("other"))
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala:
##########
@@ -42,6 +42,24 @@ abstract class LogicalPlan
*/
def metadataOutput: Seq[Attribute] = children.flatMap(_.metadataOutput)
+ /**
+ * Finds a metadata attribute of this node by its logical name. This search will work even if the
+ * metadata attribute was renamed because of a conflicting name in the data schema.
+ */
+ def getMetadataAttributeByName(name: String): Attribute = {
+ // NOTE: An already-referenced column might appear in `output` instead of `metadataOutput`.
Review Comment:
`SubqueryAlias` does not always propagate its child's `metadataOutput`, even if the child's own `output` includes metadata columns. That can result in a query plan whose `output` contains attributes which satisfy `MetadataAttribute` but are not in `metadataOutput`.
I raised https://github.com/apache/spark/pull/40321 as a possible fix for that issue.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala:
##########
@@ -42,6 +42,24 @@ abstract class LogicalPlan
*/
def metadataOutput: Seq[Attribute] = children.flatMap(_.metadataOutput)
+ /**
+ * Finds a metadata attribute of this node by its logical name. This search will work even if the
+ * metadata attribute was renamed because of a conflicting name in the data schema.
+ */
+ def getMetadataAttributeByName(name: String): Attribute = {
+ // NOTE: An already-referenced column might appear in `output` instead of `metadataOutput`.
+ (metadataOutput ++ output).collectFirst {
+ case MetadataAttributeWithLogicalName(attr, logicalName)
+ if conf.resolver(name, logicalName) => attr
+ }.getOrElse {
+ val availableMetadataColumns = (metadataOutput ++ output).collect {
+ case MetadataAttributeWithLogicalName(_, logicalName) => logicalName
Review Comment:
[MetadataAttributeWithLogicalName.unapply](https://github.com/apache/spark/pull/40300/files/8b5c9bf7ef1c465d6b7bd17a24f3750a5ce727f8#diff-cf96171d13fd77e670764766ae22afafbc4a396316bd758a89b60a6fe70d5b0dR517) already checks all metadata attributes, because a metadata column's physical is also its logical name unless/until renamed. Otherwise, every site that uses the matcher would also have to check for not-renamed columns, which is error-prone and redundant.
The doc comment for the matcher tries to make that clear?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -244,6 +245,89 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
parameters = Map("fieldName" -> "`file_name`", "fields" -> "`id`, `university`"))
}
+ metadataColumnsTest("df metadataColumn - schema conflict",
+ schemaWithNameConflicts) { (df, f0, f1) =>
+ // the user data has the schema: name, age, _metadata.id, _metadata.university
+
+ // get the real metadata column (whose name should have been adjusted)
+ val metadataColumn = df.metadataColumn("_metadata")
+ assert(metadataColumn.expr.asInstanceOf[NamedExpression].name == "__metadata")
+
+ // select user data
+ checkAnswer(
+ df.select("name", "age", "_METADATA", "_metadata")
+ .withColumn("file_name", metadataColumn.getField("file_name")),
+ Seq(
+ Row("jack", 24, Row(12345L, "uom"), Row(12345L, "uom"), f0(METADATA_FILE_NAME)),
+ Row("lily", 31, Row(54321L, "ucb"), Row(54321L, "ucb"), f1(METADATA_FILE_NAME))
+ )
+ )
+ }
+
+ metadataColumnsTest("df metadataColumn - no schema conflict",
+ schema) { (df, f0, f1) =>
+ // get the real metadata column (whose name should _NOT_ have been adjusted)
+ val metadataColumn = df.metadataColumn("_metadata")
+ assert(metadataColumn.expr.asInstanceOf[NamedExpression].name == "_metadata")
+
+ // select user data
+ checkAnswer(
+ df.select("name", "age")
+ .withColumn("file_name", metadataColumn.getField("file_name")),
+ Seq(
+ Row("jack", 24, f0(METADATA_FILE_NAME)),
+ Row("lily", 31, f1(METADATA_FILE_NAME))
+ )
+ )
+ }
+
+ metadataColumnsTest("df metadataColumn - column not found", schema) { (df, f0, f1) =>
+ // Not a column at all
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.withMetadataColumn("foo")
+ },
+ errorClass = "UNRESOLVED_COLUMN.WITH_SUGGESTION",
+ parameters = Map("objectName" -> "`foo`", "proposal" -> "`_metadata`"))
+
+ // Name exists, but does not reference a metadata column
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.withMetadataColumn("name")
+ },
+ errorClass = "UNRESOLVED_COLUMN.WITH_SUGGESTION",
+ parameters = Map("objectName" -> "`name`", "proposal" -> "`_metadata`"))
+ }
+
+ metadataColumnsTest("metadata name conflict resolved with leading underscores - one",
+ schemaWithNameConflicts) { (df, f0, f1) =>
+ // the user data has the schema: name, age, _metadata.id, _metadata.university
+
+ checkAnswer(
+ df.select("name", "age", "_metadata", "__metadata.file_name"),
+ Seq(
+ Row("jack", 24, Row(12345L, "uom"), f0(METADATA_FILE_NAME)),
+ Row("lily", 31, Row(54321L, "ucb"), f1(METADATA_FILE_NAME))
+ )
+ )
+ }
+
+ metadataColumnsTest("metadata name conflict resolved with leading underscores - several",
+ new StructType()
+ .add(schema("name").copy(name = "_metadata"))
+ .add(schema("age").copy(name = "__metadata"))
+ .add(schema("info").copy(name = "___metadata"))) { (df, f0, f1) =>
+ // the user data has the schema: _metadata, __metadata, ___metadata.id, ___metadata.university
+
+ checkAnswer(
+ df.select("_metadata", "__metadata", "___metadata", "____metadata.file_name"),
+ Seq(
+ Row("jack", 24, Row(12345L, "uom"), f0(METADATA_FILE_NAME)),
+ Row("lily", 31, Row(54321L, "ucb"), f1(METADATA_FILE_NAME))
+ )
+ )
+ }
+
Review Comment:
Where/how should I document it? (easy enough to add a test case, for example?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org