cloud-fan commented on code in PR #57487:
URL: https://github.com/apache/spark/pull/57487#discussion_r3670460884
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportStatistics.java:
##########
@@ -36,4 +38,36 @@ public interface SupportsReportStatistics extends Scan {
* Returns the estimated statistics of this data source scan.
*/
Statistics estimateStatistics();
+
+ /**
+ * Returns the estimated size in bytes of this scan without computing full
statistics.
+ * <p>
+ * When cost-based optimization or plan statistics are disabled, Spark
primarily needs the scan's
+ * size in bytes (for example, for broadcast-join thresholding). This method
lets connectors serve
+ * that size estimate cheaply and avoid computing the full statistics. The
default implementation
+ * delegates to {@link #estimateStatistics()} and returns its {@code
sizeInBytes()}, so connectors
+ * that already compute statistics cheaply do not need to override this
method.
+ *
+ * @since 4.3.0
+ */
+ default OptionalLong estimateSizeInBytes() {
+ Statistics statistics = estimateStatistics();
Review Comment:
Avoid invoking the expensive fallback twice. When statistics contain a row
count but no size, this default calls `estimateStatistics()` once, then
`computeSizeInBytes` calls it again to recover the row count. Returning
`OptionalLong.empty()` by default lets the existing fallback inspect full
statistics once, while connectors with a genuinely cheap size can override this
method.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportStatistics.java:
##########
@@ -36,4 +38,36 @@ public interface SupportsReportStatistics extends Scan {
* Returns the estimated statistics of this data source scan.
*/
Statistics estimateStatistics();
+
+ /**
+ * Returns the estimated size in bytes of this scan without computing full
statistics.
+ * <p>
+ * When cost-based optimization or plan statistics are disabled, Spark
primarily needs the scan's
+ * size in bytes (for example, for broadcast-join thresholding). This method
lets connectors serve
+ * that size estimate cheaply and avoid computing the full statistics. The
default implementation
+ * delegates to {@link #estimateStatistics()} and returns its {@code
sizeInBytes()}, so connectors
+ * that already compute statistics cheaply do not need to override this
method.
+ *
+ * @since 4.3.0
+ */
+ default OptionalLong estimateSizeInBytes() {
+ Statistics statistics = estimateStatistics();
+ return statistics != null ? statistics.sizeInBytes() :
OptionalLong.empty();
+ }
+
+ /**
+ * Returns whether the statistics reported by this scan already reflect all
filters that were
+ * fully pushed down to the data source.
+ * <p>
+ * When {@code true} (the default), the reported statistics describe exactly
the data the scan
+ * will produce. When {@code false}, they do <em>not</em> account for the
fully pushed filters
+ * (for example, they describe the whole table), so Spark may use those
fully pushed filters to
Review Comment:
Document the required-column pruning limitation here. Spark silently drops
an adjustment predicate when one of its references is absent from
`Scan.readSchema()`, so a connector returning `false` must retain pushed-filter
columns during `pruneColumns` for this feature to work. The test connector does
that explicitly, but the public contract currently gives implementors no
indication that adjustment is best-effort or depends on
`SupportsPushDownRequiredColumns`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/V2StatisticsUtils.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.internal.connector
+
+import java.util.OptionalLong
+
+import org.apache.spark.sql.connector.read.{Scan, Statistics,
SupportsReportStatistics}
+
+object V2StatisticsUtils {
+
+ def isNotEmpty(stats: Statistics): Boolean = {
+ stats != null && hasAnyValue(stats)
+ }
+
+ private def hasAnyValue(stats: Statistics): Boolean = {
+ stats.sizeInBytes().isPresent ||
+ stats.numRows().isPresent ||
+ (stats.columnStats() != null && !stats.columnStats().isEmpty)
Review Comment:
Keep the null tolerance introduced here through conversion. If the same
statistics object also reports `numRows` or `sizeInBytes`, `isNotEmpty` returns
true and `transformV2Stats` then calls `columnStats().isEmpty`, so the null map
still causes an NPE. Please normalize null to an empty map before conversion
and add a mixed row-count case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
