You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sz...@apache.org on 2026/09/23 01:43:09 UTC
(spark) branch branch-4.x updated: [SPARK-59377][SQL] DSv2 column statistics are silently dropped when the column name needs quoting or differs only in case
This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 06d24b4a1401 [SPARK-59377][SQL] DSv2 column statistics are silently dropped when the column name needs quoting or differs only in case
06d24b4a1401 is described below
commit 06d24b4a1401e2ef49a405cdf5b5822c53e0da83
Author: hemanthboyina <he...@apache.org>
AuthorDate: Tue Sep 22 18:42:46 2026 -0700
[SPARK-59377][SQL] DSv2 column statistics are silently dropped when the column name needs quoting or differs only in case
**What changes were proposed in this pull request?**
DataSourceV2Relation.transformV2Stats matched connector-reported column statistics to output attributes via
`attribute.name.equals(key.describe())`. This PR matches on the NamedReference's raw field name (key.fieldNames)
compared through SQLConf.get.resolver, restricted to single-part references, and no longer compares against describe().
**Why are the changes needed?**
NamedReference.describe() returns a quoted display string (QuotingUtils.quoteIfNeeded). For a top-level column whose name is not a plain identifier (e.g. col-1, 1col, or names with dots/spaces), describe() yields col-1, which never equals the attribute name col-1, so the column statistic is silently dropped. The comparison is [...]
regardless of `spark.sql.caseSensitive`, so a stat keyed ID is dropped for an attribute `id` under the default case-insensitive resolution . The lost stats degrade cost-based optimizer estimates with no error surfaced.
**Does this PR introduce _any_ user-facing change?**
No public API change
**How was this patch tested?**
Added unit tests to DataSourceV2RelationSuite: one for a column name needing quoting (col-1), and one asserting case-sensitivity is honored (ID→id matches under case-insensitive resolution
**Was this patch authored or co-authored using generative AI tooling?**
No
Closes #58666 from hemanthboyina/dsv2-colstat-name-match.
Authored-by: hemanthboyina <he...@apache.org>
Signed-off-by: Szehon Ho <sz...@gmail.com>
(cherry picked from commit b361b2df36edfed041c222325e2cd98a2803c522)
Signed-off-by: Szehon Ho <sz...@gmail.com>
---
.../datasources/v2/DataSourceV2Relation.scala | 35 ++-
.../datasources/v2/DataSourceV2RelationSuite.scala | 292 ++++++++++++---------
2 files changed, 202 insertions(+), 125 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 0cce60d42f3a..1978af305636 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -22,7 +22,7 @@ import java.util.{Collections, Optional, OptionalLong}
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, NamedRelation, TimeTravelSpec}
import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat, CatalogStatistics}
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, AttributeSet, Expression, SortOrder, V2ExpressionUtils}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, AttributeSeq, AttributeSet, Expression, SortOrder, V2ExpressionUtils}
import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LogicalPlan, Statistics}
import org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils
@@ -37,6 +37,7 @@ import org.apache.spark.sql.connector.read.{Scan, Statistics => V2Statistics, Su
import org.apache.spark.sql.connector.read.colstats.{ColumnStatistics, Histogram => V2Histogram, HistogramBin => V2HistogramBin}
import org.apache.spark.sql.connector.read.streaming.{Offset, SparkDataStream}
import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.connector.{SupportsRuntimeCatalystFiltering, V2StatisticsUtils}
import org.apache.spark.sql.types.{DataType, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -573,12 +574,14 @@ object DataSourceV2Relation {
}
var colStats: Seq[(Attribute, ColumnStat)] = Seq.empty[(Attribute, ColumnStat)]
+ val resolver = SQLConf.get.resolver
// columnStats() may be null even when numRows/sizeInBytes are present, so normalize it to an
// empty map before conversion to avoid an NPE.
val v2ColumnStats = Option(v2Statistics.columnStats()).getOrElse(EMPTY_V2_COLUMN_STATS)
if (!v2ColumnStats.isEmpty) {
val keys = v2ColumnStats.keySet()
-
+ val outputAttrs = AttributeSeq.fromNormalOutput(output)
+ var keyed = Seq.empty[(Attribute, String, ColumnStat)]
keys.forEach(key => {
val colStat = v2ColumnStats.get(key)
val distinct: Option[BigInt] =
@@ -602,12 +605,32 @@ object DataSourceV2Relation {
val catalystColStat = ColumnStat(distinct, min, max, nullCount, avgLen, maxLen, histogram)
- output.foreach(attribute => {
- if (attribute.name.equals(key.describe())) {
- colStats = colStats :+ (attribute -> catalystColStat)
+ // Catalyst column statistics only support top-level attributes. Prefer a unique exact name
+ // when the configured resolver matches multiple output attributes.
+ val fieldNames = key.fieldNames
+ if (fieldNames.length == 1) {
+ val fieldName = fieldNames.head
+ val exprIds = outputAttrs
+ .getCandidatesForResolution(Seq(fieldName), resolver)._1
+ .map(_.exprId).toSet
+ output.filter(attr => exprIds.contains(attr.exprId)) match {
+ case Seq(single) => keyed = keyed :+ ((single, fieldName, catalystColStat))
+ case multiple => multiple.filter(_.name == fieldName) match {
+ case Seq(exact) => keyed = keyed :+ ((exact, fieldName, catalystColStat))
+ case _ =>
+ }
}
- })
+ }
})
+ // Several keys can resolve to one attribute (e.g. "id" and "ID" when case-insensitive); keep
+ // a unique match, preferring an exact-name key, so the result is deterministic.
+ colStats = keyed.groupBy(_._1.exprId).values.toSeq.flatMap {
+ case Seq((attribute, _, stat)) => Some(attribute -> stat)
+ case many => many.filter { case (attr, fieldName, _) => fieldName == attr.name } match {
+ case Seq((attribute, _, stat)) => Some(attribute -> stat)
+ case _ => None
+ }
+ }
}
val attributeStats = AttributeMap(colStats)
// Prefer the source-reported size. Otherwise infer a projection-aware size from the row count
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala
index 79fd057f037e..35bf7611c744 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.execution.datasources.v2
import java.util
-import java.util.OptionalLong
+import java.util.{HashMap, Map => JMap, OptionalLong}
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat, CatalogStatistics}
@@ -132,24 +132,51 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
output)
}
+ private def optLong(value: Option[Long]): OptionalLong =
+ value.map(OptionalLong.of).getOrElse(OptionalLong.empty())
+
+ private def v2ColumnStat(
+ distinct: Option[Long] = None,
+ avg: Option[Long] = None): ColumnStatistics = new ColumnStatistics {
+ override def distinctCount(): OptionalLong = optLong(distinct)
+ override def avgLen(): OptionalLong = optLong(avg)
+ }
+
+ private def v2ColumnStatsMap(
+ entries: (String, ColumnStatistics)*): JMap[NamedReference, ColumnStatistics] = {
+ val map = new HashMap[NamedReference, ColumnStatistics]()
+ entries.foreach { case (name, stat) => map.put(FieldReference.column(name), stat) }
+ map
+ }
+
+ private def v2Statistics(
+ size: OptionalLong = OptionalLong.empty(),
+ rows: OptionalLong = OptionalLong.empty(),
+ colStats: JMap[NamedReference, ColumnStatistics] =
+ new HashMap[NamedReference, ColumnStatistics]()): V2Statistics =
+ new V2Statistics {
+ override def sizeInBytes(): OptionalLong = size
+ override def numRows(): OptionalLong = rows
+ override def columnStats(): JMap[NamedReference, ColumnStatistics] = colStats
+ }
+
+ /** Evaluates stats for each estimateStatistics() call. */
+ private def newStatsScan(
+ stats: => V2Statistics,
+ schema: StructType = StructType(Seq(StructField("id", IntegerType))),
+ sizeEstimate: OptionalLong = OptionalLong.empty()): Scan =
+ new Scan with SupportsReportStatistics {
+ override def readSchema(): StructType = schema
+ override def estimateStatistics(): V2Statistics = stats
+ override def estimateSizeInBytes(): OptionalLong = sizeEstimate
+ }
+
test("DataSourceV2ScanRelation.computeStats uses non-empty scan stats with CBO") {
val idAttr = AttributeReference("id", IntegerType)()
val output = Seq(idAttr)
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.of(42L)
- override def columnStats(): java.util.Map[NamedReference, ColumnStatistics] = {
- val stats = new java.util.HashMap[NamedReference, ColumnStatistics]()
- stats.put(FieldReference.column("id"), new ColumnStatistics {
- override def distinctCount(): OptionalLong = OptionalLong.of(40L)
- override def avgLen(): OptionalLong = OptionalLong.of(4L)
- })
- stats
- }
- }
- }
+ val scan = newStatsScan(v2Statistics(
+ rows = OptionalLong.of(42L),
+ colStats = v2ColumnStatsMap("id" -> v2ColumnStat(distinct = Some(40L), avg = Some(4L)))))
withSQLConf(SQLConf.CBO_ENABLED.key -> "true") {
val stats = scanRel(output, scan).computeStats()
@@ -163,16 +190,119 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
}
}
- test("DataSourceV2ScanRelation.computeStats derives size 1 for a zero-row scan with CBO") {
+ test("DataSourceV2ScanRelation.computeStats matches column stats for names needing quoting") {
+ val colAttr = AttributeReference("col-1", IntegerType)()
+ val output = Seq(colAttr)
+ val scan = newStatsScan(
+ v2Statistics(
+ rows = OptionalLong.of(42L),
+ colStats = v2ColumnStatsMap("col-1" -> v2ColumnStat(distinct = Some(40L)))),
+ schema = StructType(Seq(StructField("col-1", IntegerType))))
+
+ withSQLConf(SQLConf.CBO_ENABLED.key -> "true") {
+ val stats = scanRel(output, scan).computeStats()
+ assert(stats.rowCount.contains(BigInt(42)))
+ assert(stats.attributeStats.size === 1,
+ "column stat keyed by a name needing quoting must be attached to its attribute")
+ assert(stats.attributeStats(colAttr).distinctCount.contains(BigInt(40)))
+ }
+ }
+
+ test("DataSourceV2ScanRelation.computeStats matches column stats respecting case sensitivity") {
val idAttr = AttributeReference("id", IntegerType)()
val output = Seq(idAttr)
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.of(0L)
- }
+ val scan = newStatsScan(v2Statistics(
+ rows = OptionalLong.of(42L),
+ colStats = v2ColumnStatsMap("ID" -> v2ColumnStat(distinct = Some(40L)))))
+
+ withSQLConf(
+ SQLConf.CBO_ENABLED.key -> "true",
+ SQLConf.CASE_SENSITIVE.key -> "false") {
+ val stats = scanRel(output, scan).computeStats()
+ assert(stats.attributeStats.size === 1,
+ "case-insensitive resolution should match 'ID' to attribute 'id'")
+ assert(stats.attributeStats(idAttr).distinctCount.contains(BigInt(40)))
+ }
+
+ withSQLConf(
+ SQLConf.CBO_ENABLED.key -> "true",
+ SQLConf.CASE_SENSITIVE.key -> "true") {
+ val stats = scanRel(output, scan).computeStats()
+ assert(stats.attributeStats.isEmpty,
+ "case-sensitive resolution should not match 'ID' to attribute 'id'")
+ }
+ }
+
+ test("DataSourceV2ScanRelation.computeStats resolves case-conflicting outputs uniquely") {
+ val idAttr = AttributeReference("id", IntegerType)()
+ val upperIdAttr = AttributeReference("ID", IntegerType)()
+ val output = Seq(idAttr, upperIdAttr)
+
+ def scanKeyed(name: String): Scan = newStatsScan(
+ v2Statistics(
+ rows = OptionalLong.of(42L),
+ colStats = v2ColumnStatsMap(name -> v2ColumnStat(distinct = Some(40L)))),
+ schema = StructType(Seq(StructField("id", IntegerType), StructField("ID", IntegerType))))
+
+ withSQLConf(
+ SQLConf.CBO_ENABLED.key -> "true",
+ SQLConf.CASE_SENSITIVE.key -> "false") {
+ val idStats = scanRel(output, scanKeyed("id")).computeStats()
+ assert(idStats.attributeStats.size === 1)
+ assert(idStats.attributeStats(idAttr).distinctCount.contains(BigInt(40)))
+
+ val ambiguousStats = scanRel(output, scanKeyed("Id")).computeStats()
+ assert(ambiguousStats.attributeStats.isEmpty,
+ "an ambiguous case-insensitive match must not be attached to any attribute")
+ }
+ }
+
+ test("DataSourceV2ScanRelation.computeStats prefers the exact-name key when several keys " +
+ "map to one output") {
+ val idAttr = AttributeReference("id", IntegerType)()
+ val output = Seq(idAttr)
+ val scan = newStatsScan(v2Statistics(
+ rows = OptionalLong.of(42L),
+ colStats = v2ColumnStatsMap(
+ "ID" -> v2ColumnStat(distinct = Some(10L)),
+ "id" -> v2ColumnStat(distinct = Some(40L)))))
+
+ withSQLConf(
+ SQLConf.CBO_ENABLED.key -> "true",
+ SQLConf.CASE_SENSITIVE.key -> "false") {
+ val stats = scanRel(output, scan).computeStats()
+ assert(stats.attributeStats.size === 1)
+ assert(stats.attributeStats(idAttr).distinctCount.contains(BigInt(40)),
+ "the exact-name key must win deterministically over a case-folded key")
}
+ }
+
+ test("DataSourceV2ScanRelation.computeStats matches column stats consistently with the " +
+ "analyzer for case-folding-equal names") {
+ // scalastyle:off nonascii
+ val longS = "\u017f"
+ // scalastyle:on nonascii
+ val colAttr = AttributeReference(longS, IntegerType)()
+ val output = Seq(colAttr)
+ val scan = newStatsScan(
+ v2Statistics(
+ rows = OptionalLong.of(42L),
+ colStats = v2ColumnStatsMap("s" -> v2ColumnStat(distinct = Some(40L)))),
+ schema = StructType(Seq(StructField(longS, IntegerType))))
+
+ withSQLConf(
+ SQLConf.CBO_ENABLED.key -> "true",
+ SQLConf.CASE_SENSITIVE.key -> "false") {
+ val stats = scanRel(output, scan).computeStats()
+ assert(stats.attributeStats.isEmpty,
+ "a stat keyed by an ASCII name must not attach to a case-folding-equal non-ASCII column")
+ }
+ }
+
+ test("DataSourceV2ScanRelation.computeStats derives size 1 for a zero-row scan with CBO") {
+ val idAttr = AttributeReference("id", IntegerType)()
+ val output = Seq(idAttr)
+ val scan = newStatsScan(v2Statistics(rows = OptionalLong.of(0L)))
withSQLConf(
SQLConf.CBO_ENABLED.key -> "true",
@@ -187,14 +317,9 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats uses full stats with plan stats enabled") {
val idAttr = AttributeReference("id", IntegerType)()
val output = Seq(idAttr)
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateSizeInBytes(): OptionalLong = OptionalLong.of(50L)
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.of(1000L)
- override def numRows(): OptionalLong = OptionalLong.of(7L)
- }
- }
+ val scan = newStatsScan(
+ v2Statistics(size = OptionalLong.of(1000L), rows = OptionalLong.of(7L)),
+ sizeEstimate = OptionalLong.of(50L))
withSQLConf(
SQLConf.CBO_ENABLED.key -> "false",
@@ -209,20 +334,8 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats treats column-only scan stats as non-empty") {
val idAttr = AttributeReference("id", IntegerType)()
val output = Seq(idAttr)
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.empty()
- override def columnStats(): java.util.Map[NamedReference, ColumnStatistics] = {
- val stats = new java.util.HashMap[NamedReference, ColumnStatistics]()
- stats.put(FieldReference.column("id"), new ColumnStatistics {
- override def distinctCount(): OptionalLong = OptionalLong.of(7L)
- })
- stats
- }
- }
- }
+ val scan = newStatsScan(v2Statistics(
+ colStats = v2ColumnStatsMap("id" -> v2ColumnStat(distinct = Some(7L)))))
withSQLConf(
SQLConf.CBO_ENABLED.key -> "true",
@@ -239,14 +352,9 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats uses size-only estimates without CBO") {
val idAttr = AttributeReference("id", IntegerType)()
val output = Seq(idAttr)
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateSizeInBytes(): OptionalLong = OptionalLong.of(50L)
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.of(1000L)
- override def numRows(): OptionalLong = OptionalLong.of(5L)
- }
- }
+ val scan = newStatsScan(
+ v2Statistics(size = OptionalLong.of(1000L), rows = OptionalLong.of(5L)),
+ sizeEstimate = OptionalLong.of(50L))
withSQLConf(
SQLConf.CBO_ENABLED.key -> "false",
@@ -262,13 +370,7 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats uses default estimateSizeInBytes without CBO") {
val idAttr = AttributeReference("id", IntegerType)()
val output = Seq(idAttr)
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.of(64L)
- override def numRows(): OptionalLong = OptionalLong.of(5L)
- }
- }
+ val scan = newStatsScan(v2Statistics(size = OptionalLong.of(64L), rows = OptionalLong.of(5L)))
withSQLConf(
SQLConf.CBO_ENABLED.key -> "false",
@@ -284,13 +386,7 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats infers size-only estimates from row count") {
val idAttr = AttributeReference("id", IntegerType)()
val output = Seq(idAttr)
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.of(5L)
- }
- }
+ val scan = newStatsScan(v2Statistics(rows = OptionalLong.of(5L)))
withSQLConf(
SQLConf.CBO_ENABLED.key -> "false",
@@ -310,16 +406,10 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
// consult the full statistics exactly once, not once for the (empty) size and again for the row
// count.
var estimateStatisticsCalls = 0
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = {
- estimateStatisticsCalls += 1
- new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.of(5L)
- }
- }
- }
+ val scan = newStatsScan({
+ estimateStatisticsCalls += 1
+ v2Statistics(rows = OptionalLong.of(5L))
+ })
withSQLConf(
SQLConf.CBO_ENABLED.key -> "false",
@@ -334,13 +424,7 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats uses default size without CBO for empty stats") {
val output = Seq(AttributeReference("id", IntegerType)())
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.empty()
- }
- }
+ val scan = newStatsScan(v2Statistics())
withSQLConf(
SQLConf.CBO_ENABLED.key -> "false",
@@ -381,13 +465,7 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats uses default size for empty scan stats") {
val output = Seq(AttributeReference("id", IntegerType)())
- val scan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.empty()
- }
- }
+ val scan = newStatsScan(v2Statistics())
withSQLConf(
SQLConf.CBO_ENABLED.key -> "true",
@@ -401,18 +479,8 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
test("DataSourceV2ScanRelation.computeStats uses default size for null scan stats") {
val output = Seq(AttributeReference("id", IntegerType)())
- val nullStatsScan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = null
- }
- val nullColumnStatsScan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.empty()
- override def columnStats(): java.util.Map[NamedReference, ColumnStatistics] = null
- }
- }
+ val nullStatsScan = newStatsScan(stats = null)
+ val nullColumnStatsScan = newStatsScan(v2Statistics(colStats = null))
withSQLConf(
SQLConf.CBO_ENABLED.key -> "true",
@@ -443,24 +511,10 @@ class DataSourceV2RelationSuite extends SparkFunSuite with SQLHelper {
val output = Seq(idAttr)
// numRows present, columnStats null: isNotEmpty is true (via numRows), so the conversion runs
// and must not NPE on the null column-stats map.
- val rowCountScan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.empty()
- override def numRows(): OptionalLong = OptionalLong.of(42L)
- override def columnStats(): java.util.Map[NamedReference, ColumnStatistics] = null
- }
- }
+ val rowCountScan = newStatsScan(v2Statistics(rows = OptionalLong.of(42L), colStats = null))
// sizeInBytes present, columnStats null: same null-tolerance requirement, reached via
// sizeInBytes instead of numRows.
- val sizeScan = new Scan with SupportsReportStatistics {
- override def readSchema(): StructType = StructType(Seq(StructField("id", IntegerType)))
- override def estimateStatistics(): V2Statistics = new V2Statistics {
- override def sizeInBytes(): OptionalLong = OptionalLong.of(1000L)
- override def numRows(): OptionalLong = OptionalLong.empty()
- override def columnStats(): java.util.Map[NamedReference, ColumnStatistics] = null
- }
- }
+ val sizeScan = newStatsScan(v2Statistics(size = OptionalLong.of(1000L), colStats = null))
withSQLConf(SQLConf.CBO_ENABLED.key -> "true") {
val rowCountStats = scanRel(output, rowCountScan).computeStats()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org