This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 636643eb8f [core][spark] Use format partition row counts in scan
statistics (#9351)
636643eb8f is described below
commit 636643eb8f16b9ca698f22f5289e132a9886ae36
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Aug 22 17:29:07 2026 +0800
[core][spark] Use format partition row counts in scan statistics (#9351)
---
.../table/format/CatalogSplitEnumerator.java | 65 ++++++++--
.../paimon/table/format/FormatTableScan.java | 39 ++++--
.../paimon/table/format/SplitEnumerator.java | 38 ++++++
.../format/CatalogManagedPartitionScanTest.java | 23 ++++
.../paimon/spark/PaimonFormatTableScan.scala | 26 ++--
.../paimon/spark/read/PaimonStatistics.scala | 7 +-
.../sql/CatalogManagedPartitionAnalyzeTest.scala | 132 +++++++++++++++++++++
7 files changed, 306 insertions(+), 24 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
index eaa43e3d69..bbfad3e836 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
@@ -26,6 +26,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.source.Split;
@@ -49,6 +50,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
@@ -88,8 +90,25 @@ final class CatalogSplitEnumerator extends SplitEnumerator {
@Override
List<Split> enumeratePartitions(@Nullable PartitionPredicate
partitionFilter)
throws IOException {
+ return enumeratePartitions(findCatalogPartitions(partitionFilter),
partitionFilter);
+ }
+
+ @Override
+ ScanPlan plan(@Nullable PartitionPredicate partitionFilter) throws
IOException {
+ if (table.partitionKeys().isEmpty()) {
+ return super.plan(partitionFilter);
+ }
+ List<Partition> partitions = findCatalogPartitions(partitionFilter);
+ List<PartitionEntry> entries = toPartitionEntries(partitions,
partitionFilter);
+ return new ScanPlan(enumeratePartitions(partitions, partitionFilter),
rowCount(entries));
+ }
+
+ private List<Split> enumeratePartitions(
+ List<Partition> catalogPartitions, @Nullable PartitionPredicate
partitionFilter)
+ throws IOException {
List<Pair<LinkedHashMap<String, String>, Path>> partitions =
- findPartitions(partitionFilter);
+ toSpecsAndPaths(
+ catalogPartitions,
coreOptions.formatTablePartitionOnlyValueInPath());
List<Split> splits = new ArrayList<>();
if (partitions.isEmpty()) {
return splits;
@@ -128,6 +147,12 @@ final class CatalogSplitEnumerator extends SplitEnumerator
{
@Override
List<Pair<LinkedHashMap<String, String>, Path>> findPartitions(
@Nullable PartitionPredicate partitionFilter) {
+ return toSpecsAndPaths(
+ findCatalogPartitions(partitionFilter),
+ coreOptions.formatTablePartitionOnlyValueInPath());
+ }
+
+ private List<Partition> findCatalogPartitions(@Nullable PartitionPredicate
partitionFilter) {
Optional<Predicate> extracted =
FormatTableScan.extractPartitionPredicate(partitionFilter);
Map<String, String> prefix = leadingEqualityPrefix(extracted);
Predicate catalogFilter = extracted.orElse(null);
@@ -135,15 +160,21 @@ final class CatalogSplitEnumerator extends
SplitEnumerator {
if (partitions.isEmpty() && prefix.isEmpty() && catalogFilter == null)
{
warnIfFilesystemPartitionsExist();
}
- return toSpecsAndPaths(partitions,
coreOptions.formatTablePartitionOnlyValueInPath());
+ return partitions;
}
@Override
List<PartitionEntry> listPartitionEntries() {
- List<Partition> partitions =
partitionManager.listPartitions(Collections.emptyMap(), null);
- if (partitions.isEmpty()) {
- warnIfFilesystemPartitionsExist();
- }
+ return listPartitionEntries(null);
+ }
+
+ @Override
+ List<PartitionEntry> listPartitionEntries(@Nullable PartitionPredicate
partitionFilter) {
+ return toPartitionEntries(findCatalogPartitions(partitionFilter),
partitionFilter);
+ }
+
+ private List<PartitionEntry> toPartitionEntries(
+ List<Partition> partitions, @Nullable PartitionPredicate
partitionFilter) {
boolean onlyValueInPath =
coreOptions.formatTablePartitionOnlyValueInPath();
List<PartitionEntry> entries = new ArrayList<>(partitions.size());
Set<Map<String, String>> seen = new HashSet<>(partitions.size());
@@ -151,9 +182,14 @@ final class CatalogSplitEnumerator extends SplitEnumerator
{
if (!seen.add(partition.spec())) {
continue;
}
+ BinaryRow partitionRow =
+ toPartitionRow(normalizeSpec(partition.spec(),
onlyValueInPath));
+ if (partitionFilter != null &&
!partitionFilter.test(partitionRow)) {
+ continue;
+ }
entries.add(
new PartitionEntry(
- toPartitionRow(normalizeSpec(partition.spec(),
onlyValueInPath)),
+ partitionRow,
partition.recordCount(),
partition.fileSizeInBytes(),
partition.fileCount(),
@@ -163,6 +199,21 @@ final class CatalogSplitEnumerator extends SplitEnumerator
{
return entries;
}
+ private OptionalLong rowCount(List<PartitionEntry> entries) {
+ long rowCount = 0L;
+ for (PartitionEntry entry : entries) {
+ if (!PartitionStatistics.isKnown(entry.recordCount())) {
+ return OptionalLong.empty();
+ }
+ try {
+ rowCount = Math.addExact(rowCount, entry.recordCount());
+ } catch (ArithmeticException e) {
+ return OptionalLong.empty();
+ }
+ }
+ return OptionalLong.of(rowCount);
+ }
+
private List<Pair<LinkedHashMap<String, String>, Path>> toSpecsAndPaths(
List<Partition> partitions, boolean onlyValueInPath) {
List<Pair<LinkedHashMap<String, String>, Path>> result = new
ArrayList<>(partitions.size());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
index 3f059a0e59..127535773a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
@@ -58,11 +58,19 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
/** {@link TableScan} for {@link FormatTable}. */
public class FormatTableScan implements InnerTableScan {
+ /** A format-table scan plan with statistics collected during split
planning. */
+ public interface Plan extends TableScan.Plan {
+
+ /** Returns the row count of the planned partitions, or empty if any
count is unknown. */
+ OptionalLong rowCount();
+ }
+
final FormatTable table;
final CoreOptions coreOptions;
@Nullable private PartitionPredicate partitionFilter;
@@ -93,7 +101,7 @@ public class FormatTableScan implements InnerTableScan {
@Override
public List<PartitionEntry> listPartitionEntries() {
- return splitEnumerator.listPartitionEntries();
+ return splitEnumerator.listPartitionEntries(partitionFilter);
}
@Override
@@ -204,19 +212,34 @@ public class FormatTableScan implements InnerTableScan {
}
private class FormatTableScanPlan implements Plan {
+
+ @Nullable private SplitEnumerator.ScanPlan scanPlan;
+
@Override
public List<Split> splits() {
- List<Split> splits = new ArrayList<>();
+ List<Split> splits = new ArrayList<>(scanPlan().splits());
+ // Keep all splits for a positive limit because FormatDataSplit
has no row count.
+ if (limit != null && limit <= 0) {
+ return new ArrayList<>();
+ }
+ return splits;
+ }
+
+ @Override
+ public OptionalLong rowCount() {
+ return scanPlan().rowCount();
+ }
+
+ private synchronized SplitEnumerator.ScanPlan scanPlan() {
+ if (scanPlan != null) {
+ return scanPlan;
+ }
try {
- splits.addAll(splitEnumerator.enumerate(partitionFilter));
- // Keep all splits for a positive limit because
FormatDataSplit has no row count.
- if (limit != null && limit <= 0) {
- return new ArrayList<>();
- }
+ scanPlan = splitEnumerator.plan(partitionFilter);
} catch (IOException e) {
throw new RuntimeException("Failed to scan files", e);
}
- return splits;
+ return scanPlan;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
index 64c799dddf..c46919576d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
@@ -44,6 +44,7 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.OptionalLong;
import static
org.apache.paimon.format.text.HadoopCompressionUtils.isCompressed;
import static org.apache.paimon.format.text.TextLineReader.isDefaultDelimiter;
@@ -89,6 +90,10 @@ abstract class SplitEnumerator {
return enumeratePartitions(partitionFilter);
}
+ ScanPlan plan(@Nullable PartitionPredicate partitionFilter) throws
IOException {
+ return new ScanPlan(enumerate(partitionFilter), OptionalLong.empty());
+ }
+
/** Enumerate splits for a partitioned table; partitions come from the
concrete source. */
abstract List<Split> enumeratePartitions(@Nullable PartitionPredicate
partitionFilter)
throws IOException;
@@ -98,6 +103,39 @@ abstract class SplitEnumerator {
abstract List<PartitionEntry> listPartitionEntries();
+ List<PartitionEntry> listPartitionEntries(@Nullable PartitionPredicate
partitionFilter) {
+ List<PartitionEntry> entries = listPartitionEntries();
+ if (partitionFilter == null) {
+ return entries;
+ }
+ List<PartitionEntry> filtered = new ArrayList<>();
+ for (PartitionEntry entry : entries) {
+ if (partitionFilter.test(entry.partition())) {
+ filtered.add(entry);
+ }
+ }
+ return filtered;
+ }
+
+ static final class ScanPlan {
+
+ private final List<Split> splits;
+ private final OptionalLong rowCount;
+
+ ScanPlan(List<Split> splits, OptionalLong rowCount) {
+ this.splits = splits;
+ this.rowCount = rowCount;
+ }
+
+ List<Split> splits() {
+ return splits;
+ }
+
+ OptionalLong rowCount() {
+ return rowCount;
+ }
+ }
+
BinaryRow toPartitionRow(LinkedHashMap<String, String> partitionSpec) {
RowType partitionType = table.partitionType();
GenericRow row =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
index 56b7a18493..5ce899694d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
@@ -54,6 +54,7 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.OptionalLong;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -183,6 +184,28 @@ class CatalogManagedPartitionScanTest {
assertThat(fileIO.listedPaths).isEmpty();
}
+ @Test
+ void testPlanReusesCatalogListingForSplitsAndRowCount() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition october =
+ new Partition(partition("2025", "10").spec(), 2L, 0L, 0L, 0L,
-1, false);
+ Partition november =
+ new Partition(partition("2025", "11").spec(), 3L, 0L, 0L, 0L,
-1, false);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(),
isNull()))
+ .thenReturn(new PagedList<>(Arrays.asList(october, november),
null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ writeDataFile(fileIO, tablePath, "year=2025/month=10");
+ writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ FormatTable table = createTable(fileIO, tablePath,
partitionManager(catalog), false);
+
+ FormatTableScan.Plan plan = new FormatTableScan(table, null,
null).plan();
+
+ assertThat(plan.splits()).hasSize(2);
+ assertThat(plan.rowCount()).isEqualTo(OptionalLong.of(5L));
+ verify(catalog).listPartitionsPaged(IDENTIFIER, 1000, null, null);
+ }
+
@Test
void testUnderscoreInPartitionNameRemainsLiteralPrefix() {
LocalFileIO fileIO = LocalFileIO.create();
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonFormatTableScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonFormatTableScan.scala
index 81c3614833..913bcb66cd 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonFormatTableScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonFormatTableScan.scala
@@ -20,12 +20,16 @@ package org.apache.paimon.spark
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.predicate.Predicate
-import org.apache.paimon.spark.read.{BaseScan, PaimonSupportsRuntimeFiltering}
+import org.apache.paimon.spark.read.{BaseScan, PaimonStatistics,
PaimonSupportsRuntimeFiltering}
import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.format.FormatTableScan
import org.apache.paimon.table.source.Split
+import org.apache.spark.sql.connector.read.Statistics
import org.apache.spark.sql.types.StructType
+import java.util.OptionalLong
+
import scala.collection.JavaConverters._
/** Scan implementation for [[FormatTable]] */
@@ -38,12 +42,20 @@ case class PaimonFormatTableScan(
extends BaseScan
with PaimonSupportsRuntimeFiltering {
+ @volatile private var plannedRowCount: OptionalLong = OptionalLong.empty()
+
protected def getInputSplits: Array[Split] = {
- readBuilder
- .newScan()
- .plan()
- .splits()
- .asScala
- .toArray
+ val plan = readBuilder.newScan().plan()
+ val splits = plan.splits().asScala.toArray
+ plannedRowCount = plan match {
+ case formatPlan: FormatTableScan.Plan => formatPlan.rowCount()
+ case _ => OptionalLong.empty()
+ }
+ splits
+ }
+
+ override def estimateStatistics: Statistics = {
+ val splits = inputSplits
+ PaimonStatistics(splits, readTableRowType, table.rowType(),
table.statistics(), plannedRowCount)
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala
index d0b99bcbbf..9b06223aee 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala
@@ -39,11 +39,14 @@ case class PaimonStatistics(
splits: Array[Split],
readRowType: RowType,
tableRowType: RowType,
- paimonStats: Optional[stats.Statistics]
+ paimonStats: Optional[stats.Statistics],
+ scanRowCount: OptionalLong = OptionalLong.empty()
) extends Statistics {
lazy val numRows: OptionalLong = {
- if (splits.exists(_.rowCount() == -1)) {
+ if (scanRowCount.isPresent) {
+ scanRowCount
+ } else if (splits.exists(_.rowCount() == -1)) {
OptionalLong.empty()
} else {
OptionalLong.of(splits.map(_.rowCount()).sum)
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
index 89d57134f8..cd3193ab42 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
@@ -25,6 +25,8 @@ import
org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
import org.apache.paimon.table.FormatTable
import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionException
+import org.apache.spark.sql.catalyst.optimizer.BuildRight
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec,
SortMergeJoinExec}
import java.util.Locale
@@ -336,6 +338,136 @@ class CatalogManagedPartitionAnalyzeTest extends
PaimonSparkTestWithRestCatalogB
}
}
+ test("catalog partition row counts feed scan statistics after partition
pruning") {
+ val tableName = "analyze_scan_statistics"
+ withTable(tableName) {
+ createTable(tableName)
+ sql(s"""INSERT INTO ${qualified(tableName)} VALUES
+ |(1, 'a', '20260101', '00'),
+ |(2, 'b', '20260101', '00'),
+ |(3, 'c', '20260102', '00')
+ |""".stripMargin)
+
+ val all = getFormatTableScan(s"SELECT * FROM ${qualified(tableName)}")
+ assert(all.estimateStatistics.numRows().getAsLong == 3L)
+
+ val pruned =
+ getFormatTableScan(s"SELECT * FROM ${qualified(tableName)} WHERE dt =
'20260101'")
+ assert(pruned.estimateStatistics.numRows().getAsLong == 2L)
+ }
+ }
+
+ test("one unknown selected partition makes scan row count unknown") {
+ val tableName = "analyze_partial_scan_statistics"
+ withTable(tableName) {
+ createTable(tableName)
+ sql(
+ s"INSERT INTO ${qualified(tableName)} VALUES " +
+ s"(1, 'known', '20260101', '00')")
+ writeCsvPartition(tableName, "20260102", "00", 2)
+ repair(tableName)
+
+ val known = getFormatTableScan(s"SELECT * FROM ${qualified(tableName)}
WHERE dt = '20260101'")
+ assert(known.estimateStatistics.numRows().getAsLong == 1L)
+
+ val partiallyUnknown =
+ getFormatTableScan(s"SELECT * FROM
${qualified(tableName)}").estimateStatistics
+ assert(!partiallyUnknown.numRows().isPresent)
+ }
+ }
+
+ test("partition row count is not duplicated across format data splits") {
+ val tableName = "analyze_multi_split_statistics"
+ withTable(tableName) {
+ sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour
STRING)
+ |USING CSV
+ |PARTITIONED BY (dt, hour)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true',
+ | 'source.split.target-size' = '1 B')
+ |""".stripMargin)
+ sql(s"""INSERT INTO ${qualified(tableName)} VALUES
+ |(1, 'one', '20260101', '00'),
+ |(2, 'two', '20260101', '00'),
+ |(3, 'three', '20260101', '00')
+ |""".stripMargin)
+
+ val scan = getFormatTableScan(s"SELECT * FROM ${qualified(tableName)}")
+ assert(scan.inputSplits.length > 1)
+ assert(scan.inputSplits.forall(_.rowCount() == -1L))
+ assert(scan.estimateStatistics.numRows().getAsLong == 3L)
+ }
+ }
+
+ test("TPC-DS-style dimension join uses partition row count to avoid
fact-side shuffle") {
+ val tableName = "date_dim_format"
+ withSparkSQLConf(
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.cbo.enabled" -> "true",
+ "spark.sql.autoBroadcastJoinThreshold" -> "128",
+ "spark.sql.join.preferSortMergeJoin" -> "true"
+ ) {
+ withTable(tableName) {
+ sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING,
hour STRING)
+ |USING PARQUET
+ |PARTITIONED BY (dt, hour)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+
+ // Keep the physical file above the broadcast threshold while the
projected dimension row
+ // is tiny. The first partition is written normally, then copied as if
an external writer
+ // had added a new date partition without reporting its row count.
+ val sparkSession = spark
+ import sparkSession.implicits._
+ val payload = new scala.util.Random(42L).alphanumeric.take(256 *
1024).mkString
+ withTempView("date_dim_rows") {
+ Seq((1, payload, "20260101", "00"))
+ .toDF("id", "payload", "dt", "hour")
+ .createOrReplaceTempView("date_dim_rows")
+ sql(s"INSERT INTO ${qualified(tableName)} SELECT * FROM
date_dim_rows")
+ }
+ copyPartitionFiles(tableName, "20260101", "20260102")
+ repair(tableName)
+
+ val query =
+ s"""SELECT store_sales.id
+ |FROM range(0, 1000000) store_sales
+ |JOIN ${qualified(tableName)} date_dim
+ | ON store_sales.id = date_dim.id
+ |WHERE date_dim.dt = '20260102' AND date_dim.hour = '00'
+ |""".stripMargin
+
+ val beforeStats = getFormatTableScan(query).estimateStatistics
+ assert(!beforeStats.numRows().isPresent)
+ val beforePlan = sql(query).queryExecution.executedPlan
+ assert(
+ beforePlan.collectFirst { case join: SortMergeJoinExec => join
}.isDefined,
+ beforePlan)
+ assert(
+ beforePlan.collectFirst { case join: BroadcastHashJoinExec => join
}.isEmpty,
+ beforePlan)
+
+ sql(
+ s"ANALYZE TABLE ${qualified(tableName)} " +
+ s"PARTITION (dt = '20260102', hour = '00') COMPUTE
STATISTICS").collect()
+
+ val afterStats = getFormatTableScan(query).estimateStatistics
+ assert(afterStats.numRows().getAsLong == 1L)
+ val afterQuery = sql(query)
+ val afterPlan = afterQuery.queryExecution.executedPlan
+ val broadcastJoin = afterPlan
+ .collectFirst { case join: BroadcastHashJoinExec => join }
+ .getOrElse(fail(afterPlan.toString))
+ .asInstanceOf[BroadcastHashJoinExec]
+ assert(broadcastJoin.buildSide == BuildRight, afterPlan)
+ checkAnswer(afterQuery, Seq(org.apache.spark.sql.Row(1L)))
+ }
+ }
+ }
+
test("ANALYZE run twice reports the same measurement rather than
accumulating") {
val tableName = "analyze_idempotent"
withTable(tableName) {
