diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala index a13faf3b51560..faa519b189418 100644 --- a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala +++ b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala @@ -106,7 +106,8 @@ case class AvroPartitionReaderFactory( avroFilters, options.useStableIdForUnionType, options.stableIdPrefixForUnionType, - options.recursiveFieldMaxDepth) + options.recursiveFieldMaxDepth, + dataSchema = Some(dataSchema)) override val stopPosition = partitionedFile.start + partitionedFile.length override def next(): Boolean = hasNextRow diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala index 49d597bfc8a77..83aa33ab80bf8 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala @@ -291,7 +291,8 @@ class AvroCatalystDataConversionSuite extends SharedSparkSession filters, false, "", - -1) + -1, + dataSchema = None) val deserialized = deserializer.deserialize(data) expected match { case None => assert(deserialized == None) diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala index d5b246840902c..2cace154420da 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala @@ -75,7 +75,8 @@ class AvroRowReaderSuite extends SharedSparkSession { new NoopFilters, false, "", - -1) + -1, + dataSchema = None) override val stopPosition = fileSize override def hasNext: Boolean = hasNextRow diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSchemaHelperSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSchemaHelperSuite.scala index 9364585619788..72e6deb1e99b3 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSchemaHelperSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSchemaHelperSuite.scala @@ -87,6 +87,49 @@ class AvroSchemaHelperSuite extends SharedSparkSession { assert(nameHelper.getAvroField("nonexist", 1).isEmpty) } + test("SPARK-59108: positional field match resolves against the data schema positions") { + val dataSchema = new StructType() + .add("a", IntegerType).add("b", IntegerType).add("c", IntegerType) + val avroSchema = SchemaConverters.toAvroType(dataSchema) + val projection = new StructType().add("c", IntegerType).add("a", IntegerType) + + val helper = new AvroUtils.AvroSchemaHelper( + avroSchema, projection, Seq(""), Seq(""), true, Array(2, 0)) + assert(helper.getAvroField("c", 0) === Some(avroSchema.getFields.get(2))) + assert(helper.getAvroField("a", 1) === Some(avroSchema.getFields.get(0))) + assert(helper.matchedFields.map(_.avroField.name()) === Seq("c", "a")) + + // With no positions a field's own position is used, which is what an unprojected match needs. + val unprojected = + new AvroUtils.AvroSchemaHelper(avroSchema, projection, Seq(""), Seq(""), true) + assert(unprojected.getAvroField("c", 0) === Some(avroSchema.getFields.get(0))) + + // The shape both read paths produce is an ascending subsequence of the data schema. + val ascending = new StructType().add("a", IntegerType).add("c", IntegerType) + val ascendingHelper = new AvroUtils.AvroSchemaHelper( + avroSchema, ascending, Seq(""), Seq(""), true, Array(0, 2)) + assert(ascendingHelper.getAvroField("a", 0) === Some(avroSchema.getFields.get(0))) + assert(ascendingHelper.getAvroField("c", 1) === Some(avroSchema.getFields.get(2))) + assert(ascendingHelper.matchedFields.map(_.avroField.name()) === Seq("a", "c")) + + val msg = intercept[IllegalArgumentException] { + new AvroUtils.AvroSchemaHelper(avroSchema, projection, Seq(""), Seq(""), true, Array(2)) + }.getMessage + assert(msg.contains("Got 1 data schema positions for 2 Catalyst fields")) + + // A missing field is reported by the position that was looked for, not by the position the + // field happens to have in the projection. + val twoFieldAvro = SchemaConverters.toAvroType( + new StructType().add("a", IntegerType).add("b", IntegerType)) + val pastTheEnd = new AvroUtils.AvroSchemaHelper( + twoFieldAvro, new StructType().add("c", IntegerType, nullable = false), + Seq(""), Seq(""), true, Array(2)) + val missing = intercept[IncompatibleSchemaException] { + pastTheEnd.validateNoExtraCatalystFields(ignoreNullable = false) + }.getMessage + assert(missing.contains("Cannot find field at position 2")) + } + test("properly match fields between Avro and Catalyst schemas") { val catalystSchema = StructType( Seq("catalyst1", "catalyst2", "shared1", "shared2").map(StructField(_, IntegerType)) diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala index 3643a95abe19c..5aa6e4d703f52 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala @@ -229,7 +229,8 @@ object AvroSerdeSuite { new NoopFilters, false, "", - -1) + -1, + dataSchema = None) } /** diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala index b03d30859e5f6..e7214519577c7 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala @@ -45,7 +45,6 @@ import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone import org.apache.spark.sql.execution.{FileSourceScanExec, FormattedMode, SparkPlan} import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition} import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileDataSourceV2, FileTable} -import org.apache.spark.sql.execution.planmerging.MergeSubplans import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.LegacyBehaviorPolicy import org.apache.spark.sql.internal.LegacyBehaviorPolicy._ @@ -1736,6 +1735,143 @@ abstract class AvroSuite } } + test("SPARK-59108: positionalFieldMatching resolves fields against the full schema") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 5).selectExpr("id AS a", "id * 100 AS b", "id * 10000 AS c") + .write.format("avro").save(path) + // The names differ from the file's, so only the positions can pair the two schemas. + val renamedSchema = new StructType() + .add("x", LongType).add("y", LongType).add("z", LongType) + val df = spark.read.format("avro") + .option("positionalFieldMatching", true.toString) + .schema(renamedSchema) + .load(path) + + val rows = (0 until 5).map(i => Row(i.toLong, i * 100L, i * 10000L)) + checkAnswer(df, rows) + // A column keeps its own Avro field however few of them the query projects. + checkAnswer(df.select("z"), rows.map(r => Row(r.get(2)))) + checkAnswer(df.select("y"), rows.map(r => Row(r.get(1)))) + checkAnswer(df.select("x", "z"), rows.map(r => Row(r.get(0), r.get(2)))) + checkAnswer(df.select("z", "x"), rows.map(r => Row(r.get(2), r.get(0)))) + checkAnswer(df.select("y", "z"), rows.map(r => Row(r.get(1), r.get(2)))) + checkAnswer(df.selectExpr("sum(z)"), Row(100000L)) + // With pushdown on, the filter runs inside the deserializer; with it off, it runs above the + // scan. + // Either way a wrong pairing drops rows rather than only returning wrong values for them. + Seq("true", "false").foreach { pushDown => + withSQLConf(SQLConf.AVRO_FILTER_PUSHDOWN_ENABLED.key -> pushDown) { + checkAnswer(df.where("z = 20000").select("z"), Row(20000L)) + checkAnswer(df.where("z > 20000").select("x"), Seq(Row(3L), Row(4L))) + } + } + // A projection of no columns at all. + checkAnswer(df.selectExpr("count(1)"), Row(5L)) + + // The projected schema carries the schema's own spelling whatever casing the query used, so + // the name lookup that resolves a position finds the field either way. + val mixedCase = spark.read.format("avro") + .option("positionalFieldMatching", true.toString) + .schema(new StructType().add("Xx", LongType).add("yY", LongType).add("ZZ", LongType)) + .load(path) + Seq("true", "false").foreach { caseSensitive => + withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive) { + checkAnswer(mixedCase.select("ZZ"), rows.map(r => Row(r.get(2)))) + } + } + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + checkAnswer(mixedCase.select("zz"), rows.map(r => Row(r.get(2)))) + } + } + } + + test("SPARK-59108: positionalFieldMatching with a partition column in the schema") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 4).selectExpr("id AS a", "id * 100 AS b", "id % 2 AS p") + .write.partitionBy("p").format("avro").save(path) + // p is a partition column, so the files hold a and b only and the data schema is x and z. + val df = spark.read.format("avro") + .option("positionalFieldMatching", true.toString) + .schema("x long, p int, z long") + .load(path) + + checkAnswer(df.select("z"), (0 until 4).map(i => Row(i * 100L))) + checkAnswer(df.select("x"), (0 until 4).map(i => Row(i.toLong))) + checkAnswer(df.select("p", "z"), (0 until 4).map(i => Row(i % 2, i * 100L))) + checkAnswer(df.where("p = 1").select("z"), Seq(Row(100L), Row(300L))) + } + } + + test("SPARK-59108: positionalFieldMatching with a nested record and the avroSchema option") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 3).selectExpr( + "id AS a", + "named_struct('f1', id * 10, 'f2', cast(id AS string)) AS r", + "id * 1000 AS c") + .write.format("avro").save(path) + + // Only the top level is a projection, so the nested record keeps resolving by its own + // positions. Reading the struct alone would take Avro field 0, a long, and fail. + val df = spark.read.format("avro") + .option("positionalFieldMatching", true.toString) + .schema("x long, s struct, z long") + .load(path) + checkAnswer(df.select("s"), (0 until 3).map(i => Row(Row(i * 10L, i.toString)))) + checkAnswer(df.select("s.g2"), (0 until 3).map(i => Row(i.toString))) + checkAnswer(df.select("z"), (0 until 3).map(i => Row(i * 1000L))) + + // The avroSchema option supplies the Avro side, and the data schema is inferred from it, so + // the positions are the option's. + val avroSubset = + """{"type":"record","name":"topLevelRecord","fields":[ + |{"name":"a","type":"long"}, + |{"name":"c","type":"long"}]}""".stripMargin + val fromOption = spark.read.format("avro") + .option("positionalFieldMatching", true.toString) + .option("avroSchema", avroSubset) + .load(path) + checkAnswer(fromOption.select("c"), (0 until 3).map(i => Row(i * 1000L))) + checkAnswer(fromOption.select("a"), (0 until 3).map(i => Row(i.toLong))) + } + } + + test("SPARK-59108: a position past the end of the Avro schema reads null") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 3).selectExpr("id AS a", "id * 100 AS b").write.format("avro").save(path) + val df = spark.read.format("avro") + .option("positionalFieldMatching", true.toString) + .schema("x long, y long, z long") + .load(path) + + // z is at position 2 of the schema and the file has two fields, so it has no Avro field to + // read and comes back null however few columns the query projects. + checkAnswer(df.select("z"), Seq(Row(null), Row(null), Row(null))) + checkAnswer(df, (0 until 3).map(i => Row(i.toLong, i * 100L, null))) + } + } + + test("SPARK-59108: positionalFieldMatching fails a mispaired type rather than reading it") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 3).selectExpr("id AS a", "cast(id AS string) AS b", "id * 10 AS c") + .write.format("avro").save(path) + val df = spark.read.format("avro") + .option("positionalFieldMatching", true.toString) + .schema("x long, y long, z long") + .load(path) + + // y takes Avro field 1, which is a string, so the read fails instead of returning the values + // of a neighbouring field. + val ex = intercept[SparkException](df.select("y").collect()) + assert(Utils.exceptionString(ex).contains("Cannot convert Avro")) + checkAnswer(df.select("z"), (0 until 3).map(i => Row(i * 10L))) + } + } + test("int/long double/float conversion") { val catalystSchema = StructType(Seq( @@ -3733,37 +3869,30 @@ class AvroV1Suite extends AvroSuite { .sparkConf .set(SQLConf.USE_V1_SOURCE_LIST, "avro") - test("SPARK-59107: positionalFieldMatching makes an avro read projection-sensitive") { - // Strictness pinned rather than inherited, so that positional matching is the only reason the - // read is projection-sensitive. AQE off because `AdaptiveSparkPlanExec` is a leaf node, so with - // it on the scans underneath it are not reachable from the executed plan. + test("SPARK-59108: two positional reads of different columns share one widened scan") { + // SPARK-59107 named avro under this option, so the two subqueries used to keep their own scans. + // They share one now, and the values are the file's either way because each column resolves + // against the data schema. AQE off because `AdaptiveSparkPlanExec` is a leaf node, so with it + // on the scan underneath is not reachable from the executed plan. withSQLConf( SQLConf.IGNORE_CORRUPT_FILES.key -> "false", SQLConf.IGNORE_MISSING_FILES.key -> "false", SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { withTempPath { dir => val path = dir.getCanonicalPath - spark.range(0, 5).selectExpr("id AS a", "id * 10 AS b").write.format("avro").save(path) + spark.range(0, 5).selectExpr("id AS a", "id * 10 AS b", "id * 100 AS c") + .write.format("avro").save(path) withTempView("t") { - spark.read.option("positionalFieldMatching", "true").format("avro").load(path) + spark.read.option("positionalFieldMatching", true.toString).format("avro").load(path) .createOrReplaceTempView("t") - val query = "SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)" - // Compared against the same query with merging excluded rather than against a literal - // row: positional matching resolves a column against its position in the read schema, so - // what `sum(b)` answers depends on its own subquery's projection. What this test pins is - // that merging changes neither value. - val unmerged = withSQLConf( - SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> MergeSubplans.ruleName) { - sql(query).collect().toSeq - } - val df = sql(query) - checkAnswer(df, unmerged) + // b and c sit at data schema positions 1 and 2, so the merged read of the two has to + // resolve against the data schema rather than against its own projection. + val df = sql("SELECT (SELECT sum(b) FROM t), (SELECT sum(c) FROM t)") + checkAnswer(df, Row(100L, 1000L)) val scanColumns = df.queryExecution.executedPlan .collectWithSubqueries { case s: FileSourceScanExec => s } .map(_.requiredSchema.fieldNames.sorted.toSeq) - .sortBy(_.mkString(",")) - // One entry per column means the two subqueries kept their own scans. - assert(scanColumns === Seq(Seq("a"), Seq("b"))) + assert(scanColumns === Seq(Seq("b", "c"))) } } } diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala b/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala index eb64fa2a6e5f6..6561f68f5d14a 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala @@ -22,9 +22,11 @@ import java.io.{ByteArrayInputStream, IOException} import org.apache.avro.file.{DataFileConstants, DataFileStream} import org.apache.avro.generic.{GenericDatumReader, GenericRecord} +import org.apache.spark.sql.Row + /** - * Binds [[ArchiveReadSuiteBase]]'s hooks to Avro, adding the streaming-reader regression tests - * that have no format-agnostic analogue. + * Binds [[ArchiveReadSuiteBase]]'s hooks to Avro, adding the Avro-only tests that have no + * format-agnostic analogue. */ trait AvroArchiveReadBase extends ArchiveReadSuiteBase { @@ -47,6 +49,22 @@ trait AvroArchiveReadBase extends ArchiveReadSuiteBase { // ----- Avro-specific tests ------------------------------------------------- + test("Avro: positionalFieldMatching resolves a pruned read against the full schema") { + // This is a second deserializer construction site, and it is handed a pruned required schema + // like the per-file reader is (SPARK-59108). The two columns have different types, so a wrong + // pairing fails the read. + withArchiveFile() { archive => + writeArchive(archive, Seq(entryName(0) -> encodeFile(sampleDf((1, "Alice"), (2, "Bob"))))) + val df = read( + archive.getCanonicalPath, + extraOptions = Map("positionalFieldMatching" -> "true"), + schema = "num INT, label STRING") + checkAnswer(df.select("label"), Seq(Row("Alice"), Row("Bob"))) + checkAnswer(df.select("num"), Seq(Row(1), Row(2))) + checkAnswer(df, Seq(Row(1, "Alice"), Row(2, "Bob"))) + } + } + test("Avro: a truncated entry fails fast instead of spinning") { // A DataFileStream must throw on a truncated entry rather than loop forever. Cut at the header // and mid-file; the per-test timeout catches a regression to a spin. diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md index 16be980896b5f..553f4d16512b4 100644 --- a/docs/sql-performance-tuning.md +++ b/docs/sql-performance-tuning.md @@ -272,7 +272,7 @@ SELECT They are merged into one aggregate that computes `min` and `max` together, so `store_sales` is read once. In `EXPLAIN` output a merged subplan shows up as a subquery whose single output column is named `mergedValue`, and the sites that share it as `ReusedSubquery`. -Two subplans are merged when their plans match node by node: `Project` lists are unioned, `Aggregate`s must have the same grouping and use the same aggregation implementation (so a `min` is not merged with a `collect_list`), `Filter`s must have the same condition, `Join`s must have the same type, condition and hints, and the leaves must read the same input. A V1 file relation whose rows depend on which columns the read asked for is merged only when both subplans read the same columns of it: `csv`, `json` and `xml`, whose parsers decide what counts as a malformed record from the required schema, `avro` read with `positionalFieldMatching`, which pairs a column with the Avro field at its position in that schema, and any file relation read with `spark.sql.files.ignoreCorruptFiles` enabled, as a read option or through the configuration, where a failure in a column only one side reads is swallowed together with the rest of that file's rows. `spark.sql.files.ignoreMissingFiles` counts too, not for that reason but because one predicate answers for both. Subplans that differ only in their `WHERE` conditions can be merged as well, by turning each side's condition into a boolean column and giving each side's aggregate expressions a `FILTER (WHERE ...)` clause. That is controlled by the configurations below. Queries that still contain a `WITH` clause when this rule runs (one that was not inlined) are skipped. +Two subplans are merged when their plans match node by node: `Project` lists are unioned, `Aggregate`s must have the same grouping and use the same aggregation implementation (so a `min` is not merged with a `collect_list`), `Filter`s must have the same condition, `Join`s must have the same type, condition and hints, and the leaves must read the same input. A V1 file relation whose rows depend on which columns the read asked for is merged only when both subplans read the same columns of it: `csv`, `json` and `xml`, whose parsers decide what counts as a malformed record from the required schema, and any file relation read with `spark.sql.files.ignoreCorruptFiles` enabled, as a read option or through the configuration, where a failure in a column only one side reads is swallowed together with the rest of that file's rows. `spark.sql.files.ignoreMissingFiles` counts too, not for that reason but because one predicate answers for both. Subplans that differ only in their `WHERE` conditions can be merged as well, by turning each side's condition into a boolean column and giving each side's aggregate expressions a `FILTER (WHERE ...)` clause. That is controlled by the configurations below. Queries that still contain a `WITH` clause when this rule runs (one that was not inlined) are skipped. When only one of the two subplans has a filter, merging is always beneficial, because the unfiltered side reads all the data anyway. This case is on by default, unless the filter has to cross a `Join` to reach the aggregate, which needs the through-join configuration below. When both sides have a filter (the symmetric case), the merged scan filter becomes `OR(f1, f2)`, which is less selective than either original filter and can therefore read more data - for example when the filters prune partitions or Parquet row groups. That is why the symmetric case is disabled by default. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala index ce16c4a2cc3ae..4170b9caa7643 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala @@ -44,6 +44,13 @@ import org.apache.spark.unsafe.types.UTF8String /** * A deserializer to deserialize data in avro format to data in catalyst format. + * + * @param dataSchema The schema `rootCatalystType` was projected from, for a read that prunes + * columns. A positional field match pairs a Catalyst field with the Avro field + * at the same position, and that position is the one in the full schema rather + * than in the projection: without this, reading only the third column would take + * the first Avro field. `None` when the Catalyst type is not a projection; + * unused when field matching is by name. */ private[sql] class AvroDeserializer( rootAvroType: Schema, @@ -53,7 +60,8 @@ private[sql] class AvroDeserializer( filters: StructFilters, useStableIdForUnionType: Boolean, stableIdPrefixForUnionType: String, - recursiveFieldMaxDepth: Int) { + recursiveFieldMaxDepth: Int, + dataSchema: Option[StructType]) { def this( rootAvroType: Schema, @@ -70,7 +78,8 @@ private[sql] class AvroDeserializer( new NoopFilters, useStableIdForUnionType, stableIdPrefixForUnionType, - recursiveFieldMaxDepth) + recursiveFieldMaxDepth, + dataSchema = None) } private lazy val decimalConversions = new DecimalConversion() @@ -91,7 +100,8 @@ private[sql] class AvroDeserializer( val resultRow = new SpecificInternalRow(st.map(_.dataType)) val fieldUpdater = new RowUpdater(resultRow) val applyFilters = filters.skipRow(resultRow, _) - val writer = getRecordWriter(rootAvroType, st, Nil, Nil, applyFilters) + val writer = + getRecordWriter(rootAvroType, st, Nil, Nil, applyFilters, positionsInDataSchema(st)) (data: Any) => { val record = data.asInstanceOf[GenericRecord] val skipRow = writer(fieldUpdater, record) @@ -307,8 +317,8 @@ private[sql] class AvroDeserializer( case (RECORD, st: StructType) => // Avro datasource doesn't accept filters with nested attributes. See SPARK-32328. // We can always return `false` from `applyFilters` for nested records. - val writeRecord = - getRecordWriter(avroType, st, avroPath, catalystPath, applyFilters = _ => false) + val writeRecord = getRecordWriter( + avroType, st, avroPath, catalystPath, applyFilters = _ => false, Array.empty) (updater, ordinal, value) => val row = new SpecificInternalRow(st) writeRecord(new RowUpdater(row), value.asInstanceOf[GenericRecord]) @@ -448,15 +458,44 @@ private[sql] class AvroDeserializer( } } + /** + * The position of each `projection` field in `dataSchema`, which is what a positional field match + * resolves against. Empty when there is no data schema to resolve against, or when field matching + * is by name and the positions are unused. + * + * This takes a data schema position for an Avro field position, which `recursiveFieldMaxDepth` + * can break: `SchemaConverters` drops a field it will not recurse into, so the data schema is a + * gapped view of the Avro schema and every field after the gap resolves one position early. + * Positional matching is already wrong for such a schema without this method, because the fields + * after the gap shift by one whatever the projection is. + */ + private def positionsInDataSchema(projection: StructType): Array[Int] = dataSchema match { + case Some(schema) if positionalFieldMatch => + projection.map(field => schema.fieldIndex(field.name)).toArray + case _ => Array.empty + } + + /** + * Creates a writer that reads a record's fields into `catalystType`'s fields. + * + * @param dataSchemaPositions The positions a positional field match resolves `catalystType`'s + * fields against, empty to use each field's own position. Only the + * root record passes them: a nested record is never a projection, + * because V1 nested pruning is limited to Parquet and ORC + * (`SchemaPruning.canPruneDataSchema`) and V2's + * `FileScanBuilder.supportsNestedSchemaPruning` is false for Avro. + */ private def getRecordWriter( avroType: Schema, catalystType: StructType, avroPath: Seq[String], catalystPath: Seq[String], - applyFilters: Int => Boolean): (CatalystDataUpdater, GenericRecord) => Boolean = { + applyFilters: Int => Boolean, + dataSchemaPositions: Array[Int]) + : (CatalystDataUpdater, GenericRecord) => Boolean = { val avroSchemaHelper = new AvroUtils.AvroSchemaHelper( - avroType, catalystType, avroPath, catalystPath, positionalFieldMatch) + avroType, catalystType, avroPath, catalystPath, positionalFieldMatch, dataSchemaPositions) avroSchemaHelper.validateNoExtraCatalystFields(ignoreNullable = true) // no need to validateNoExtraAvroFields since extra Avro fields are ignored diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala index 5988724070961..69124741884e3 100755 --- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala @@ -106,7 +106,7 @@ private[sql] class AvroFileFormat extends FileFormat // A tar archive (always a single split, see `isSplitable`) is streamed entry by entry when // archive reads are enabled; otherwise the file is read directly. The V2 data source has // no archive support, so this dispatch lives here. - readArchive(file, conf, parsedOptions, requiredSchema, filters) + readArchive(file, conf, parsedOptions, dataSchema, requiredSchema, filters) } else { val userProvidedSchema = parsedOptions.schema @@ -161,7 +161,8 @@ private[sql] class AvroFileFormat extends FileFormat avroFilters, parsedOptions.useStableIdForUnionType, parsedOptions.stableIdPrefixForUnionType, - parsedOptions.recursiveFieldMaxDepth) + parsedOptions.recursiveFieldMaxDepth, + dataSchema = Some(dataSchema)) override val stopPosition = file.start + file.length override def hasNext: Boolean = hasNextRow @@ -188,6 +189,7 @@ private[sql] class AvroFileFormat extends FileFormat file: PartitionedFile, conf: Configuration, parsedOptions: AvroOptions, + dataSchema: StructType, requiredSchema: StructType, filters: Seq[Filter]): Iterator[InternalRow] = { val userProvidedSchema = parsedOptions.schema @@ -218,7 +220,8 @@ private[sql] class AvroFileFormat extends FileFormat avroFilters, parsedOptions.useStableIdForUnionType, parsedOptions.stableIdPrefixForUnionType, - parsedOptions.recursiveFieldMaxDepth) + parsedOptions.recursiveFieldMaxDepth, + dataSchema = Some(dataSchema)) // The record is deserialized eagerly in `hasNext` because `AvroDeserializer#deserialize` may // filter rows (returning None); the stream is closed once its records are exhausted. new Iterator[InternalRow] with Closeable { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala index 266e6ee835ced..56bac9c448c26 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala @@ -373,17 +373,28 @@ private[sql] object AvroUtils extends Logging { * @param positionalFieldMatch If true, perform field matching in a positional fashion * (structural comparison between schemas, ignoring names); * otherwise, perform field matching using field names. + * @param dataSchemaPositions The position of each `catalystSchema` field in the schema it was + * projected from, for a positional match against a projection. A + * positional match pairs a Catalyst field with the Avro field at the + * same position, and that position is the one in the full schema, so + * a read of only the third column still takes the third Avro field. + * Empty when `catalystSchema` is not a projection, in which case a + * field's own position is used. */ class AvroSchemaHelper( avroSchema: Schema, catalystSchema: StructType, avroPath: Seq[String], catalystPath: Seq[String], - positionalFieldMatch: Boolean) { + positionalFieldMatch: Boolean, + dataSchemaPositions: Array[Int] = Array.empty) { if (avroSchema.getType != Schema.Type.RECORD) { throw new IncompatibleSchemaException( s"Attempting to treat ${avroSchema.getName} as a RECORD, but it was: ${avroSchema.getType}") } + require(dataSchemaPositions.isEmpty || dataSchemaPositions.length == catalystSchema.length, + s"Got ${dataSchemaPositions.length} data schema positions for " + + s"${catalystSchema.length} Catalyst fields") private[this] val avroFieldArray = avroSchema.getFields.asScala.toArray private[this] val fieldMap = avroSchema.getFields.asScala @@ -407,8 +418,9 @@ private[sql] object AvroUtils extends Logging { if (getAvroField(sqlField.name, sqlPos).isEmpty && (!ignoreNullable || !sqlField.nullable)) { if (positionalFieldMatch) { - throw new IncompatibleSchemaException("Cannot find field at position " + - s"$sqlPos of ${toFieldStr(avroPath)} from Avro schema (using positional matching)") + throw new IncompatibleSchemaException( + s"Cannot find field at position ${avroPosition(sqlPos)} of " + + s"${toFieldStr(avroPath)} from Avro schema (using positional matching)") } else { throw new IncompatibleSchemaException( s"Cannot find ${toFieldStr(catalystPath :+ sqlField.name)} in Avro schema") @@ -462,11 +474,15 @@ private[sql] object AvroUtils extends Logging { /** Get the Avro field corresponding to the provided Catalyst field name/position, if any. */ def getAvroField(fieldName: String, catalystPos: Int): Option[Schema.Field] = { if (positionalFieldMatch) { - avroFieldArray.lift(catalystPos) + avroFieldArray.lift(avroPosition(catalystPos)) } else { getFieldByName(fieldName) } } + + /** The Avro field position a positional match pairs the given Catalyst position with. */ + private def avroPosition(catalystPos: Int): Int = + if (dataSchemaPositions.isEmpty) catalystPos else dataSchemaPositions(catalystPos) } /** diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala index 66f87d93e9bab..2eb4d04c7d897 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala @@ -29,7 +29,6 @@ import org.json4s.jackson.Serialization import org.apache.spark.{SparkException, SparkUpgradeException} import org.apache.spark.sql.{sources, SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_LEGACY_INT96_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY, SPARK_VERSION_METADATA_KEY} -import org.apache.spark.sql.avro.{AvroFileFormat, AvroOptions} import org.apache.spark.sql.catalyst.FileSourceOptions import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils} import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, Expression, ExpressionSet, PredicateHelper} @@ -174,14 +173,12 @@ object DataSourceUtils extends PredicateHelper { * Two things put a V1 file source here. Its parser may resolve or validate a column against the * set of columns it was asked for, which lets a wider read drop or rewrite rows that the narrower * one returned: CSV, JSON and XML build their parser from the required schema and take `mode` and - * the corrupt-record column from it, and Avro under `positionalFieldMatching` pairs a column with - * the Avro field at its position in that schema. SPARK-59108 proposes removing that at the root; - * the Avro arm of `hasProjectionSensitiveParser` can go once that fix is on this branch. Or the - * read is not strict: under `ignoreCorruptFiles` a failure in a column only the wider read - * touches is swallowed together with the rest of that file's rows, whatever the format. - * `ignoreMissingFiles` has no such mechanism, since a missing file is skipped whatever is - * projected; it is here to match `FileSourceOptions.hasStrictFileReads`, the same predicate the - * reader and the cache-repeatability check in `InMemoryRelation` use. + * the corrupt-record column from it. Or the read is not strict: under `ignoreCorruptFiles` a + * failure in a column only the wider read touches is swallowed together with the rest of that + * file's rows, whatever the format. `ignoreMissingFiles` has no such mechanism, since a missing + * file is skipped whatever is projected; it is here to match + * `FileSourceOptions.hasStrictFileReads`, the same predicate the reader and the + * cache-repeatability check in `InMemoryRelation` use. * * Callers that widen a read need this. Subplan merging is one: top-level column pruning for a V1 * file source happens in physical planning, from the attributes referenced above the relation, so @@ -191,19 +188,12 @@ object DataSourceUtils extends PredicateHelper { private[sql] def isProjectionSensitiveRead(relation: BaseRelation): Boolean = relation match { case hs: HadoopFsRelation => !new FileSourceOptions(hs.options).hasStrictFileReads || - hasProjectionSensitiveParser(hs.fileFormat, hs.options) + hasProjectionSensitiveParser(hs.fileFormat) case _ => false } - private def hasProjectionSensitiveParser( - fileFormat: FileFormat, options: Map[String, String]): Boolean = fileFormat match { + private def hasProjectionSensitiveParser(fileFormat: FileFormat): Boolean = fileFormat match { case _: CSVFileFormat | _: JsonFileFormat | _: XmlFileFormat => true - // Read the option off the map rather than through `AvroOptions`, whose constructor resolves - // `avroSchemaUrl` and would do I/O here, and read it leniently so a malformed value still - // fails where Avro reports it rather than here. - case _: AvroFileFormat => - CaseInsensitiveMap(options).get(AvroOptions.POSITIONAL_FIELD_MATCHING) - .exists("true".equalsIgnoreCase) case _ => false }