You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "johanl-db (via GitHub)" <gi...@apache.org> on 2023/03/24 11:43:48 UTC
[GitHub] [spark] johanl-db opened a new pull request, #40545: [WIP][SPARK-42918] Introduce abstractions to create constant and generated metadata fields
johanl-db opened a new pull request, #40545:
URL: https://github.com/apache/spark/pull/40545
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This change refactors the metadata attribute introduced in https://github.com/apache/spark/pull/39314 to allow easier creation and handling of constant and generated metadata columns.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
--
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
[GitHub] [spark] ryan-johnson-databricks commented on a diff in pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "ryan-johnson-databricks (via GitHub)" <gi...@apache.org>.
ryan-johnson-databricks commented on code in PR #40545:
URL: https://github.com/apache/spark/pull/40545#discussion_r1150634235
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -236,33 +247,42 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// For generated metadata columns, they are set as nullable when passed to readers,
// as the values will be null when trying to read the missing column from the file.
// They are then replaced by the actual values later in the process.
- // All metadata columns will be non-null in the returned output.
- // We then change the nullability to non-nullable in the metadata projection node below.
- val constantMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
- val generatedMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
+ // We then restore the specified nullability in the metadata projection node below.
+ // Also remember the attribute for each column name, so we can easily map back to it.
+ val constantMetadataColumns = mutable.Buffer.empty[Attribute]
+ val generatedMetadataColumns = mutable.Buffer.empty[Attribute]
+ val metadataColumnsByName = mutable.Map.empty[String, Attribute]
metadataStructOpt.foreach { metadataStruct =>
- metadataStruct.dataType.asInstanceOf[StructType].fields.foreach { field =>
- field.name match {
- case FileFormat.ROW_INDEX =>
- if ((readDataColumns ++ partitionColumns).map(_.name.toLowerCase(Locale.ROOT))
- .contains(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)) {
- throw new AnalysisException(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME +
- " is a reserved column name that cannot be read in combination with " +
- s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} column.")
- }
- generatedMetadataColumns +=
- FileSourceGeneratedMetadataAttribute(
- FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = true)
- case _ =>
- constantMetadataColumns +=
- FileSourceConstantMetadataAttribute(field.name, field.dataType)
- }
+ val schemaColumns = (readDataColumns ++ partitionColumns)
+ .map(_.name.toLowerCase(Locale.ROOT))
+ .toSet
+
+ def createMetadataColumn(field: StructField) = field match {
Review Comment:
I'm not convinced that introducing this single-use helper improves readability or maintainability?
```scala
def helper(field: StructField) = field match {
case ...
case ...
}
metadataStruct.dataType.asInstanceOf[StructType].fields.foreach(createMetadataColumn)
```
vs. just inlining the logic:
```scala
metadataStruct.dataType.asInstanceOf[StructType].fields.foreach {
case ...
case ...
}
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -236,33 +247,42 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// For generated metadata columns, they are set as nullable when passed to readers,
// as the values will be null when trying to read the missing column from the file.
// They are then replaced by the actual values later in the process.
- // All metadata columns will be non-null in the returned output.
- // We then change the nullability to non-nullable in the metadata projection node below.
- val constantMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
- val generatedMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
+ // We then restore the specified nullability in the metadata projection node below.
+ // Also remember the attribute for each column name, so we can easily map back to it.
+ val constantMetadataColumns = mutable.Buffer.empty[Attribute]
+ val generatedMetadataColumns = mutable.Buffer.empty[Attribute]
+ val metadataColumnsByName = mutable.Map.empty[String, Attribute]
metadataStructOpt.foreach { metadataStruct =>
- metadataStruct.dataType.asInstanceOf[StructType].fields.foreach { field =>
- field.name match {
- case FileFormat.ROW_INDEX =>
- if ((readDataColumns ++ partitionColumns).map(_.name.toLowerCase(Locale.ROOT))
- .contains(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)) {
- throw new AnalysisException(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME +
- " is a reserved column name that cannot be read in combination with " +
- s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} column.")
- }
- generatedMetadataColumns +=
- FileSourceGeneratedMetadataAttribute(
- FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = true)
- case _ =>
- constantMetadataColumns +=
- FileSourceConstantMetadataAttribute(field.name, field.dataType)
- }
+ val schemaColumns = (readDataColumns ++ partitionColumns)
+ .map(_.name.toLowerCase(Locale.ROOT))
+ .toSet
+
+ def createMetadataColumn(field: StructField) = field match {
+ case FileSourceGeneratedMetadataStructField(field, internalName) =>
+ if (schemaColumns.contains(internalName)) {
+ throw new AnalysisException(internalName +
+ s"${internalName} is a reserved column name that cannot be read in combination " +
+ s"with ${FileFormat.METADATA_NAME}.${field.name} column.")
+ }
+
+ // NOTE: Readers require the internal column to be nullable because it's not part of the
+ // file's public schema. The projection below will restore the correct nullability for
+ // the column while constructing the final metadata struct.
+ val attr = field.copy(internalName, nullable = true).toAttribute
+ metadataColumnsByName.put(field.name, attr)
+ generatedMetadataColumns += attr
+
+ case FileSourceConstantMetadataStructField(field) =>
+ val attr = field.toAttribute
+ metadataColumnsByName.put(field.name, attr)
+ constantMetadataColumns += attr
+
+ case field => throw new AnalysisException(s"Unrecognized file metadata field: $field")
}
- }
- val metadataColumns: Seq[Attribute] =
- constantMetadataColumns.toSeq ++ generatedMetadataColumns.toSeq
+ metadataStruct.dataType.asInstanceOf[StructType].fields.foreach(createMetadataColumn)
Review Comment:
Rather than cast the struct here, would it make sense to directly match on it?
```scala
metadataStructOpt.foreach { case AttributeReference(_, schema: StructType, _, _) =>
...
schema.fields.foreach(...)
}
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala:
##########
@@ -501,80 +506,93 @@ object FileSourceMetadataAttribute {
val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ val METADATA: Metadata = new MetadataBuilder()
+ .withMetadata(MetadataAttribute.METADATA)
+ .putBoolean(METADATA_COL_ATTR_KEY, value = true)
Review Comment:
Isn't this already part of `MetadataAttribute.METADATA`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -282,29 +302,21 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// all references will be bound to output attributes which are either
// [[FileSourceConstantMetadataAttribute]] or [[FileSourceGeneratedMetadataAttribute]] after
// the flattening from the metadata struct.
- def rebindFileSourceMetadataAttributesInFilters(
- filters: Seq[Expression]): Seq[Expression] = {
- // The row index field attribute got renamed.
- def newFieldName(name: String) = name match {
- case FileFormat.ROW_INDEX => FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
- case other => other
- }
-
+ def rebindFileSourceMetadataAttributesInFilters(filters: Seq[Expression]): Seq[Expression] =
filters.map { filter =>
filter.transform {
// Replace references to the _metadata column. This will affect references to the column
// itself but also where fields from the metadata struct are used.
case MetadataStructColumn(AttributeReference(_, fields @ StructType(_), _, _)) =>
Review Comment:
aside: `fields @ StructType(_)` is equialent to `fields: StructType`, no?
--
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
[GitHub] [spark] cloud-fan closed pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
URL: https://github.com/apache/spark/pull/40545
--
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
[GitHub] [spark] johanl-db commented on a diff in pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on code in PR #40545:
URL: https://github.com/apache/spark/pull/40545#discussion_r1149566989
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala:
##########
@@ -519,6 +519,13 @@ object FileSourceMetadataAttribute {
def cleanupFileSourceMetadataInformation(attr: Attribute): Attribute =
attr.withMetadata(removeInternalMetadata(attr.metadata))
+ /**
+ * Cleanup the internal metadata information of a struct field, if it is
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -221,14 +225,23 @@ object FileFormat {
FileSourceConstantMetadataStructField(FILE_MODIFICATION_TIME, TimestampType, nullable = false))
/**
- * Create a file metadata struct column containing fields supported by the given file format.
+ * All fields the file format's _metadata struct defines.
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -322,18 +329,13 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// extra Project node: wrap flat metadata columns to a metadata struct
val withMetadataProjections = metadataStructOpt.map { metadataStruct =>
- val structColumns = metadataColumns.map { col => col.name match {
- case FileFormat.FILE_PATH | FileFormat.FILE_NAME | FileFormat.FILE_SIZE |
- FileFormat.FILE_BLOCK_START | FileFormat.FILE_BLOCK_LENGTH |
- FileFormat.FILE_MODIFICATION_TIME =>
- col
- case FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME =>
- generatedMetadataColumns
- .find(_.name == FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)
- // Change the `_tmp_metadata_row_index` to `row_index`,
- // and also change the nullability to not nullable,
- // which is consistent with the nullability of `row_index` field
- .get.withName(FileFormat.ROW_INDEX).withNullability(false)
+ val structColumns = metadataStruct.dataType.asInstanceOf[StructType].fields.map { field =>
+ // Construct the metadata struct the query expects to see, using the columns we previously
+ // created. Be sure to restore the proper name and nullability for each metadata field.
+ metadataColumnsByName(field.name) match {
Review Comment:
I simplified this code to always restore the name and nullability, I don't think it warrants a method anymore.
I actually think it was not correct before that since you could have a generated column created with `nullable = true` on [L272](https://github.com/apache/spark/pull/40545/commits/2f5f9a26ec5cdf0c04893a18e2cfa386228e7aa2#diff-7e4f78e90e8699733afbe43e2b265b95e514896ac68f1fb9e60705d59a0b7ed9R272) and nullability wouldn't be reset if its `name` and `internalName` are the same
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -236,37 +247,41 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// For generated metadata columns, they are set as nullable when passed to readers,
// as the values will be null when trying to read the missing column from the file.
// They are then replaced by the actual values later in the process.
- // All metadata columns will be non-null in the returned output.
- // We then change the nullability to non-nullable in the metadata projection node below.
- val constantMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
- val generatedMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
+ // We then restore the specified nullability in the metadata projection node below.
+ // Also remember the attribute for each column name, so we can easily map back to it.
+ val constantMetadataColumns = mutable.Buffer.empty[Attribute]
+ val generatedMetadataColumns = mutable.Buffer.empty[Attribute]
+ val metadataColumnsByName = mutable.Map.empty[String, Attribute]
metadataStructOpt.foreach { metadataStruct =>
- metadataStruct.dataType.asInstanceOf[StructType].fields.foreach { field =>
- field.name match {
- case FileFormat.ROW_INDEX =>
- if ((readDataColumns ++ partitionColumns).map(_.name.toLowerCase(Locale.ROOT))
- .contains(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)) {
- throw new AnalysisException(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME +
- " is a reserved column name that cannot be read in combination with " +
- s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} column.")
- }
- generatedMetadataColumns +=
- FileSourceGeneratedMetadataStructField(
- name = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME,
- internalName = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME,
- dataType = LongType,
- nullable = true).toAttribute
- case _ =>
- constantMetadataColumns +=
- FileSourceConstantMetadataStructField(field.name, field.dataType).toAttribute
- }
+ val schemaColumns = (readDataColumns ++ partitionColumns)
+ .map(_.name.toLowerCase(Locale.ROOT))
+ .toSet
+
+ metadataStruct.dataType.asInstanceOf[StructType].fields.foreach {
Review Comment:
Added method `createMetadataColumn`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -221,14 +225,23 @@ object FileFormat {
FileSourceConstantMetadataStructField(FILE_MODIFICATION_TIME, TimestampType, nullable = false))
/**
- * Create a file metadata struct column containing fields supported by the given file format.
+ * All fields the file format's _metadata struct defines.
*/
- def createFileMetadataCol(fileFormat: FileFormat): AttributeReference = {
- val fields = if (fileFormat.isInstanceOf[ParquetFileFormat]) {
- BASE_METADATA_FIELDS :+ StructField(FileFormat.ROW_INDEX, LongType, nullable = false)
+ def metadataSchemaFields(fileFormat: FileFormat): Seq[StructField] =
+ if (fileFormat.isInstanceOf[ParquetFileFormat]) {
+ BASE_METADATA_FIELDS :+ ROW_INDEX_FIELD
} else {
BASE_METADATA_FIELDS
}
+
+ /**
+ * Create a file metadata struct column containing fields supported by the given file format.
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -198,6 +198,10 @@ object FileFormat {
// until they can be placed in the _metadata struct.
val ROW_INDEX_TEMPORARY_COLUMN_NAME = s"_tmp_metadata_$ROW_INDEX"
+ // The field readers can use to access the generated row index column.
Review Comment:
Removed
--
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
[GitHub] [spark] ryan-johnson-databricks commented on a diff in pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "ryan-johnson-databricks (via GitHub)" <gi...@apache.org>.
ryan-johnson-databricks commented on code in PR #40545:
URL: https://github.com/apache/spark/pull/40545#discussion_r1150611339
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala:
##########
@@ -501,80 +506,93 @@ object FileSourceMetadataAttribute {
val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ val METADATA: Metadata = new MetadataBuilder()
+ .withMetadata(MetadataAttribute.METADATA)
+ .putBoolean(METADATA_COL_ATTR_KEY, value = true)
+ .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true)
+ .build()
+
/**
- * Cleanup the internal metadata information of an attribute if it is
- * a [[FileSourceConstantMetadataAttribute]] or [[FileSourceGeneratedMetadataAttribute]].
+ * Removes the internal field metadata.
*/
def cleanupFileSourceMetadataInformation(attr: Attribute): Attribute =
- removeInternalMetadata(attr)
+ attr.withMetadata(removeInternalMetadata(attr.metadata))
Review Comment:
I only see a call site for the `StructField` overload. Now that `FileFormat` starts from a `StructType` and converts to `AttributeReference` from there, do we expect to still need the `Attribute` version of this method going forward?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala:
##########
@@ -501,80 +506,93 @@ object FileSourceMetadataAttribute {
val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ val METADATA: Metadata = new MetadataBuilder()
+ .withMetadata(MetadataAttribute.METADATA)
+ .putBoolean(METADATA_COL_ATTR_KEY, value = true)
+ .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true)
+ .build()
+
/**
- * Cleanup the internal metadata information of an attribute if it is
- * a [[FileSourceConstantMetadataAttribute]] or [[FileSourceGeneratedMetadataAttribute]].
+ * Removes the internal field metadata.
*/
def cleanupFileSourceMetadataInformation(attr: Attribute): Attribute =
- removeInternalMetadata(attr)
+ attr.withMetadata(removeInternalMetadata(attr.metadata))
+
+ /**
+ * Removes the internal field metadata.
+ */
+ def cleanupFileSourceMetadataInformation(field: StructField): StructField =
+ field.copy(metadata = removeInternalMetadata(field.metadata))
def apply(name: String, dataType: DataType, nullable: Boolean = false): AttributeReference =
- AttributeReference(name, dataType, nullable = nullable,
- new MetadataBuilder()
- .putBoolean(METADATA_COL_ATTR_KEY, value = true)
- .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true).build())()
+ AttributeReference(name, dataType, nullable = nullable, METADATA)()
/** Matches if attr is any File source metadata attribute (including constant and generated). */
- def unapply(attr: AttributeReference): Option[AttributeReference] =
- attr match {
- case MetadataAttribute(attr)
- if attr.metadata.contains(FILE_SOURCE_METADATA_COL_ATTR_KEY)
- && attr.metadata.getBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY) => Some(attr)
- case _ => None
- }
+ def unapply(attr: AttributeReference): Option[AttributeReference] = {
Review Comment:
tiny nit: the method doesn't require `{}` and removing it makes a cleaner diff
--
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
[GitHub] [spark] cloud-fan commented on pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #40545:
URL: https://github.com/apache/spark/pull/40545#issuecomment-1491879620
thanks, merging to master!
--
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
[GitHub] [spark] jaceklaskowski commented on a diff in pull request #40545: [WIP][SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "jaceklaskowski (via GitHub)" <gi...@apache.org>.
jaceklaskowski commented on code in PR #40545:
URL: https://github.com/apache/spark/pull/40545#discussion_r1148384721
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -198,6 +198,10 @@ object FileFormat {
// until they can be placed in the _metadata struct.
val ROW_INDEX_TEMPORARY_COLUMN_NAME = s"_tmp_metadata_$ROW_INDEX"
+ // The field readers can use to access the generated row index column.
Review Comment:
I don't think the description says anything useful in addition to what's implied anyway from `ROW_INDEX_FIELD` being a public val (_constant_) that others (e.g., "readers") could use. Please remove the comment.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -236,37 +247,41 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// For generated metadata columns, they are set as nullable when passed to readers,
// as the values will be null when trying to read the missing column from the file.
// They are then replaced by the actual values later in the process.
- // All metadata columns will be non-null in the returned output.
- // We then change the nullability to non-nullable in the metadata projection node below.
- val constantMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
- val generatedMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
+ // We then restore the specified nullability in the metadata projection node below.
+ // Also remember the attribute for each column name, so we can easily map back to it.
+ val constantMetadataColumns = mutable.Buffer.empty[Attribute]
+ val generatedMetadataColumns = mutable.Buffer.empty[Attribute]
+ val metadataColumnsByName = mutable.Map.empty[String, Attribute]
metadataStructOpt.foreach { metadataStruct =>
- metadataStruct.dataType.asInstanceOf[StructType].fields.foreach { field =>
- field.name match {
- case FileFormat.ROW_INDEX =>
- if ((readDataColumns ++ partitionColumns).map(_.name.toLowerCase(Locale.ROOT))
- .contains(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)) {
- throw new AnalysisException(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME +
- " is a reserved column name that cannot be read in combination with " +
- s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} column.")
- }
- generatedMetadataColumns +=
- FileSourceGeneratedMetadataStructField(
- name = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME,
- internalName = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME,
- dataType = LongType,
- nullable = true).toAttribute
- case _ =>
- constantMetadataColumns +=
- FileSourceConstantMetadataStructField(field.name, field.dataType).toAttribute
- }
+ val schemaColumns = (readDataColumns ++ partitionColumns)
+ .map(_.name.toLowerCase(Locale.ROOT))
+ .toSet
+
+ metadataStruct.dataType.asInstanceOf[StructType].fields.foreach {
Review Comment:
Can we introduce a (even internal) method with a name that says what this `foreach` does? Or even a helper function that `foreach` uses for the check. It's going to make comprehension so much easier 🙏
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala:
##########
@@ -519,6 +519,13 @@ object FileSourceMetadataAttribute {
def cleanupFileSourceMetadataInformation(attr: Attribute): Attribute =
attr.withMetadata(removeInternalMetadata(attr.metadata))
+ /**
+ * Cleanup the internal metadata information of a struct field, if it is
Review Comment:
nit: Just "Removes the internal field metadata" would be enough. It's not clear even from the implementation itself that it's important that the `StructField` be of the types mentioned.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -221,14 +225,23 @@ object FileFormat {
FileSourceConstantMetadataStructField(FILE_MODIFICATION_TIME, TimestampType, nullable = false))
/**
- * Create a file metadata struct column containing fields supported by the given file format.
+ * All fields the file format's _metadata struct defines.
Review Comment:
Replace with "Supported metadata fields of the given [[FileFormat]]"
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -221,14 +225,23 @@ object FileFormat {
FileSourceConstantMetadataStructField(FILE_MODIFICATION_TIME, TimestampType, nullable = false))
/**
- * Create a file metadata struct column containing fields supported by the given file format.
+ * All fields the file format's _metadata struct defines.
*/
- def createFileMetadataCol(fileFormat: FileFormat): AttributeReference = {
- val fields = if (fileFormat.isInstanceOf[ParquetFileFormat]) {
- BASE_METADATA_FIELDS :+ StructField(FileFormat.ROW_INDEX, LongType, nullable = false)
+ def metadataSchemaFields(fileFormat: FileFormat): Seq[StructField] =
+ if (fileFormat.isInstanceOf[ParquetFileFormat]) {
+ BASE_METADATA_FIELDS :+ ROW_INDEX_FIELD
} else {
BASE_METADATA_FIELDS
}
+
+ /**
+ * Create a file metadata struct column containing fields supported by the given file format.
Review Comment:
nit: Creates...[[FileFormat]]
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -322,18 +329,13 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// extra Project node: wrap flat metadata columns to a metadata struct
val withMetadataProjections = metadataStructOpt.map { metadataStruct =>
- val structColumns = metadataColumns.map { col => col.name match {
- case FileFormat.FILE_PATH | FileFormat.FILE_NAME | FileFormat.FILE_SIZE |
- FileFormat.FILE_BLOCK_START | FileFormat.FILE_BLOCK_LENGTH |
- FileFormat.FILE_MODIFICATION_TIME =>
- col
- case FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME =>
- generatedMetadataColumns
- .find(_.name == FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)
- // Change the `_tmp_metadata_row_index` to `row_index`,
- // and also change the nullability to not nullable,
- // which is consistent with the nullability of `row_index` field
- .get.withName(FileFormat.ROW_INDEX).withNullability(false)
+ val structColumns = metadataStruct.dataType.asInstanceOf[StructType].fields.map { field =>
+ // Construct the metadata struct the query expects to see, using the columns we previously
+ // created. Be sure to restore the proper name and nullability for each metadata field.
+ metadataColumnsByName(field.name) match {
Review Comment:
Can we introduce a method `restoreProperNameAndNullability(field: StructField)` for this comment and the code that follows? We could then remove the comment altogether since the method name would say it all.
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #40545:
URL: https://github.com/apache/spark/pull/40545#discussion_r1153976307
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -220,9 +220,20 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
attributeReference match {
case attr @ FileSourceMetadataAttribute(
Review Comment:
I'm a bit confused. We remove all file source related metadata from the metadata col attribute in https://github.com/apache/spark/pull/40545/files#diff-4445cc3828e35092eb261467b499b8b0ef69ae694ea8ce25abf16b8ef4b72fbaR240 , how are we able to match it with `FileSourceMetadataAttribute` here?
--
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
[GitHub] [spark] johanl-db commented on a diff in pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on code in PR #40545:
URL: https://github.com/apache/spark/pull/40545#discussion_r1150727764
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala:
##########
@@ -501,80 +506,93 @@ object FileSourceMetadataAttribute {
val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ val METADATA: Metadata = new MetadataBuilder()
+ .withMetadata(MetadataAttribute.METADATA)
+ .putBoolean(METADATA_COL_ATTR_KEY, value = true)
+ .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true)
+ .build()
+
/**
- * Cleanup the internal metadata information of an attribute if it is
- * a [[FileSourceConstantMetadataAttribute]] or [[FileSourceGeneratedMetadataAttribute]].
+ * Removes the internal field metadata.
*/
def cleanupFileSourceMetadataInformation(attr: Attribute): Attribute =
- removeInternalMetadata(attr)
+ attr.withMetadata(removeInternalMetadata(attr.metadata))
Review Comment:
[FileFormatWriter](https://github.com/apache/spark/blob/b36d1484c1a090a33d9add056730128b9ba5729f/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala#L111) still uses it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -236,33 +247,42 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// For generated metadata columns, they are set as nullable when passed to readers,
// as the values will be null when trying to read the missing column from the file.
// They are then replaced by the actual values later in the process.
- // All metadata columns will be non-null in the returned output.
- // We then change the nullability to non-nullable in the metadata projection node below.
- val constantMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
- val generatedMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
+ // We then restore the specified nullability in the metadata projection node below.
+ // Also remember the attribute for each column name, so we can easily map back to it.
+ val constantMetadataColumns = mutable.Buffer.empty[Attribute]
+ val generatedMetadataColumns = mutable.Buffer.empty[Attribute]
+ val metadataColumnsByName = mutable.Map.empty[String, Attribute]
metadataStructOpt.foreach { metadataStruct =>
- metadataStruct.dataType.asInstanceOf[StructType].fields.foreach { field =>
- field.name match {
- case FileFormat.ROW_INDEX =>
- if ((readDataColumns ++ partitionColumns).map(_.name.toLowerCase(Locale.ROOT))
- .contains(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)) {
- throw new AnalysisException(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME +
- " is a reserved column name that cannot be read in combination with " +
- s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} column.")
- }
- generatedMetadataColumns +=
- FileSourceGeneratedMetadataAttribute(
- FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = true)
- case _ =>
- constantMetadataColumns +=
- FileSourceConstantMetadataAttribute(field.name, field.dataType)
- }
+ val schemaColumns = (readDataColumns ++ partitionColumns)
+ .map(_.name.toLowerCase(Locale.ROOT))
+ .toSet
+
+ def createMetadataColumn(field: StructField) = field match {
Review Comment:
I don't have a strong opinion on this, reverting to not using a helper method. The comment a few line above should be enough to get context on what happens here.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala:
##########
@@ -501,80 +506,93 @@ object FileSourceMetadataAttribute {
val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ val METADATA: Metadata = new MetadataBuilder()
+ .withMetadata(MetadataAttribute.METADATA)
+ .putBoolean(METADATA_COL_ATTR_KEY, value = true)
Review Comment:
Yes, removed
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -236,33 +247,42 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// For generated metadata columns, they are set as nullable when passed to readers,
// as the values will be null when trying to read the missing column from the file.
// They are then replaced by the actual values later in the process.
- // All metadata columns will be non-null in the returned output.
- // We then change the nullability to non-nullable in the metadata projection node below.
- val constantMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
- val generatedMetadataColumns: mutable.Buffer[Attribute] = mutable.Buffer.empty
+ // We then restore the specified nullability in the metadata projection node below.
+ // Also remember the attribute for each column name, so we can easily map back to it.
+ val constantMetadataColumns = mutable.Buffer.empty[Attribute]
+ val generatedMetadataColumns = mutable.Buffer.empty[Attribute]
+ val metadataColumnsByName = mutable.Map.empty[String, Attribute]
metadataStructOpt.foreach { metadataStruct =>
- metadataStruct.dataType.asInstanceOf[StructType].fields.foreach { field =>
- field.name match {
- case FileFormat.ROW_INDEX =>
- if ((readDataColumns ++ partitionColumns).map(_.name.toLowerCase(Locale.ROOT))
- .contains(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME)) {
- throw new AnalysisException(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME +
- " is a reserved column name that cannot be read in combination with " +
- s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} column.")
- }
- generatedMetadataColumns +=
- FileSourceGeneratedMetadataAttribute(
- FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = true)
- case _ =>
- constantMetadataColumns +=
- FileSourceConstantMetadataAttribute(field.name, field.dataType)
- }
+ val schemaColumns = (readDataColumns ++ partitionColumns)
+ .map(_.name.toLowerCase(Locale.ROOT))
+ .toSet
+
+ def createMetadataColumn(field: StructField) = field match {
+ case FileSourceGeneratedMetadataStructField(field, internalName) =>
+ if (schemaColumns.contains(internalName)) {
+ throw new AnalysisException(internalName +
+ s"${internalName} is a reserved column name that cannot be read in combination " +
+ s"with ${FileFormat.METADATA_NAME}.${field.name} column.")
+ }
+
+ // NOTE: Readers require the internal column to be nullable because it's not part of the
+ // file's public schema. The projection below will restore the correct nullability for
+ // the column while constructing the final metadata struct.
+ val attr = field.copy(internalName, nullable = true).toAttribute
+ metadataColumnsByName.put(field.name, attr)
+ generatedMetadataColumns += attr
+
+ case FileSourceConstantMetadataStructField(field) =>
+ val attr = field.toAttribute
+ metadataColumnsByName.put(field.name, attr)
+ constantMetadataColumns += attr
+
+ case field => throw new AnalysisException(s"Unrecognized file metadata field: $field")
}
- }
- val metadataColumns: Seq[Attribute] =
- constantMetadataColumns.toSeq ++ generatedMetadataColumns.toSeq
+ metadataStruct.dataType.asInstanceOf[StructType].fields.foreach(createMetadataColumn)
Review Comment:
We know that `schema` is a `StructType` here because we matched on it in `MetadataStructColumn` above but the type checker thinks this can be any `DataType` so we have a partial match. The cast looks like the most readable option
--
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
[GitHub] [spark] johanl-db commented on a diff in pull request #40545: [SPARK-42918] Generalize handling of metadata attributes in FileSourceStrategy
Posted by "johanl-db (via GitHub)" <gi...@apache.org>.
johanl-db commented on code in PR #40545:
URL: https://github.com/apache/spark/pull/40545#discussion_r1154397094
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -220,9 +220,20 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
attributeReference match {
case attr @ FileSourceMetadataAttribute(
Review Comment:
We only cleanup the individual metadata fields but the _metadata attribute itself does have the right information to match on `FileSourceMetadataAttribute` :
```
def createFileMetadataCol(fileFormat: FileFormat): AttributeReference = {
// Strip out the fields' metadata to avoid exposing it to the user. [[FileSourceStrategy]]
// avoids confusion by mapping back to [[metadataSchemaFields]].
val fields = metadataSchemaFields(fileFormat)
.map(FileSourceMetadataAttribute.cleanupFileSourceMetadataInformation)
FileSourceMetadataAttribute(FileFormat.METADATA_NAME, StructType(fields))
}
```
The following is true:
```
val metadata_attr = createFileMetadataCol(fileFormat: FileFormat)
metadata_attr.metadata.getBoolean(METADATA_COL_ATTR_KEY) == true
metadata_attr.metadata.getBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY) == true
metadata_attr.dataType.forall(_.metadata == Metadata.empty)
--
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