From 038c839895dace0eaa942c6a69398f1d5fe3dfb4 Mon Sep 17 00:00:00 2001 From: Peter Toth Date: Wed, 1 Apr 2026 12:04:59 +0200 Subject: [PATCH 1/4] [SPARK-56321][SQL] Pass function catalog to toCatalystOrdering in V2ScanPartitioningAndOrdering ### What changes were proposed in this pull request? `V2ScanPartitioningAndOrdering.ordering` was calling `V2ExpressionUtils.toCatalystOrdering` without the `funCatalog` argument. This meant that function-based sort expressions reported by a data source via `SupportsReportOrdering` (e.g. transform functions like `bucket(n, col)`) could not be resolved against the function catalog and would be silently dropped. The fix passes `relation.funCatalog` as the third argument, consistent with how `toCatalystOpt` is already called in the `partitioning` rule of the same object. ### Why are the changes needed? Without the function catalog, sort orders involving catalog functions reported by `SupportsReportOrdering` are not resolved, causing them to be ignored by the planner even when the data source correctly reports them. ### Does this PR introduce _any_ user-facing change? Yes. Data sources implementing `SupportsReportOrdering` with function-based sort expressions that require the function catalog will now have those sort orders correctly recognized by Spark, potentially eliminating unnecessary sort operations. ### How was this patch tested? `WriteDistributionAndOrderingSuite` already covers this due to `InMemoryBaseTable` is updated to use `InMemoryBatchScanWithOrdering` (a new inner classimplementing `SupportsReportOrdering`) when a table ordering is configured. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Sonnet 4.6 --- .../connector/catalog/InMemoryBaseTable.scala | 24 +++++++++++++++++-- .../v2/V2ScanPartitioningAndOrdering.scala | 3 ++- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala index e7762565f47e..9cc32efedf82 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala @@ -429,8 +429,15 @@ abstract class InMemoryBaseTable( private var _pushedFilters: Array[Filter] = Array.empty override def build: Scan = { - val scan = InMemoryBatchScan( - data.map(_.asInstanceOf[InputPartition]).toImmutableArraySeq, schema, tableSchema, options) + val scan = if (InMemoryBaseTable.this.ordering.nonEmpty) { + new InMemoryBatchScanWithOrdering( + data.map(_.asInstanceOf[InputPartition]).toImmutableArraySeq, schema, tableSchema, + options) + } else { + InMemoryBatchScan( + data.map(_.asInstanceOf[InputPartition]).toImmutableArraySeq, schema, tableSchema, + options) + } if (evaluableFilters.nonEmpty) { scan.filter(evaluableFilters) } @@ -596,6 +603,19 @@ abstract class InMemoryBaseTable( } } + // Extends InMemoryBatchScan with SupportsReportOrdering. Only instantiated when the table has a + // non-empty ordering, so that V2ScanPartitioningAndOrdering only sets ordering = Some(...) on the + // logical plan when there is actual ordering to report. + private class InMemoryBatchScanWithOrdering( + data: Seq[InputPartition], + readSchema: StructType, + tableSchema: StructType, + options: CaseInsensitiveStringMap) + extends InMemoryBatchScan(data, readSchema, tableSchema, options) + with SupportsReportOrdering { + override def outputOrdering(): Array[SortOrder] = InMemoryBaseTable.this.ordering + } + abstract class InMemoryWriterBuilder(val info: LogicalWriteInfo) extends SupportsTruncate with SupportsDynamicOverwrite with SupportsStreamingUpdateAsAppend { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala index 5d06c8786d89..7f51875b971f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala @@ -69,7 +69,8 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging { private def ordering(plan: LogicalPlan) = plan.transformDown { case d @ DataSourceV2ScanRelation(relation, scan: SupportsReportOrdering, _, _, _) => - val ordering = V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation) + val ordering = + V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation, relation.funCatalog) d.copy(ordering = Some(ordering)) } } From 50cb940e9fe5ecdfcd54ba2cd84392394c4ea021 Mon Sep 17 00:00:00 2001 From: Peter Toth Date: Wed, 1 Apr 2026 19:48:35 +0200 Subject: [PATCH 2/4] add test --- .../WriteDistributionAndOrderingSuite.scala | 25 +++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala index 588490e07dfd..0207b5a424a1 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala @@ -22,7 +22,7 @@ import java.sql.Date import java.util.Collections import org.apache.spark.sql.{catalyst, AnalysisException, DataFrame, Row} -import org.apache.spark.sql.catalyst.expressions.{ApplyFunctionExpression, Cast, Literal} +import org.apache.spark.sql.catalyst.expressions.{ApplyFunctionExpression, Cast, Literal, TransformExpression} import org.apache.spark.sql.catalyst.expressions.objects.Invoke import org.apache.spark.sql.catalyst.plans.physical import org.apache.spark.sql.catalyst.plans.physical.{CoalescedBoundary, CoalescedHashPartitioning, HashPartitioning, RangePartitioning, UnknownPartitioning} @@ -30,9 +30,10 @@ import org.apache.spark.sql.connector.catalog.{Column, Identifier} import org.apache.spark.sql.connector.catalog.functions._ import org.apache.spark.sql.connector.distributions.{Distribution, Distributions} import org.apache.spark.sql.connector.expressions._ -import org.apache.spark.sql.connector.expressions.LogicalExpressions._ +import org.apache.spark.sql.connector.expressions.Expressions._ import org.apache.spark.sql.execution.{QueryExecution, SortExec, SparkPlan} import org.apache.spark.sql.execution.adaptive.AQEShuffleReadExec +import org.apache.spark.sql.execution.datasources.v2.BatchScanExec import org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike import org.apache.spark.sql.execution.streaming.runtime.MemoryStream @@ -1532,4 +1533,24 @@ class WriteDistributionAndOrderingSuite extends DistributionAndOrderingSuiteBase Seq(None) } } + + test("SPARK-56321: Scan with SupportsReportOrdering and function-based sort resolves correctly") { + val bucketById = bucket(4, "id") + val tableOrdering = Array(sort(bucketById, SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)) + catalog.createTable(ident, columns, Array(bucketById), emptyProps, + Distributions.unspecified(), tableOrdering, None, None) + + sql(s"INSERT INTO testcat.ns1.test_table VALUES (1, 'a', date '2021-01-01')") + + val df = sql("SELECT id, data FROM testcat.ns1.test_table") + // Without the fix, this would throw AnalysisException during planning because the bucket + // transform in the reported ordering could not be resolved without funCatalog. + val scans = collect(df.queryExecution.executedPlan) { case s: BatchScanExec => s } + assert(scans.size === 1) + val ordering = scans.head.outputOrdering + assert(ordering.nonEmpty, + "scan should report non-empty outputOrdering via SupportsReportOrdering") + assert(ordering.head.child.isInstanceOf[TransformExpression], + "bucket-based sort order should resolve to a TransformExpression") + } } From 0ecbf0dc40bc3b077c8df0c7c260aadbfda7aba2 Mon Sep 17 00:00:00 2001 From: Peter Toth Date: Wed, 1 Apr 2026 20:01:49 +0200 Subject: [PATCH 3/4] remove comments --- .../apache/spark/sql/connector/catalog/InMemoryBaseTable.scala | 3 --- .../sql/connector/WriteDistributionAndOrderingSuite.scala | 2 -- 2 files changed, 5 deletions(-) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala index 9cc32efedf82..1fb167dda6e4 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala @@ -603,9 +603,6 @@ abstract class InMemoryBaseTable( } } - // Extends InMemoryBatchScan with SupportsReportOrdering. Only instantiated when the table has a - // non-empty ordering, so that V2ScanPartitioningAndOrdering only sets ordering = Some(...) on the - // logical plan when there is actual ordering to report. private class InMemoryBatchScanWithOrdering( data: Seq[InputPartition], readSchema: StructType, diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala index 0207b5a424a1..7c8ca8888f5d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala @@ -1543,8 +1543,6 @@ class WriteDistributionAndOrderingSuite extends DistributionAndOrderingSuiteBase sql(s"INSERT INTO testcat.ns1.test_table VALUES (1, 'a', date '2021-01-01')") val df = sql("SELECT id, data FROM testcat.ns1.test_table") - // Without the fix, this would throw AnalysisException during planning because the bucket - // transform in the reported ordering could not be resolved without funCatalog. val scans = collect(df.queryExecution.executedPlan) { case s: BatchScanExec => s } assert(scans.size === 1) val ordering = scans.head.outputOrdering From d26b3158bf3fe8bca859293f3a6346e1ecca40b7 Mon Sep 17 00:00:00 2001 From: Peter Toth Date: Thu, 2 Apr 2026 08:32:49 +0200 Subject: [PATCH 4/4] [SPARK-56321][SQL] Pass function catalog to toCatalystOrdering in V2ScanPartitioningAndOrdering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### What changes were proposed in this pull request? `V2ScanPartitioningAndOrdering.ordering` was calling `V2ExpressionUtils.toCatalystOrdering` without the `funCatalog` argument. This meant that function-based sort expressions reported by a data source via `SupportsReportOrdering` (e.g. `bucket(n, col)`) could not be resolved against the function catalog and caused an `AnalysisException` during query planning. The fix passes `relation.funCatalog` as the third argument, consistent with how `toCatalystOpt` is already called in the `partitioning` rule of the same object. `InMemoryBaseTable` is also updated to use a new `InMemoryBatchScanWithOrdering` inner class (implementing `SupportsReportOrdering`) when the table is created with a non-empty ordering. This makes the test infrastructure correctly exercise the `SupportsReportOrdering` code path. ### Why are the changes needed? Without the function catalog, sort orders involving catalog functions reported by `SupportsReportOrdering` cannot be resolved, causing an `AnalysisException` during query planning instead of correctly recognizing the reported ordering. ### Does this PR introduce _any_ user-facing change? Yes. Data sources implementing `SupportsReportOrdering` with function-based sort expressions (e.g. `bucket`) that require the function catalog will now have those sort orders correctly recognized by Spark instead of throwing an `AnalysisException`. ### How was this patch tested? A new test `SPARK-56321: scan with SupportsReportOrdering and function-based sort resolves correctly` is added to `WriteDistributionAndOrderingSuite`. It creates a table with `bucket(4, "id")` partitioning and ordering, queries it, and asserts that the scan reports a non-empty `outputOrdering` with a `TransformExpression` — verifying the bucket transform was resolved correctly via the function catalog. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Sonnet 4.6 --- .../spark/sql/connector/WriteDistributionAndOrderingSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala index 7c8ca8888f5d..ec1b34b8c210 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala @@ -1534,7 +1534,7 @@ class WriteDistributionAndOrderingSuite extends DistributionAndOrderingSuiteBase } } - test("SPARK-56321: Scan with SupportsReportOrdering and function-based sort resolves correctly") { + test("SPARK-56321: Scan with SupportsReportOrdering and function-based sort order") { val bucketById = bucket(4, "id") val tableOrdering = Array(sort(bucketById, SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)) catalog.createTable(ident, columns, Array(bucketById), emptyProps,