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 a680634fab303..e096300e89c2c 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 @@ -292,7 +292,8 @@ class AvroCatalystDataConversionSuite extends SparkFunSuite 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 376d12a8a4923..76a8ce62cae8b 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 @@ -42,7 +42,7 @@ import org.apache.spark.sql.catalyst.expressions.AttributeReference import org.apache.spark.sql.catalyst.plans.logical.Filter import org.apache.spark.sql.catalyst.util.DateTimeTestUtils import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC} -import org.apache.spark.sql.execution.{FormattedMode, SparkPlan} +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.functions._ @@ -1735,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( @@ -3374,6 +3511,33 @@ class AvroV1Suite extends AvroSuite { .sparkConf .set(SQLConf.USE_V1_SOURCE_LIST, "avro") + test("SPARK-59108: two positional reads of different columns share one widened scan") { + // Subplan merging widens the projection of a shared V1 file scan, and this branch has nothing + // that keeps an avro relation out of it, so the two subqueries share one scan whatever this + // fix does. What it changes is the values they get: each column resolves against the data + // schema rather than against the merged projection. AQE off because `AdaptiveSparkPlanExec` + // is a leaf node, so with it on the scan underneath is not reachable from the executed plan. + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + withTempPath { dir => + val path = dir.getCanonicalPath + 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.toString).format("avro").load(path) + .createOrReplaceTempView("t") + // 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) + assert(scanColumns === Seq(Seq("b", "c"))) + } + } + } + } + test("SPARK-36271: V1 insert should check schema field name too") { withView("v") { spark.range(1).createTempView("v") 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 ba574c091dae9..003333c81ad48 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) @@ -285,8 +295,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]) @@ -426,15 +436,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 90781d4ad7077..a93757018b631 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 @@ -148,7 +148,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 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 02220e6c85688..1331649844c1a 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 @@ -286,17 +286,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 @@ -320,8 +331,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") @@ -375,11 +387,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) } /**