comphead commented on code in PR #7196:
URL: https://github.com/apache/datafusion-comet/pull/7196#discussion_r4379730719
##########
native/core/src/parquet/field_resolver.rs:
##########
@@ -47,14 +47,87 @@ pub(crate) enum Selection<'a> {
ByName(&'a [usize]),
}
+/// The strongest claim on one file field. Root remapping and nested
conversion give a file
+/// field to this claimant only. Repeated id claims keep the last requested
field, matching
+/// Spark parquet-mr's `ParquetRowConverter` id-to-index `toMap`. Equal name
claims keep the
+/// earliest requested field, retaining the root remap's name-matching
behavior.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum FieldClaim {
+ /// The requested field selects this file field by Parquet field id.
+ IdMatch(usize),
+ /// An id-bearing requested field has this file field's folded name. It
must not reach this
+ /// field by name when its id is missing or selects a different field.
Only an id match can
+ /// override this claim. The root hides it with a placeholder; nested
fields read NULL.
+ Shadowed,
+ /// The requested field selects this file field by folded name.
+ NameMatch(usize),
+ /// No requested field claims this file field. The root keeps its original
name.
+ Unchanged,
+}
+
+impl FieldClaim {
+ fn owned_by(self, requested: usize) -> bool {
+ matches!(self, Self::IdMatch(owner) | Self::NameMatch(owner) if owner
== requested)
+ }
+
+ fn strength(self) -> u8 {
+ match self {
+ Self::IdMatch(_) => 3,
+ Self::Shadowed => 2,
+ Self::NameMatch(_) => 1,
+ Self::Unchanged => 0,
+ }
+ }
+
+ fn claim(&mut self, claim: Self) {
+ if claim.strength() > self.strength()
+ || matches!((*self, claim), (Self::IdMatch(_), Self::IdMatch(_)))
Review Comment:
With the new required-root fallback, I could not find a read that gives one
file field two id claims, two name claims, or a shadow plus a name claim. A
repeated required id falls back in `CometScanTypeChecker` (top-level and
nested), a required root sharing an id with an unread root falls back in the
new `hasDuplicateFieldIds(fields, requiredFieldNames)`, and siblings whose
names fold together fall back through `hasCaseInsensitiveDuplicateFieldNames`
and `hasDuplicateFieldNames`. A tie between two unread roots changes no
projected column. That leaves id against name, the #6994 shape. If that holds,
this last-wins arm only runs in unit tests, and main's tie order could stay.
Then `repeated_requested_ids_read_last_field_at_root_and_nested`,
`id_ties_keep_last_and_name_ties_keep_first`,
`three_or_more_id_claims_keep_last_over_name_and_shadow` and
`shadowed_struct_fields_follow_all_missing_null_policy`, about 200 lines, could
go, or shrink to one test if parquet-mr parity is the reaso
n to keep it. I traced the guards and have not run them.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -138,6 +139,11 @@ fn is_pure_structural_narrowing(
// Spark and Comet's converter reject it, but DataFusion's
cast would silently
// return the exact-case field, so the cast must not be
retained.
let folded_target = fold_name(target_field.name(),
parquet_options.case_sensitive)?;
+ // The generic cast can give one source to several targets.
Comet keeps only
+ // the earliest equal-strength name claim, so that shape needs
its converter.
+ if !claimed_names.insert(folded_target.clone()) {
Review Comment:
I think this check cannot fire. An id-bearing target already returns `false`
at the `use_field_id` check above. Two targets that fold together but differ in
case leave one without an exact-name source, which returns `false` at the
`find(|f| f.name() == target_field.name())` below. What remains is
byte-identical sibling names, which `DataTypeSupport.isTypeSupported` declines
(`struct with duplicate field names`).
`is_pure_structural_narrowing_denies_conflicting_name_claims` builds exactly
that `("a", "a")` shape. Could the check, its per-target `String` clone and
that test go?
##########
spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala:
##########
@@ -1269,6 +1270,517 @@ class CometNativeReaderSuite extends CometTestBase with
AdaptiveSparkPlanHelper
nullable = true,
new MetadataBuilder().putLong("parquet.field.id", id.toLong).build())
+ private def withFieldClaimFile(nested: Boolean)(f: (String, StructType) =>
Unit): Unit = {
+ val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build()
+ def wrapSchema(fields: StructType): StructType =
+ if (nested) StructType(Seq(StructField("s", fields, nullable = true,
id10)))
+ else fields
+ val writeSchema = wrapSchema(StructType(Seq(withFieldId("A", 1))))
+ val readSchema = wrapSchema(
+ StructType(Seq(withFieldId("x", 1), StructField("a", LongType, nullable
= true))))
+ withSQLConf(
+ SQLConf.CASE_SENSITIVE.key -> "false",
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ val rows = (1L to 3L).map(value => if (nested) Row(Row(value)) else
Row(value))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(rows, 1),
writeSchema)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ f(path.getCanonicalPath, readSchema)
+ }
+ }
+ }
+
+ private def fieldClaimReference(query: => DataFrame): Seq[Row] = {
+ var rows = Seq.empty[Row]
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false") {
+ val df = query
+ val scans = collect(df.queryExecution.executedPlan) { case scan:
FileSourceScanExec =>
+ scan
+ }
+ assert(scans.nonEmpty && scans.forall(!_.supportsColumnar))
+ rows = df.collect().toSeq
+ }
+ rows
+ }
+
+ private def checkNativeFieldClaimAnswer(df: DataFrame, expected: Seq[Row]):
Unit = {
+ assert(
+ collect(df.queryExecution.executedPlan) { case scan: CometNativeScanExec
=> scan }.nonEmpty,
+ s"expected native field claim
resolution:\n${df.queryExecution.executedPlan}")
+ checkAnswer(df, expected)
+ }
+
+ for {
+ nested <- Seq(false, true)
+ pruning <- (if (nested) Seq(false, true) else Seq(false))
+ selected <- Seq("both", "id", "name")
+ } {
+ test(
+ s"Parquet field claims projection - nested=$nested pruning=$pruning
selected=$selected") {
+ withFieldClaimFile(nested) { (path, schema) =>
+ withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key ->
pruning.toString) {
+ def query: DataFrame = {
+ val df = spark.read.schema(schema).parquet(path)
+ val prefix = if (nested) "s." else ""
+ val names = selected match {
+ case "both" => Seq("x", "a")
+ case "id" => Seq("x")
+ case "name" => Seq("a")
+ }
+ df.select(names.map(name => col(prefix + name)): _*)
+ }
+ val expected = fieldClaimReference(query)
+ checkNativeFieldClaimAnswer(query, expected)
+ }
+ }
+ }
+ }
+
+ for {
+ nested <- Seq(false, true)
+ pushdown <- Seq(false, true)
+ predicate <- Seq("missing is null", "missing is not null", "id greater
than one")
+ } {
+ test(
+ s"Parquet field claims filters - nested=$nested pushdown=$pushdown
predicate=$predicate") {
+ withFieldClaimFile(nested) { (path, schema) =>
+ withSQLConf(
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key ->
pushdown.toString) {
+ def query: DataFrame = {
+ val df = spark.read.schema(schema).parquet(path)
+ val prefix = if (nested) "s." else ""
+ val x = col(prefix + "x")
+ val a = col(prefix + "a")
+ val condition = predicate match {
+ case "missing is null" => a.isNull
+ case "missing is not null" => a.isNotNull
+ case "id greater than one" => x > 1L
+ }
+ df.where(condition).select(x, a)
+ }
+ val expected = fieldClaimReference(query)
+ val values = predicate match {
+ case "missing is null" => Seq(1L, 2L, 3L)
+ case "missing is not null" => Seq.empty[Long]
+ case "id greater than one" => Seq(2L, 3L)
+ }
+ assert(expected.sortBy(_.toString) == values.map(value => Row(value,
null)))
+ val df = query
+ checkNativeFieldClaimAnswer(df, expected)
+ if (!nested && pushdown && predicate == "id greater than one") {
+ val scans = collect(df.queryExecution.executedPlan) {
+ case scan: CometNativeScanExec =>
+ scan
+ }
+ assert(
+ scans.exists(scan =>
+ scan.metrics("pushdown_rows_pruned").value +
+ scan.metrics("pushdown_rows_matched").value > 0))
+ }
+ }
+ }
+ }
+ }
+
+ for {
+ nested <- Seq(false, true)
+ reverse <- Seq(false, true)
+ kind <- Seq("struct", "array", "map")
+ } {
+ test(s"Parquet field claims complex source - nested=$nested
reverse=$reverse kind=$kind") {
+ val (dataType, values): (DataType, Seq[Any]) = kind match {
+ case "struct" =>
+ (new StructType().add("v", LongType), Seq(Row(11L), Row(null), null))
+ case "array" =>
+ (
+ ArrayType(LongType, containsNull = true),
+ Seq(Seq(11L, null, 33L), Seq.empty[Long], null))
+ case "map" =>
+ (
+ MapType(StringType, LongType, valueContainsNull = true),
+ Seq(Map("k" -> 11L, "nil" -> null), Map.empty[String, Long], null))
+ }
+ val id1 = new MetadataBuilder().putLong("parquet.field.id", 1L).build()
+ val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build()
+ def wrapSchema(fields: StructType): StructType =
+ if (nested) StructType(Seq(StructField("s", fields, nullable = true,
id10))) else fields
+ def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields
+ val writeSchema =
+ wrapSchema(StructType(Seq(StructField("A", dataType, nullable = true,
id1))))
+ val fields = Seq(
+ StructField("x", dataType, nullable = true, id1),
+ StructField("a", dataType, nullable = true))
+ val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else
fields))
+ val expected =
+ values.map(value => wrapRow(if (reverse) Row(null, value) else
Row(value, null)))
+ withSQLConf(
+ SQLConf.CASE_SENSITIVE.key -> "false",
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .createDataFrame(
+ spark.sparkContext.parallelize(values.map(value =>
wrapRow(Row(value))), 1),
+ writeSchema)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ def query: DataFrame =
spark.read.schema(readSchema).parquet(path.getCanonicalPath)
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val df = query
+ val scans = collect(df.queryExecution.executedPlan) { case scan:
FileSourceScanExec =>
+ scan
+ }
+ assert(scans.nonEmpty && scans.forall(!_.supportsColumnar))
+ checkAnswer(df, expected)
+ }
+ checkNativeFieldClaimAnswer(query, expected)
+ }
+ }
+ }
+ }
+
+ for {
+ nested <- Seq(false, true)
+ reverse <- Seq(false, true)
+ count <- Seq(2, 3, 4)
+ } {
+ test(
+ s"Parquet field claims - repeated requested Parquet ids match parquet-mr
- nested=$nested reverse=$reverse count=$count") {
+ val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build()
+ def wrapSchema(fields: StructType): StructType =
+ if (nested) StructType(Seq(StructField("s", fields, nullable = true,
id10)))
+ else fields
+ def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields
+
+ val fields = Seq("x", "y", "z", "w").take(count).map(name =>
withFieldId(name, 1))
+ val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else
fields))
+ val writeSchema = wrapSchema(StructType(Seq(withFieldId("A", 1))))
+ val expected = Seq(11L, 22L, 33L).map(value =>
+ wrapRow(Row.fromSeq(Seq.fill[Any](count - 1)(null) :+ value)))
+ withSQLConf(
+ SQLConf.CASE_SENSITIVE.key -> "false",
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ val rows = Seq(11L, 22L, 33L).map(value => wrapRow(Row(value)))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(rows, 1),
writeSchema)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ // ParquetRowConverter builds its requested-id-to-index map with
toMap, so the last
+ // field with the repeated id owns the value. Check the Spark result
in both orders.
+ val sparkRows = fieldClaimReference {
+ spark.read.schema(readSchema).parquet(path.getCanonicalPath)
+ }
+ assert(sparkRows.sortBy(_.toString) == expected.sortBy(_.toString))
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.key
-> "true") {
+ val df =
spark.read.schema(readSchema).parquet(path.getCanonicalPath)
+ // Keep the existing JVM guard: native matching is tested
separately by the Rust
+ // Parquet roundtrip, which bypasses this guard and exercises the
actual converter.
+ val plan = df.queryExecution.executedPlan
+ assert(collect(plan) { case scan: CometNativeScanExec =>
+ scan
+ }.isEmpty)
+ val reasons = new ExtendedExplainInfo().getFallbackReasons(plan)
+ assert(reasons.exists(_.contains("duplicate Parquet field ids")),
s"$reasons\n$plan")
+ assert(df.collect().toSeq.sortBy(_.toString) ==
sparkRows.sortBy(_.toString))
+ }
+ }
+ }
+ }
+ }
+
+ test("Parquet field claims - pruned losing container does not add a schema
error") {
+ withFieldClaimFile(nested = true) { (path, _) =>
+ val fields = StructType(
+ Seq(
+ withFieldId("x", 1),
+ StructField("a", new StructType().add("v", LongType), nullable =
true)))
+ val schema = new StructType().add("s", fields)
+ withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") {
+ def query: DataFrame =
spark.read.schema(schema).parquet(path).select("s.x")
+ val expected = fieldClaimReference(query)
+ assert(expected == Seq(Row(1L), Row(2L), Row(3L)))
+ checkNativeFieldClaimAnswer(query, expected)
+ }
+ }
+ }
+
+ for {
+ shape <- Seq("struct", "array", "map")
+ reverse <- Seq(false, true)
+ empty <- Seq(false, true)
+ } {
+ test(
+ s"Parquet field claims - losing container still fails schema validation
- shape=$shape reverse=$reverse empty=$empty") {
+ def wrapType(fields: StructType): DataType = shape match {
+ case "struct" => fields
+ case "array" => ArrayType(fields, containsNull = true)
+ case "map" => MapType(StringType, fields, valueContainsNull = true)
+ }
+ val writeType = wrapType(StructType(Seq(withFieldId("A", 1))))
+ val fields = Seq(
+ withFieldId("x", 1),
+ StructField("a", new StructType().add("v", LongType), nullable = true))
+ val readType = wrapType(StructType(if (reverse) fields.reverse else
fields))
+ val value: Any = shape match {
+ case "struct" => Row(7L)
+ case "array" => Seq(Row(7L))
+ case "map" => Map("k" -> Row(7L))
+ }
+ withSQLConf(
+ SQLConf.CASE_SENSITIVE.key -> "false",
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ val rows = if (empty) Seq.empty[Row] else Seq(Row(value))
+ spark
+ .createDataFrame(
+ spark.sparkContext.parallelize(rows, 1),
+ new StructType().add("s", writeType))
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ def query: DataFrame =
+ spark.read.schema(new StructType().add("s",
readType)).parquet(path.getCanonicalPath)
+
+ for (vectorized <- Seq(false, true)) {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
+ val error = intercept[Exception](query.collect())
+ assert(causeMessages(error).contains("is not a group"),
causeMessages(error))
+ }
+ }
+ val df = query
+ assert(
+ collect(df.queryExecution.executedPlan) { case scan:
CometNativeScanExec =>
+ scan
+ }.nonEmpty,
+ s"expected native schema
validation:\n${df.queryExecution.executedPlan}")
+ val error = intercept[Exception](df.collect())
+ val messages = causeMessages(error)
+ assert(messages.contains("INT64") && messages.contains("A"),
messages)
+ }
+ }
+ }
+ }
+
+ for {
+ reverse <- Seq(false, true)
+ readIds <- Seq(false, true)
+ selected <- Seq("x", "y", "z")
+ } {
+ test(
+ s"Parquet field claims - repeated root ids single-column projection -
reverse=$reverse readIds=$readIds selected=$selected") {
+ val fields = Seq(withFieldId("x", 1), withFieldId("y", 1),
withFieldId("z", 2))
+ val readSchema = StructType(if (reverse) fields.reverse else fields)
+ val writeSchema = StructType(Seq(withFieldId("A", 1), withFieldId("z",
2)))
+ val expected = Seq(Row(if (selected == "z") 9L else if (readIds) 7L else
null))
+ withSQLConf(
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readIds.toString) {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(Seq(Row(7L,
9L)), 1), writeSchema)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ def query: DataFrame =
+
spark.read.schema(readSchema).parquet(path.getCanonicalPath).select(selected)
+
+ // Either Spark reader can read one alias alone. Keep both aliases
in dataSchema to
+ // exercise the difference between Spark's projection and native
root remapping.
+ for (vectorized <- Seq(false, true)) {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
+ val df = query
+ val scans = collect(df.queryExecution.executedPlan) {
+ case scan: FileSourceScanExec =>
+ scan
+ }
+ assert(scans.nonEmpty && scans.forall(_.supportsColumnar ==
vectorized))
+ scans.foreach { scan =>
+ assert(scan.requiredSchema.fieldNames.toSeq == Seq(selected))
+ assert(scan.relation.dataSchema.fieldNames.toSeq ==
readSchema.fieldNames.toSeq)
+ }
+ checkAnswer(df, expected)
+ }
+ }
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ val df = query
+ checkAnswer(df, expected)
+ val plan = df.queryExecution.executedPlan
+ val native = collect(plan) { case scan: CometNativeScanExec =>
scan }.nonEmpty
+ if (readIds && selected != "z") {
+ assert(!native, s"expected a fallback for a projected repeated
id:\n$plan")
+ val reasons = new ExtendedExplainInfo().getFallbackReasons(plan)
+ assert(
+ reasons.exists(_.contains("duplicate Parquet field ids")),
+ s"$reasons\n$plan")
+ } else {
+ // Unread duplicate IDs must not disable an unrelated root or
name-based reads.
+ assert(native, s"unexpected fallback for an unrelated root or
disabled IDs:\n$plan")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ for {
+ nested <- Seq(false, true)
+ offheap <- Seq(false, true)
Review Comment:
Now that the Spark vectorized branch only checks the plan, `offheap` no
longer changes anything this test asserts. Only Spark's vectorized reader reads
`spark.sql.columnVector.offheap.enabled`, and I could not find it under
`spark/src/main` or `native`. So two of the four cases repeat the other two.
The rows are also covered by the "Parquet field claims prefer id over name"
matrix, which adds both field orders and parent nulls. Could this block go,
with `IntegerType` added to that matrix if the type matters?
##########
spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala:
##########
@@ -1269,6 +1270,517 @@ class CometNativeReaderSuite extends CometTestBase with
AdaptiveSparkPlanHelper
nullable = true,
new MetadataBuilder().putLong("parquet.field.id", id.toLong).build())
+ private def withFieldClaimFile(nested: Boolean)(f: (String, StructType) =>
Unit): Unit = {
+ val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build()
+ def wrapSchema(fields: StructType): StructType =
+ if (nested) StructType(Seq(StructField("s", fields, nullable = true,
id10)))
+ else fields
+ val writeSchema = wrapSchema(StructType(Seq(withFieldId("A", 1))))
+ val readSchema = wrapSchema(
+ StructType(Seq(withFieldId("x", 1), StructField("a", LongType, nullable
= true))))
+ withSQLConf(
+ SQLConf.CASE_SENSITIVE.key -> "false",
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ val rows = (1L to 3L).map(value => if (nested) Row(Row(value)) else
Row(value))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(rows, 1),
writeSchema)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ f(path.getCanonicalPath, readSchema)
+ }
+ }
+ }
+
+ private def fieldClaimReference(query: => DataFrame): Seq[Row] = {
+ var rows = Seq.empty[Row]
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false") {
+ val df = query
+ val scans = collect(df.queryExecution.executedPlan) { case scan:
FileSourceScanExec =>
+ scan
+ }
+ assert(scans.nonEmpty && scans.forall(!_.supportsColumnar))
+ rows = df.collect().toSeq
+ }
+ rows
+ }
+
+ private def checkNativeFieldClaimAnswer(df: DataFrame, expected: Seq[Row]):
Unit = {
+ assert(
+ collect(df.queryExecution.executedPlan) { case scan: CometNativeScanExec
=> scan }.nonEmpty,
+ s"expected native field claim
resolution:\n${df.queryExecution.executedPlan}")
+ checkAnswer(df, expected)
+ }
+
+ for {
+ nested <- Seq(false, true)
+ pruning <- (if (nested) Seq(false, true) else Seq(false))
+ selected <- Seq("both", "id", "name")
+ } {
+ test(
+ s"Parquet field claims projection - nested=$nested pruning=$pruning
selected=$selected") {
+ withFieldClaimFile(nested) { (path, schema) =>
+ withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key ->
pruning.toString) {
+ def query: DataFrame = {
+ val df = spark.read.schema(schema).parquet(path)
+ val prefix = if (nested) "s." else ""
+ val names = selected match {
+ case "both" => Seq("x", "a")
+ case "id" => Seq("x")
+ case "name" => Seq("a")
+ }
+ df.select(names.map(name => col(prefix + name)): _*)
+ }
+ val expected = fieldClaimReference(query)
+ checkNativeFieldClaimAnswer(query, expected)
+ }
+ }
+ }
+ }
+
+ for {
+ nested <- Seq(false, true)
+ pushdown <- Seq(false, true)
+ predicate <- Seq("missing is null", "missing is not null", "id greater
than one")
+ } {
+ test(
+ s"Parquet field claims filters - nested=$nested pushdown=$pushdown
predicate=$predicate") {
+ withFieldClaimFile(nested) { (path, schema) =>
+ withSQLConf(
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key ->
pushdown.toString) {
+ def query: DataFrame = {
+ val df = spark.read.schema(schema).parquet(path)
+ val prefix = if (nested) "s." else ""
+ val x = col(prefix + "x")
+ val a = col(prefix + "a")
+ val condition = predicate match {
+ case "missing is null" => a.isNull
+ case "missing is not null" => a.isNotNull
+ case "id greater than one" => x > 1L
+ }
+ df.where(condition).select(x, a)
+ }
+ val expected = fieldClaimReference(query)
+ val values = predicate match {
+ case "missing is null" => Seq(1L, 2L, 3L)
+ case "missing is not null" => Seq.empty[Long]
+ case "id greater than one" => Seq(2L, 3L)
+ }
+ assert(expected.sortBy(_.toString) == values.map(value => Row(value,
null)))
+ val df = query
+ checkNativeFieldClaimAnswer(df, expected)
+ if (!nested && pushdown && predicate == "id greater than one") {
+ val scans = collect(df.queryExecution.executedPlan) {
+ case scan: CometNativeScanExec =>
+ scan
+ }
+ assert(
+ scans.exists(scan =>
+ scan.metrics("pushdown_rows_pruned").value +
+ scan.metrics("pushdown_rows_matched").value > 0))
+ }
+ }
+ }
+ }
+ }
+
+ for {
+ nested <- Seq(false, true)
+ reverse <- Seq(false, true)
+ kind <- Seq("struct", "array", "map")
+ } {
+ test(s"Parquet field claims complex source - nested=$nested
reverse=$reverse kind=$kind") {
+ val (dataType, values): (DataType, Seq[Any]) = kind match {
+ case "struct" =>
+ (new StructType().add("v", LongType), Seq(Row(11L), Row(null), null))
+ case "array" =>
+ (
+ ArrayType(LongType, containsNull = true),
+ Seq(Seq(11L, null, 33L), Seq.empty[Long], null))
+ case "map" =>
+ (
+ MapType(StringType, LongType, valueContainsNull = true),
+ Seq(Map("k" -> 11L, "nil" -> null), Map.empty[String, Long], null))
+ }
+ val id1 = new MetadataBuilder().putLong("parquet.field.id", 1L).build()
+ val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build()
+ def wrapSchema(fields: StructType): StructType =
+ if (nested) StructType(Seq(StructField("s", fields, nullable = true,
id10))) else fields
+ def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields
+ val writeSchema =
+ wrapSchema(StructType(Seq(StructField("A", dataType, nullable = true,
id1))))
+ val fields = Seq(
+ StructField("x", dataType, nullable = true, id1),
+ StructField("a", dataType, nullable = true))
+ val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else
fields))
+ val expected =
+ values.map(value => wrapRow(if (reverse) Row(null, value) else
Row(value, null)))
+ withSQLConf(
+ SQLConf.CASE_SENSITIVE.key -> "false",
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .createDataFrame(
+ spark.sparkContext.parallelize(values.map(value =>
wrapRow(Row(value))), 1),
+ writeSchema)
+ .write
+ .parquet(path.getCanonicalPath)
+ }
+ def query: DataFrame =
spark.read.schema(readSchema).parquet(path.getCanonicalPath)
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val df = query
+ val scans = collect(df.queryExecution.executedPlan) { case scan:
FileSourceScanExec =>
+ scan
+ }
+ assert(scans.nonEmpty && scans.forall(!_.supportsColumnar))
+ checkAnswer(df, expected)
+ }
+ checkNativeFieldClaimAnswer(query, expected)
+ }
+ }
+ }
+ }
+
+ for {
+ nested <- Seq(false, true)
+ reverse <- Seq(false, true)
+ count <- Seq(2, 3, 4)
Review Comment:
All 12 cases request every alias together, so on the Comet side they assert
the fallback main already has: `duplicate Parquet field ids among top-level
fields` at the root and `struct with duplicate Parquet field ids` when nested.
Those are already pinned by "native scan declines a nested struct whose fields
repeat a Parquet field id" and "native scan declines top-level fields that
repeat a Parquet field id" (lines 1826 and 1861). `count` and `reverse` do not
change which guard fires. Could this block go, or shrink to one root and one
nested case if the parquet-mr reference check is worth keeping?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
