You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2023/01/05 11:50:40 UTC
[GitHub] [spark] olaky opened a new pull request, #39408: [SPARK-41896] Filtering by row index returns empty results
olaky opened a new pull request, #39408:
URL: https://github.com/apache/spark/pull/39408
### What changes were proposed in this pull request?
When filtering by row index, the right number of rows must be returned. At the moment, all files are filtered out.
### Why are the changes needed?
Return correct results when filtering by row index
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added new tests
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1065846695
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,26 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. The rebinding will
+ // categorize the references as either [[FileSourceConstantMetadataAttribute]] or
+ // [[FileSourceGeneratedMetadataAttribute]].
+ def rebindFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case fieldGetter @ GetStructField(FileSourceMetadataAttribute(_), _, _) =>
Review Comment:
what if the predicate is `func_taking_struct(_metadata)`? shall we replace `_metadata` with `CreateStruct(metadataColumns)`?
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063498040
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,24 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+
+ // The metadata attribute references in the filters also have to be categorized as either
+ // constant or generated metadata attributes. Only data filters can contain metadata filters.
+ def categorizeFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case attr: AttributeReference if FileSourceMetadataAttribute.unapply(attr).isDefined =>
+ if (attr.dataType.asInstanceOf[StructType].fieldNames
+ .forall(fieldName => constantMetadataColumns.exists(fieldName == _.name))) {
Review Comment:
That is a really good idea, I did settle for something along those lines: I am rebinding the references to the flattened columns. This way the filters do again reference columns that are actually part of the schema
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069358007
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -214,9 +214,20 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
.filterNot(partitionColumns.contains)
// Metadata attributes are part of a column of type struct up to this point. Here we extract
- // this column from the schema.
+ // this column from the schema and specify a matcher for that.
+ object MetadataStructColumn {
+ def unapply(attributeReference: AttributeReference): Option[AttributeReference] = {
Review Comment:
Unfortunately we require the exprId towards the end of the function
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1066672989
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala:
##########
@@ -76,28 +76,35 @@ abstract class PartitioningAwareFileIndex(
// be applied to files.
val fileMetadataFilterOpt = dataFilters.filter { f =>
f.references.nonEmpty && f.references.forall {
- case FileSourceMetadataAttribute(_) => true
+ case FileSourceConstantMetadataAttribute(_) => true
case _ => false
}
}.reduceOption(expressions.And)
- // - create a bound references for filters: put the metadata struct at 0 position for each file
- // - retrieve the final metadata struct (could be pruned) from filters
+ // - Retrieve all required metadata attributes and put them into a sequence
+ // - Bind all file constant metadata attribute references to their respective index
+ val requiredMetadataColumnNames: mutable.Buffer[String] = mutable.Buffer.empty
val boundedFilterMetadataStructOpt = fileMetadataFilterOpt.map { fileMetadataFilter =>
- val metadataStruct = fileMetadataFilter.references.head.dataType
- val boundedFilter = Predicate.createInterpreted(fileMetadataFilter.transform {
- case _: AttributeReference => BoundReference(0, metadataStruct, nullable = true)
+ val metadataStruct = fileMetadataFilter.references.head
Review Comment:
Outch, this was indeed a wrong reference. The type from the attribute reference should be used
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1066677496
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,26 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. The rebinding will
+ // categorize the references as either [[FileSourceConstantMetadataAttribute]] or
+ // [[FileSourceGeneratedMetadataAttribute]].
+ def rebindFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case fieldGetter @ GetStructField(FileSourceMetadataAttribute(_), _, _) =>
Review Comment:
Good point, I will have to deep dive a bit here. Basically I am not sure if this is actually possible or if a filter referencing metadata in the struct form will error out because the struct is not there yet once the initial filter expression is built
--
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] dongjoon-hyun commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063025718
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,60 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
+
+ test(s"Filter on row_index and a stored column at the same time") {
Review Comment:
`s"` -> `"`.
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1065844409
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,26 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. The rebinding will
+ // categorize the references as either [[FileSourceConstantMetadataAttribute]] or
+ // [[FileSourceGeneratedMetadataAttribute]].
+ def rebindFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case fieldGetter @ GetStructField(FileSourceMetadataAttribute(_), _, _) =>
+ val fieldName = fieldGetter.extractFieldName
+ // The row_index column has been renamed.
+ val columnName = if (fieldName == FileFormat.ROW_INDEX) {
+ FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
+ } else {
+ fieldName
+ }
+ metadataColumns.find(_.name == columnName).get
Review Comment:
shall we consider case sensitivity?
--
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] dongjoon-hyun commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063025838
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,60 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
+
+ test(s"Filter on row_index and a stored column at the same time") {
Review Comment:
Also, please use `SPARK-41896: ` as the test name prefix.
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069080192
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala:
##########
@@ -76,28 +76,34 @@ abstract class PartitioningAwareFileIndex(
// be applied to files.
val fileMetadataFilterOpt = dataFilters.filter { f =>
f.references.nonEmpty && f.references.forall {
- case FileSourceMetadataAttribute(_) => true
+ case FileSourceConstantMetadataAttribute(_) => true
case _ => false
}
}.reduceOption(expressions.And)
- // - create a bound references for filters: put the metadata struct at 0 position for each file
- // - retrieve the final metadata struct (could be pruned) from filters
+ // - Retrieve all required metadata attributes and put them into a sequence
+ // - Bind all file constant metadata attribute references to their respective index
+ val requiredMetadataColumnNames: mutable.Buffer[String] = mutable.Buffer.empty
Review Comment:
nvm, this is not for column pruning, but just make it easy to bind the index.
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069349167
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -214,9 +214,20 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
.filterNot(partitionColumns.contains)
// Metadata attributes are part of a column of type struct up to this point. Here we extract
- // this column from the schema.
+ // this column from the schema and specify a matcher for that.
+ object MetadataStructColumn {
+ def unapply(attributeReference: AttributeReference): Option[AttributeReference] = {
Review Comment:
how about returning `Option[StructType]`? I'm looking at `metadataStructOpt`, and seems we only need its struct type. The same to rebinding predicates
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069077127
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala:
##########
@@ -76,28 +76,34 @@ abstract class PartitioningAwareFileIndex(
// be applied to files.
val fileMetadataFilterOpt = dataFilters.filter { f =>
f.references.nonEmpty && f.references.forall {
- case FileSourceMetadataAttribute(_) => true
+ case FileSourceConstantMetadataAttribute(_) => true
Review Comment:
After this change, now we exclude any unflattened metadata predicates. This is probably why the new test passes
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1062496723
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,24 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+
+ // The metadata attribute references in the filters also have to be categorized as either
+ // constant or generated metadata attributes. Only data filters can contain metadata filters.
+ def categorizeFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case attr: AttributeReference if FileSourceMetadataAttribute.unapply(attr).isDefined =>
Review Comment:
one more generic way to do this I could imagine is to unbind the references and have that process run again for all filters, but that seems rather wasteful
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063129844
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,24 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+
+ // The metadata attribute references in the filters also have to be categorized as either
+ // constant or generated metadata attributes. Only data filters can contain metadata filters.
+ def categorizeFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case attr: AttributeReference if FileSourceMetadataAttribute.unapply(attr).isDefined =>
+ if (attr.dataType.asInstanceOf[StructType].fieldNames
+ .forall(fieldName => constantMetadataColumns.exists(fieldName == _.name))) {
Review Comment:
Seems we flatten the metadata column in the scan node, shall we flatten the metadata cols in the filter as well?
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1066667954
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,26 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. The rebinding will
+ // categorize the references as either [[FileSourceConstantMetadataAttribute]] or
+ // [[FileSourceGeneratedMetadataAttribute]].
+ def rebindFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case fieldGetter @ GetStructField(FileSourceMetadataAttribute(_), _, _) =>
+ val fieldName = fieldGetter.extractFieldName
+ // The row_index column has been renamed.
+ val columnName = if (fieldName == FileFormat.ROW_INDEX) {
+ FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
+ } else {
+ fieldName
+ }
+ metadataColumns.find(_.name == columnName).get
Review Comment:
The flattening does not change the name, so I guess not doing that is a bit more safe on making this a 1 to 1 comparison? On the other hand, duplicate names because of different casing are not allowed AFAIK
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1066843508
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,26 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. The rebinding will
+ // categorize the references as either [[FileSourceConstantMetadataAttribute]] or
+ // [[FileSourceGeneratedMetadataAttribute]].
+ def rebindFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case fieldGetter @ GetStructField(FileSourceMetadataAttribute(_), _, _) =>
Review Comment:
I added a test for this and it works
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069318980
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,36 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. After the rebinding
+ // 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
+ }
+
+ 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 FileSourceMetadataAttribute(
+ AttributeReference("_metadata", fields @ StructType(_), _, _)) =>
Review Comment:
This reminds me of a similar code block in this file
```
// Metadata attributes are part of a column of type struct up to this point. Here we extract
// this column from the schema.
val metadataStructOpt = l.output.collectFirst {
case FileSourceMetadataAttribute(attr) => attr
}
```
Shall we unify them?
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069076438
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,90 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
+ test("SPARK-41896: Filter on row_index and a stored column at the same time") {
+ withTempPath { dir =>
+ val storedIdName = "stored_id"
+ val storedIdUpperLimitExclusive = 520
+ val rowIndexLowerLimitInclusive = 10
+
+ spark.range(start = 500, end = 600)
+ .toDF(storedIdName)
+ .write
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Select stored_id 510 to 519 via a stored_id and row_index filter.
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .select(storedIdName, METADATA_ROW_INDEX)
+ .where(col(storedIdName).lt(lit(storedIdUpperLimitExclusive)))
+ .where(col(METADATA_ROW_INDEX).geq(lit(rowIndexLowerLimitInclusive)))
+ .collect()
+
+ assert(collectedRows.length === 10)
+ assert(collectedRows.forall(_.getLong(0) < storedIdUpperLimitExclusive))
+ assert(collectedRows.forall(_.getLong(1) >= rowIndexLowerLimitInclusive))
+ }
+ }
+
+ test("SPARK-41896: Filter on constant and generated metadata attributes at the same time") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val partitionColumnName = "partition"
+ val numFiles = 4
+ val totalNumRows = 40
+
+ spark.range(end = totalNumRows)
+ .toDF(idColumnName)
+ .withColumn(partitionColumnName, col(idColumnName).mod(lit(numFiles)))
+ .write
+ .partitionBy(partitionColumnName)
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Get one file path.
+ val randomTableFilePath = spark.read.load(dir.getAbsolutePath)
+ .select(METADATA_FILE_PATH).collect().head.getString(0)
+
+ val halfTheNumberOfRowsPerFile = totalNumRows / (numFiles * 2)
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .where(col(METADATA_FILE_PATH).equalTo(lit(randomTableFilePath)))
+ .where(col(METADATA_ROW_INDEX).leq(lit(halfTheNumberOfRowsPerFile)))
Review Comment:
is there a way to test that the metadata predicate is indeed pushed and evaluated?
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069078184
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala:
##########
@@ -76,28 +76,34 @@ abstract class PartitioningAwareFileIndex(
// be applied to files.
val fileMetadataFilterOpt = dataFilters.filter { f =>
f.references.nonEmpty && f.references.forall {
- case FileSourceMetadataAttribute(_) => true
+ case FileSourceConstantMetadataAttribute(_) => true
case _ => false
}
}.reduceOption(expressions.And)
- // - create a bound references for filters: put the metadata struct at 0 position for each file
- // - retrieve the final metadata struct (could be pruned) from filters
+ // - Retrieve all required metadata attributes and put them into a sequence
+ // - Bind all file constant metadata attribute references to their respective index
+ val requiredMetadataColumnNames: mutable.Buffer[String] = mutable.Buffer.empty
Review Comment:
is this really helpful? All the values are already in memory. We just need to build a row and evaluate the filter. Doing column pruning doesn't seem to help for this case.
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on PR #39408:
URL: https://github.com/apache/spark/pull/39408#issuecomment-1382658324
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069282419
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,90 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
+ test("SPARK-41896: Filter on row_index and a stored column at the same time") {
+ withTempPath { dir =>
+ val storedIdName = "stored_id"
+ val storedIdUpperLimitExclusive = 520
+ val rowIndexLowerLimitInclusive = 10
+
+ spark.range(start = 500, end = 600)
+ .toDF(storedIdName)
+ .write
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Select stored_id 510 to 519 via a stored_id and row_index filter.
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .select(storedIdName, METADATA_ROW_INDEX)
+ .where(col(storedIdName).lt(lit(storedIdUpperLimitExclusive)))
+ .where(col(METADATA_ROW_INDEX).geq(lit(rowIndexLowerLimitInclusive)))
+ .collect()
+
+ assert(collectedRows.length === 10)
+ assert(collectedRows.forall(_.getLong(0) < storedIdUpperLimitExclusive))
+ assert(collectedRows.forall(_.getLong(1) >= rowIndexLowerLimitInclusive))
+ }
+ }
+
+ test("SPARK-41896: Filter on constant and generated metadata attributes at the same time") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val partitionColumnName = "partition"
+ val numFiles = 4
+ val totalNumRows = 40
+
+ spark.range(end = totalNumRows)
+ .toDF(idColumnName)
+ .withColumn(partitionColumnName, col(idColumnName).mod(lit(numFiles)))
+ .write
+ .partitionBy(partitionColumnName)
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Get one file path.
+ val randomTableFilePath = spark.read.load(dir.getAbsolutePath)
+ .select(METADATA_FILE_PATH).collect().head.getString(0)
+
+ val halfTheNumberOfRowsPerFile = totalNumRows / (numFiles * 2)
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .where(col(METADATA_FILE_PATH).equalTo(lit(randomTableFilePath)))
+ .where(col(METADATA_ROW_INDEX).leq(lit(halfTheNumberOfRowsPerFile)))
+ .collect()
+
+ // The query will match half the rows in one of the files.
+ assert(collectedRows.length === halfTheNumberOfRowsPerFile)
+ }
+ }
+
+ test("SPARK-41896: Filter by a function that takes the metadata struct as argument") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val numFiles = 4
+ spark.range(end = numFiles)
+ .toDF(idColumnName)
+ .withColumn("partition", col(idColumnName))
+ .write
+ .format("parquet")
+ .partitionBy("partition")
+ .save(dir.getAbsolutePath)
+
+ // Select path and partition value for a random file.
+ val testFileData = spark.read.load(dir.getAbsolutePath)
+ .select(idColumnName, METADATA_FILE_PATH).collect().head
+ val testFilePartition = testFileData.getLong(0)
+ val testFilePath = testFileData.getString(1)
+
+ // Create and use a filter using the file path.
+ spark.udf.register("isTestFile",
+ (metadata: Row) => { metadata.getAs[String]("file_path") == testFilePath })
+ val udfFilterResult = spark.read.load(dir.getAbsolutePath)
+ .select(idColumnName, METADATA_FILE_PATH)
+ .where("isTestFile(_metadata)")
Review Comment:
I did assume that this works because at some point in the evaluation the plan has to contain GetStructField, and where I transform the plan this is already the case.
I am changing the implementation and then we will not depend on this any more
--
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] olaky commented on pull request #39408: [SPARK-41896] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on PR #39408:
URL: https://github.com/apache/spark/pull/39408#issuecomment-1372236394
cc @cloud-fan
--
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] olaky commented on pull request #39408: [SPARK-41896] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on PR #39408:
URL: https://github.com/apache/spark/pull/39408#issuecomment-1372206581
> Is there anything missing? looks like only test code
yes sorry, so far this only the tests for visibility. I will be adding a fix as well
--
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] AmplabJenkins commented on pull request #39408: [SPARK-41896] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on PR #39408:
URL: https://github.com/apache/spark/pull/39408#issuecomment-1372298824
Can one of the admins verify this patch?
--
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] dongjoon-hyun commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063025466
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,24 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+
Review Comment:
Redundant empty line.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,60 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
Review Comment:
Redundant empty line.
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069072114
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,90 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
+ test("SPARK-41896: Filter on row_index and a stored column at the same time") {
+ withTempPath { dir =>
+ val storedIdName = "stored_id"
+ val storedIdUpperLimitExclusive = 520
+ val rowIndexLowerLimitInclusive = 10
+
+ spark.range(start = 500, end = 600)
+ .toDF(storedIdName)
+ .write
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Select stored_id 510 to 519 via a stored_id and row_index filter.
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .select(storedIdName, METADATA_ROW_INDEX)
+ .where(col(storedIdName).lt(lit(storedIdUpperLimitExclusive)))
+ .where(col(METADATA_ROW_INDEX).geq(lit(rowIndexLowerLimitInclusive)))
+ .collect()
+
+ assert(collectedRows.length === 10)
+ assert(collectedRows.forall(_.getLong(0) < storedIdUpperLimitExclusive))
+ assert(collectedRows.forall(_.getLong(1) >= rowIndexLowerLimitInclusive))
+ }
+ }
+
+ test("SPARK-41896: Filter on constant and generated metadata attributes at the same time") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val partitionColumnName = "partition"
+ val numFiles = 4
+ val totalNumRows = 40
+
+ spark.range(end = totalNumRows)
+ .toDF(idColumnName)
+ .withColumn(partitionColumnName, col(idColumnName).mod(lit(numFiles)))
+ .write
+ .partitionBy(partitionColumnName)
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Get one file path.
+ val randomTableFilePath = spark.read.load(dir.getAbsolutePath)
+ .select(METADATA_FILE_PATH).collect().head.getString(0)
+
+ val halfTheNumberOfRowsPerFile = totalNumRows / (numFiles * 2)
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .where(col(METADATA_FILE_PATH).equalTo(lit(randomTableFilePath)))
+ .where(col(METADATA_ROW_INDEX).leq(lit(halfTheNumberOfRowsPerFile)))
+ .collect()
+
+ // The query will match half the rows in one of the files.
+ assert(collectedRows.length === halfTheNumberOfRowsPerFile)
+ }
+ }
+
+ test("SPARK-41896: Filter by a function that takes the metadata struct as argument") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val numFiles = 4
+ spark.range(end = numFiles)
+ .toDF(idColumnName)
+ .withColumn("partition", col(idColumnName))
+ .write
+ .format("parquet")
+ .partitionBy("partition")
+ .save(dir.getAbsolutePath)
+
+ // Select path and partition value for a random file.
+ val testFileData = spark.read.load(dir.getAbsolutePath)
+ .select(idColumnName, METADATA_FILE_PATH).collect().head
+ val testFilePartition = testFileData.getLong(0)
+ val testFilePath = testFileData.getString(1)
+
+ // Create and use a filter using the file path.
+ spark.udf.register("isTestFile",
+ (metadata: Row) => { metadata.getAs[String]("file_path") == testFilePath })
+ val udfFilterResult = spark.read.load(dir.getAbsolutePath)
+ .select(idColumnName, METADATA_FILE_PATH)
+ .where("isTestFile(_metadata)")
Review Comment:
do you know why this works? We do not flatten metadata column reference without `GetStructField`
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069071458
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,90 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
+ test("SPARK-41896: Filter on row_index and a stored column at the same time") {
+ withTempPath { dir =>
+ val storedIdName = "stored_id"
+ val storedIdUpperLimitExclusive = 520
+ val rowIndexLowerLimitInclusive = 10
+
+ spark.range(start = 500, end = 600)
+ .toDF(storedIdName)
+ .write
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Select stored_id 510 to 519 via a stored_id and row_index filter.
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .select(storedIdName, METADATA_ROW_INDEX)
+ .where(col(storedIdName).lt(lit(storedIdUpperLimitExclusive)))
+ .where(col(METADATA_ROW_INDEX).geq(lit(rowIndexLowerLimitInclusive)))
+ .collect()
+
+ assert(collectedRows.length === 10)
+ assert(collectedRows.forall(_.getLong(0) < storedIdUpperLimitExclusive))
+ assert(collectedRows.forall(_.getLong(1) >= rowIndexLowerLimitInclusive))
+ }
+ }
+
+ test("SPARK-41896: Filter on constant and generated metadata attributes at the same time") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val partitionColumnName = "partition"
+ val numFiles = 4
+ val totalNumRows = 40
+
+ spark.range(end = totalNumRows)
+ .toDF(idColumnName)
+ .withColumn(partitionColumnName, col(idColumnName).mod(lit(numFiles)))
+ .write
+ .partitionBy(partitionColumnName)
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Get one file path.
+ val randomTableFilePath = spark.read.load(dir.getAbsolutePath)
+ .select(METADATA_FILE_PATH).collect().head.getString(0)
+
+ val halfTheNumberOfRowsPerFile = totalNumRows / (numFiles * 2)
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .where(col(METADATA_FILE_PATH).equalTo(lit(randomTableFilePath)))
+ .where(col(METADATA_ROW_INDEX).leq(lit(halfTheNumberOfRowsPerFile)))
+ .collect()
+
+ // The query will match half the rows in one of the files.
+ assert(collectedRows.length === halfTheNumberOfRowsPerFile)
+ }
+ }
+
+ test("SPARK-41896: Filter by a function that takes the metadata struct as argument") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val numFiles = 4
+ spark.range(end = numFiles)
+ .toDF(idColumnName)
+ .withColumn("partition", col(idColumnName))
+ .write
+ .format("parquet")
+ .partitionBy("partition")
+ .save(dir.getAbsolutePath)
+
+ // Select path and partition value for a random file.
+ val testFileData = spark.read.load(dir.getAbsolutePath)
+ .select(idColumnName, METADATA_FILE_PATH).collect().head
+ val testFilePartition = testFileData.getLong(0)
+ val testFilePath = testFileData.getString(1)
+
+ // Create and use a filter using the file path.
+ spark.udf.register("isTestFile",
Review Comment:
wrap it with `withUserDefinedFunction` to clean up the function at the end
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
URL: https://github.com/apache/spark/pull/39408
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1065842280
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,26 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. The rebinding will
+ // categorize the references as either [[FileSourceConstantMetadataAttribute]] or
+ // [[FileSourceGeneratedMetadataAttribute]].
Review Comment:
seems the comment is stale. We don't categorize the attribute 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] olaky commented on a diff in pull request #39408: [SPARK-41896] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1062515380
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala:
##########
@@ -76,7 +76,7 @@ abstract class PartitioningAwareFileIndex(
// be applied to files.
val fileMetadataFilterOpt = dataFilters.filter { f =>
f.references.nonEmpty && f.references.forall {
- case FileSourceMetadataAttribute(_) => true
+ case FileSourceConstantMetadataAttribute(_) => true
Review Comment:
It would also be possible to exclude row index by name here, but that seems to defeat the purpose of categorising the metadata columns
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063497223
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,24 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+
+ // The metadata attribute references in the filters also have to be categorized as either
+ // constant or generated metadata attributes. Only data filters can contain metadata filters.
+ def categorizeFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case attr: AttributeReference if FileSourceMetadataAttribute.unapply(attr).isDefined =>
+ if (attr.dataType.asInstanceOf[StructType].fieldNames
+ .forall(fieldName => constantMetadataColumns.exists(fieldName == _.name))) {
Review Comment:
I settled for a different implementation
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1066664596
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,26 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+ // Rebind metadata attribute references in filters after the metadata attribute struct has
+ // been flattened. Only data filters can contain metadata references. The rebinding will
+ // categorize the references as either [[FileSourceConstantMetadataAttribute]] or
+ // [[FileSourceGeneratedMetadataAttribute]].
Review Comment:
We actually do as a side effect (the output attributes are all categorized). I clarified that in the comment
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1065845336
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala:
##########
@@ -76,28 +76,35 @@ abstract class PartitioningAwareFileIndex(
// be applied to files.
val fileMetadataFilterOpt = dataFilters.filter { f =>
f.references.nonEmpty && f.references.forall {
- case FileSourceMetadataAttribute(_) => true
+ case FileSourceConstantMetadataAttribute(_) => true
case _ => false
}
}.reduceOption(expressions.And)
- // - create a bound references for filters: put the metadata struct at 0 position for each file
- // - retrieve the final metadata struct (could be pruned) from filters
+ // - Retrieve all required metadata attributes and put them into a sequence
+ // - Bind all file constant metadata attribute references to their respective index
+ val requiredMetadataColumnNames: mutable.Buffer[String] = mutable.Buffer.empty
val boundedFilterMetadataStructOpt = fileMetadataFilterOpt.map { fileMetadataFilter =>
- val metadataStruct = fileMetadataFilter.references.head.dataType
- val boundedFilter = Predicate.createInterpreted(fileMetadataFilter.transform {
- case _: AttributeReference => BoundReference(0, metadataStruct, nullable = true)
+ val metadataStruct = fileMetadataFilter.references.head
Review Comment:
hmm, I thought the attrs are already flattened and there is no struct anymore
--
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 #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063128796
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,24 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+
+ // The metadata attribute references in the filters also have to be categorized as either
+ // constant or generated metadata attributes. Only data filters can contain metadata filters.
+ def categorizeFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case attr: AttributeReference if FileSourceMetadataAttribute.unapply(attr).isDefined =>
Review Comment:
We can just write `case FileSourceMetadataAttribute(attr) =>`
--
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] LuciferYang commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1063345512
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -258,6 +258,24 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
val outputAttributes = readDataColumns ++ generatedMetadataColumns ++
partitionColumns ++ constantMetadataColumns
+
+ // The metadata attribute references in the filters also have to be categorized as either
+ // constant or generated metadata attributes. Only data filters can contain metadata filters.
+ def categorizeFileSourceMetadataAttributesInFilters(
+ filters: Seq[Expression]): Seq[Expression] =
+ filters.map { filter =>
+ filter.transform {
+ case attr: AttributeReference if FileSourceMetadataAttribute.unapply(attr).isDefined =>
+ if (attr.dataType.asInstanceOf[StructType].fieldNames
+ .forall(fieldName => constantMetadataColumns.exists(fieldName == _.name))) {
Review Comment:
a complex conditional expression, extracting a function may look clearer
--
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] LuciferYang commented on pull request #39408: [SPARK-41896] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on PR #39408:
URL: https://github.com/apache/spark/pull/39408#issuecomment-1372158935
Is there anything missing? looks like only test code
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069313774
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructSuite.scala:
##########
@@ -677,4 +677,90 @@ class FileMetadataStructSuite extends QueryTest with SharedSparkSession {
assert(analyzedStruct.fields.forall(!_.nullable), analyzedStruct.fields.mkString(", "))
assert(executedStruct.fields.forall(!_.nullable), executedStruct.fields.mkString(", "))
}
+
+ test("SPARK-41896: Filter on row_index and a stored column at the same time") {
+ withTempPath { dir =>
+ val storedIdName = "stored_id"
+ val storedIdUpperLimitExclusive = 520
+ val rowIndexLowerLimitInclusive = 10
+
+ spark.range(start = 500, end = 600)
+ .toDF(storedIdName)
+ .write
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Select stored_id 510 to 519 via a stored_id and row_index filter.
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .select(storedIdName, METADATA_ROW_INDEX)
+ .where(col(storedIdName).lt(lit(storedIdUpperLimitExclusive)))
+ .where(col(METADATA_ROW_INDEX).geq(lit(rowIndexLowerLimitInclusive)))
+ .collect()
+
+ assert(collectedRows.length === 10)
+ assert(collectedRows.forall(_.getLong(0) < storedIdUpperLimitExclusive))
+ assert(collectedRows.forall(_.getLong(1) >= rowIndexLowerLimitInclusive))
+ }
+ }
+
+ test("SPARK-41896: Filter on constant and generated metadata attributes at the same time") {
+ withTempPath { dir =>
+ val idColumnName = "id"
+ val partitionColumnName = "partition"
+ val numFiles = 4
+ val totalNumRows = 40
+
+ spark.range(end = totalNumRows)
+ .toDF(idColumnName)
+ .withColumn(partitionColumnName, col(idColumnName).mod(lit(numFiles)))
+ .write
+ .partitionBy(partitionColumnName)
+ .format("parquet")
+ .save(dir.getAbsolutePath)
+
+ // Get one file path.
+ val randomTableFilePath = spark.read.load(dir.getAbsolutePath)
+ .select(METADATA_FILE_PATH).collect().head.getString(0)
+
+ val halfTheNumberOfRowsPerFile = totalNumRows / (numFiles * 2)
+ val collectedRows = spark.read.load(dir.getAbsolutePath)
+ .where(col(METADATA_FILE_PATH).equalTo(lit(randomTableFilePath)))
+ .where(col(METADATA_ROW_INDEX).leq(lit(halfTheNumberOfRowsPerFile)))
Review Comment:
I added more asserts here: We select half the columns from one of the files. What I assert now is that all rows do indeed come from one file and have a row index lower than halfTheNumberOfRowsPerFile. That proves that the filter is evaluated if I am not mistaken
--
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] olaky commented on a diff in pull request #39408: [SPARK-41896][SQL] Filtering by row index returns empty results
Posted by GitBox <gi...@apache.org>.
olaky commented on code in PR #39408:
URL: https://github.com/apache/spark/pull/39408#discussion_r1069311234
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala:
##########
@@ -76,28 +76,34 @@ abstract class PartitioningAwareFileIndex(
// be applied to files.
val fileMetadataFilterOpt = dataFilters.filter { f =>
f.references.nonEmpty && f.references.forall {
- case FileSourceMetadataAttribute(_) => true
+ case FileSourceConstantMetadataAttribute(_) => true
Review Comment:
Here we do expect to only see FileSourceContant/GeneratedMetadata attributes. It is hard to test this though in an automated fashion, and I think actually adding an assert here is not the best idea (not filtering is a better option than throwing an exception).
I manually verified this for now
--
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