diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md index 96fe98812eb..84d55c8276e 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -94,6 +94,18 @@ TopK, and ranking window order keys run natively under both default and nondefau searches boundaries over a nullable nested key still runs on Spark ([#6477](https://github.com/apache/datafusion-comet/issues/6477)). +Spark's TopK (`TakeOrderedAndProject`, an `ORDER BY` with a `LIMIT`) evaluates a sort key after +the first only for the rows whose earlier keys tie. Comet evaluates every key over the whole batch, +so under `spark.sql.ansi.enabled=true` a later key such as `b + 1` could raise +`ARITHMETIC_OVERFLOW` on a row that Spark never evaluates it for. A TopK therefore runs on Spark +when Comet cannot show that a later sort key never fails, unless +`spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible=true` +([#7246](https://github.com/apache/datafusion-comet/issues/7246)). The check does not depend on +`spark.sql.ansi.enabled`: a later key built from an expression outside the set Comet recognizes as +unable to fail runs on Spark in either mode. A sort without a `LIMIT` that has no exchange beneath +it, such as `SORT BY`, still runs natively and can raise such an error +([#7470](https://github.com/apache/datafusion-comet/issues/7470)). + ## Hash Aggregation Ungrouped `AVG` and `TRY_AVG` on `DECIMAL(p, s)` fall back to Spark when the intermediate sum diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala index f57b6074a81..47350bcb348 100644 --- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala +++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala @@ -24,7 +24,7 @@ import scala.util.Try import org.apache.spark.SPARK_VERSION import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression, Literal} import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType, MapType, NullType, NumericType, StructType, TimestampNTZType, TimestampType} +import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DateType, DecimalType, MapType, NullType, NumericType, StringType, StructType, TimestampNTZType, TimestampType} import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, withFallbackReason} import org.apache.comet.DataTypeSupport.{containsDuplicateFieldNames, isComplexType} @@ -572,6 +572,39 @@ object CometCast private[comet] def canProduceNullMapKey(cast: Cast): Boolean = evalMode(cast) == CometEvalMode.TRY && mayNullMapKey(cast.child.dataType, cast.dataType) + /** + * Whether `cast` cannot raise an error for any value. A TRY cast returns NULL for a value it + * cannot convert, and so does a LEGACY cast between the atomic types listed here. An ANSI cast + * cannot fail only where the conversion is always exact, such as a widening one or a string to + * its bytes. A cast to or from a timestamp is excluded in every mode: the native kernels go + * through chrono, which rejects years Spark accepts, and raise even for TRY. So is a cast from + * a date to a string, whose native formatting raises in every mode for such a year. Both are + * checked at any depth, since a nested cast applies the same kernels to each element, field, + * key and value (https://github.com/apache/datafusion-comet/issues/7471). + */ + private[comet] def cannotFail(cast: Cast): Boolean = { + val from = cast.child.dataType + val to = cast.dataType + val hasTimestamp = Seq(from, to).exists( + SupportLevel.containsType(_, classOf[TimestampType], classOf[TimestampNTZType])) + val dateToString = SupportLevel.containsType(from, classOf[DateType]) && + SupportLevel.containsType(to, classOf[StringType]) + !hasTimestamp && !dateToString && (evalMode(cast) match { + case CometEvalMode.TRY => true + case CometEvalMode.LEGACY if isAtomicCastType(from) && isAtomicCastType(to) => true + case _ if from.isInstanceOf[StringType] && to == DataTypes.BinaryType => true + case _ => keyCastCannotFail(from, to) + }) + } + + // Timestamps are left out because `cannotFail` rejects them before reaching this check. + private def isAtomicCastType(dataType: DataType): Boolean = dataType match { + case _: NumericType | _: StringType | DataTypes.BooleanType | DataTypes.BinaryType | + DataTypes.DateType => + true + case _ => false + } + private def mayNullMapKey(fromType: DataType, toType: DataType): Boolean = (fromType, toType) match { case (MapType(fromKey, fromValue, _), MapType(toKey, toValue, _)) => diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index c8b89b0742c..552f72bbfd6 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.expressions.aggregate._ import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke} import org.apache.spark.sql.catalyst.expressions.xml.{XPathBoolean, XPathDouble, XPathFloat, XPathInt, XPathList, XPathLong, XPathShort, XPathString} +import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero import org.apache.spark.sql.comet.DecimalPrecision import org.apache.spark.sql.execution.{ScalarSubquery, SparkPlan} import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils @@ -1394,6 +1395,72 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { } } + /** + * Whether `expr` cannot raise an error for any row, so native execution may evaluate it for + * rows that Spark skips. It answers from Spark's semantics, independently of the native + * `cannot_fail` in `case_when.rs`, which judges the physical expressions it is given. It is + * conservative: an allowed expression whose children also cannot fail. Anything not listed is + * treated as able to fail. + */ + def cannotFail(expr: Expression): Boolean = { + def numeric: Boolean = expr.dataType.isInstanceOf[NumericType] + def nonAnsiNumeric(evalMode: EvalMode.Value): Boolean = evalMode != EvalMode.ANSI && numeric + // Floating-point arithmetic never overflows into an error, even in ANSI mode + def floating: Boolean = expr.dataType == FloatType || expr.dataType == DoubleType + val nodeCannotFail = expr match { + case _: Attribute | _: BoundReference | _: Literal => true + case _: GetStructField => true + case _: BinaryComparison | _: And | _: Or | _: Not | _: IsNull | _: IsNotNull => true + case _: In | _: InSet | _: Greatest | _: Least => true + case _: If | _: CaseWhen | _: Coalesce | _: IsNaN | _: NaNvl => true + case _: KnownFloatingPointNormalized | _: NormalizeNaNAndZero => true + case _: Substring | _: Upper | _: Lower | _: Length | _: OctetLength | _: BitLength => true + case _: StringTrim | _: StringTrimLeft | _: StringTrimRight => true + case _: StringReplace | _: InitCap => true + // These fail only when the result would exceed 2GB, more than a row can hold + case _: ConcatWs => true + case c: Concat => c.children.nonEmpty && c.dataType == StringType + case _: Reverse => expr.dataType == StringType + case _: Year | _: Quarter | _: Month | _: DayOfMonth | _: DayOfWeek | _: WeekDay | + _: DayOfYear | _: Hour | _: Minute | _: Second => + true + case _: Murmur3Hash | _: XxHash64 | _: Md5 | _: Sha1 | _: Sha2 | _: Crc32 => true + // Invalid input gives NaN or NULL, and a floating-point floor or ceil saturates to a long + case _: Floor | _: Ceil | _: Sqrt | _: Exp | _: Log | _: Signum => true + // Rounding an integer fails on overflow in ANSI mode; a decimal is never treated as safe + case r: Round => roundCannotFail(r.dataType, r.ansiEnabled) + case r: BRound => roundCannotFail(r.dataType, r.ansiEnabled) + case c: Cast => CometCast.cannotFail(c) + case _: UnaryPositive => true + case u: UnaryMinus => floating || (!u.failOnError && numeric) + case a: Abs => floating || (!a.failOnError && numeric) + case a: Add => floating || nonAnsiNumeric(a.evalMode) + case s: Subtract => floating || nonAnsiNumeric(s.evalMode) + case m: Multiply => floating || nonAnsiNumeric(m.evalMode) + case d: Divide => nonAnsiNumeric(d.evalMode) + case d: IntegralDivide => nonAnsiNumeric(d.evalMode) + case r: Remainder => nonAnsiNumeric(r.evalMode) + case p: Pmod => nonAnsiNumeric(p.evalMode) + case _ => false + } + nodeCannotFail && expr.children.forall(cannotFail) + } + + private def roundCannotFail(dataType: DataType, ansiEnabled: Boolean): Boolean = + dataType match { + case FloatType | DoubleType => true + case ByteType | ShortType | IntegerType | LongType => !ansiEnabled + case _ => false + } + + /** + * The first sort key after the first that can fail. Spark's ordering evaluates a sort key only + * for the rows whose earlier keys tie, but native execution evaluates every key over the whole + * batch, so such a key may raise on a row that Spark skips (#7246). + */ + def laterSortKeyThatCanFail(sortOrder: Seq[SortOrder]): Option[SortOrder] = + sortOrder.drop(1).find(s => !cannotFail(s.child)) + /** * Has native execution evaluate the arguments of `scalarFunc`, the serialized `expr`, the way * Spark evaluates those of a null-intolerant `BinaryExpression` or `TernaryExpression`: each diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala index f9597ca046b..a68b4b9acc1 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala @@ -33,8 +33,8 @@ import org.apache.spark.sql.vectorized.ColumnarBatch import org.apache.comet.{CometConf, ConfigEntry} import org.apache.comet.CometSparkSessionExtensions.isCometShuffleEnabled -import org.apache.comet.serde.{Compatible, OperatorOuterClass, SupportLevel, Unsupported} -import org.apache.comet.serde.QueryPlanSerde.{exprToProto, supportedSortType, withSharedSubexpressions} +import org.apache.comet.serde.{Compatible, Incompatible, OperatorOuterClass, SupportLevel, Unsupported} +import org.apache.comet.serde.QueryPlanSerde.{exprToProto, laterSortKeyThatCanFail, supportedSortType, withSharedSubexpressions} import org.apache.comet.serde.operator.CometSink object CometTakeOrderedAndProjectExec extends CometSink[TakeOrderedAndProjectExec] { @@ -63,6 +63,10 @@ object CometTakeOrderedAndProjectExec extends CometSink[TakeOrderedAndProjectExe if (!supportedSortType(op, op.sortOrder)) { return Unsupported(Some("unsupported data type in sort order")) } + laterSortKeyThatCanFail(op.sortOrder).foreach { s => + return Incompatible( + Some(s"sort key ${s.child.sql} can fail on rows that Spark's ordering skips")) + } Compatible() } diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql b/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql index 559574fc38f..9eec890d230 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql @@ -108,11 +108,18 @@ SELECT array_contains(ai, CAST(s AS INT)) FROM (SELECT * FROM array_null_short_circuit ORDER BY i LIMIT 10) -- A TakeOrderedAndProject compares rows one sort key at a time and evaluates its projection apart --- from the keys, so a cast in two sort keys, or in a sort key and the projection, is skipped too +-- from the keys, so a cast in two sort keys, or in a sort key and the projection, is skipped too. +-- A later sort key that can fail runs the top-K on Spark (#7246), so allow it here. +statement +SET spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible=true + query expect_native(array_contains, array_position) SELECT i FROM (SELECT * FROM array_null_short_circuit ORDER BY array_contains(ai, CAST(s AS INT)), array_position(ai, CAST(s AS INT)), i LIMIT 10) +statement +RESET spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible + query expect_native(array_contains, array_position) SELECT array_contains(ai, CAST(s AS INT)) FROM (SELECT * FROM array_null_short_circuit ORDER BY array_position(ai, CAST(s AS INT)), i LIMIT 10) diff --git a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql new file mode 100644 index 00000000000..33e34807e52 --- /dev/null +++ b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql @@ -0,0 +1,219 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Spark's top-K ordering evaluates a later sort key only for rows whose earlier keys tie, so a +-- later key that fails under ANSI mode does not fail on a row Spark skips. Comet evaluates every +-- key over the whole batch, so a top-K whose later key can fail runs on Spark (#7246). +-- Each table is one batch. + +-- Config: spark.sql.ansi.enabled=true +-- Config: spark.sql.adaptive.enabled=false +-- Config: spark.sql.parquet.outputTimestampType=TIMESTAMP_MICROS + +statement +CREATE TABLE topk(a int, b int) USING parquet + +statement +INSERT INTO topk SELECT /*+ REPARTITION(1) */ * FROM VALUES + (1, 2147483647), (2, 5), (3, 2147483647), (4, 7), (5, 9) AS t(a, b) + +-- No two rows tie on a, so Spark never evaluates b + 1 +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, b FROM topk ORDER BY a, b + 1 LIMIT 2 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, b FROM topk ORDER BY a DESC, CAST(CAST(b AS STRING) AS INT) * 2 LIMIT 2 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, b FROM topk ORDER BY a, abs(b) LIMIT 2 + +-- Later keys that cannot fail stay native +query +SELECT a, b FROM topk ORDER BY a, b LIMIT 2 + +query +SELECT a, b FROM topk ORDER BY a, b IS NULL, CAST(b AS BIGINT), try_add(b, 1) LIMIT 2 + +query +SELECT a, b FROM topk ORDER BY a, substr(CAST(b AS STRING), 1, 3), upper(CAST(b AS STRING)), + length(lower(CAST(b AS STRING))) LIMIT 2 + +-- A later key that can fail falls back even when an allowed expression wraps it +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, b FROM topk ORDER BY a, coalesce(b + 1, 0) LIMIT 2 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, b FROM topk ORDER BY a, IF(a > 3, b + 1, 0) LIMIT 2 + +-- Rounding an integer overflows under ANSI mode +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, b FROM topk ORDER BY a, round(b, -1) LIMIT 2 + +query +SELECT a, b FROM topk ORDER BY a, try_cast(CAST(b AS STRING) AS INT) LIMIT 2 + +statement +CREATE TABLE topk_types(a int, b int, l bigint, d double, s string, dt date, ts timestamp) +USING parquet + +statement +INSERT INTO topk_types SELECT /*+ REPARTITION(1) */ * FROM VALUES + (1, 2147483647, 9223372036854775807, double('NaN'), ' x ', date'2024-01-31', + timestamp'2024-01-31 23:59:59'), + (2, NULL, NULL, -0.0D, NULL, NULL, NULL), + (3, -2147483648, -9223372036854775808, 0.0D, 'abc', date'1970-01-01', + timestamp'1970-01-01 00:00:00'), + (4, 7, 7, 1.5D, 'Spark', date'2000-02-29', timestamp'2000-02-29 12:30:45'), + (5, 0, 0, -1.0E308D, '', date'9999-12-31', timestamp'9999-12-31 23:59:59') + AS t(a, b, l, d, s, dt, ts) + +-- Later keys built from functions that cannot fail stay native +query +SELECT a, s FROM topk_types ORDER BY a, trim(s), ltrim(s), rtrim(s), replace(s, 'a', 'b'), + reverse(s), octet_length(s), bit_length(s), concat(s, 'x'), concat_ws('-', s, s) LIMIT 3 + +query +SELECT a, dt FROM topk_types ORDER BY a, year(dt), quarter(dt), month(dt), dayofmonth(dt), + dayofweek(dt), weekday(dt), dayofyear(dt), hour(ts), minute(ts), second(ts) LIMIT 3 + +query +SELECT a, d FROM topk_types ORDER BY a, floor(d), ceil(d), sqrt(d), exp(d), ln(d), signum(d), + isnan(d), nanvl(d, 0D), round(d, 1), greatest(b, 0), least(l, 0), b IN (1, 2) LIMIT 3 + +query +SELECT a, s FROM topk_types ORDER BY a, hash(b, s), xxhash64(s), md5(s), sha1(s), sha2(s, 256), + crc32(s) LIMIT 3 + +query +SELECT a, s FROM topk_types ORDER BY a, initcap(s) LIMIT 3 + +-- Native timestamp casts reject years that Spark accepts, even for TRY +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, ts FROM topk_types ORDER BY a, CAST(ts AS STRING) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, ts FROM topk_types ORDER BY a, try_cast(ts AS DATE) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, dt FROM topk_types ORDER BY a, try_cast(dt AS TIMESTAMP) LIMIT 3 + +-- Native date formatting rejects years that Spark accepts, in every mode +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a, dt FROM topk_types ORDER BY a, CAST(dt AS STRING) LIMIT 3 + +statement +CREATE TABLE topk_nested(a int, tsa array, dta array, sts struct, + std struct, md map, mt map, ast array>) +USING parquet + +statement +INSERT INTO topk_nested SELECT /*+ REPARTITION(1) */ a, array(ts), array(dt), named_struct('f', ts), + named_struct('f', dt), map(1, dt), map(ts, 1), array(named_struct('f', ts)) FROM VALUES + (1, timestamp_micros(9000000000000000000), date_from_unix_date(100000000)), + (1, timestamp'2024-01-31 23:59:59', date'2024-01-31'), + (2, timestamp'1970-01-01 00:00:00', date'1970-01-01') AS t(a, ts, dt) + +-- The same native kernels cast each element, field, key and value of a nested type +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(tsa AS STRING) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(dta AS STRING) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(tsa AS array) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(dta AS array) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(sts AS STRING) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(std AS STRING) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(std AS struct) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(sts AS struct) LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(std AS struct) LIMIT 3 + +-- A map cannot be a sort key, so test it under IS NULL, which cannot fail by itself +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(md AS map) IS NULL LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(mt AS map) IS NULL LIMIT 3 + +query expect_fallback(can fail on rows that Spark's ordering skips) +SELECT a FROM topk_nested ORDER BY a, try_cast(ast AS STRING) LIMIT 3 + +-- Floating-point arithmetic cannot fail even under ANSI mode +query +SELECT a, d FROM topk_types ORDER BY a, d * 1.1D, d + 1, d - 1, -d, abs(d) LIMIT 3 + +statement +CREATE TABLE topk_struct(a int, s struct) USING parquet + +statement +INSERT INTO topk_struct SELECT /*+ REPARTITION(1) */ a, IF(x IS NULL, NULL, named_struct('x', x, + 'y', y)) FROM VALUES (1, 2147483647, 'p'), (1, 5, 'q'), (2, NULL, NULL), (3, -1, '') AS t(a, x, y) + +-- Reading a struct field cannot fail +query +SELECT a, s FROM topk_struct ORDER BY a, s.x, s.y LIMIT 3 + +-- The first key is evaluated for every row in Spark too when the input has at least two rows +-- (https://github.com/apache/datafusion-comet/issues/7527) +query +SELECT a, b FROM topk ORDER BY a + 1, b LIMIT 2 + +statement +CREATE TABLE topk_ties(a int, b int) USING parquet + +statement +INSERT INTO topk_ties SELECT /*+ REPARTITION(1) */ * FROM VALUES + (1, 2147483647), (1, 5), (2, 7) AS t(a, b) + +-- Rows tie on a, so Spark evaluates b + 1 and fails +query expect_error(ARITHMETIC_OVERFLOW) +SELECT a, b FROM topk_ties ORDER BY a, b + 1 LIMIT 2 + +-- Without ANSI mode the key cannot fail and stays native +statement +SET spark.sql.ansi.enabled=false + +query +SELECT a, b FROM topk ORDER BY a, b + 1 LIMIT 2 + +query +SELECT a, b FROM topk_ties ORDER BY a, b + 1 LIMIT 2 + +query +SELECT a, b FROM topk_ties ORDER BY a, -b, abs(b) LIMIT 2 + +query +SELECT a, b FROM topk ORDER BY a, round(b, -1), CAST(CAST(b AS STRING) AS INT) LIMIT 2 + +query +SELECT a, b FROM topk_types ORDER BY a, b / 0, b % 0, b DIV 0, l + 1, l * 2 LIMIT 3 + +query +SELECT a, s FROM topk_struct ORDER BY a, s.x, s.y LIMIT 3