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 d904d00009 [core][flink][spark] Treat non-positive row counts as
unknown (#9456)
d904d00009 is described below
commit d904d000091339b4b030bb581a9e4ebd45785b45
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Sat Aug 29 11:25:03 2026 +0800
[core][flink][spark] Treat non-positive row counts as unknown (#9456)
### Purpose
Follow up on #9379 by making row-count inference conservative across
Paimon table scans.
Paimon split and scan statistics use non-positive row counts for unknown
values, while HMS-compatible catalogs may expose any value less than or
equal to zero when statistics are unavailable. Treating such a value as
an exact row count can make an optimizer plan against an apparently
empty scan that still contains data.
This change:
- treats non-positive catalog partition row counts as unknown in Core
format-table scan plans;
- makes Spark reject non-positive scan and split row counts, while
preserving exact zero for structurally empty scans and positive
scan-level statistics;
- makes Flink return `TableStats.UNKNOWN` for non-positive or overflowed
row-count sums from both partition entries and splits;
- uses checked addition so overflow cannot turn a large table into a
misleading row count.
Empty input collections still produce an exact zero. There is no public
API change.
### Tests
- [x] `CatalogManagedPartitionScanTest` (23 tests)
- [x] `PaimonStatisticsTest` (5 tests)
- [x] `FlinkTableSourceStatisticsTest` (3 tests)
- [x] `CatalogManagedPartitionAnalyzeTest` Spark integration suite (22
tests)
- [x] Spotless apply/check for Core, Flink common, and Spark common
---
.../table/format/CatalogSplitEnumerator.java | 7 +-
.../format/CatalogManagedPartitionScanTest.java | 52 ++++++++++
.../paimon/flink/source/FlinkTableSource.java | 32 +++++-
.../source/FlinkTableSourceStatisticsTest.java | 114 +++++++++++++++++++++
.../paimon/spark/read/PaimonStatistics.scala | 29 ++++--
.../paimon/spark/read/PaimonStatisticsTest.scala | 75 ++++++++++++++
6 files changed, 298 insertions(+), 11 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 bbfad3e836..9095df296f 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,7 +26,6 @@ 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;
@@ -202,7 +201,11 @@ final class CatalogSplitEnumerator extends SplitEnumerator
{
private OptionalLong rowCount(List<PartitionEntry> entries) {
long rowCount = 0L;
for (PartitionEntry entry : entries) {
- if (!PartitionStatistics.isKnown(entry.recordCount())) {
+ // At this scan-estimation boundary, only positive catalog counts
are trustworthy.
+ // Catalogs such as HMS overload non-positive values for
unavailable statistics, so an
+ // entry carrying zero cannot safely prove that its partition is
empty. An empty entry
+ // list still reaches the exact structural result of zero below.
+ if (entry.recordCount() <= 0) {
return OptionalLong.empty();
}
try {
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 2563a54b35..c1f644febc 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
@@ -207,6 +207,43 @@ class CatalogManagedPartitionScanTest {
verify(catalog).listPartitionsPaged(IDENTIFIER, 1000, null, null);
}
+ @Test
+ void testPlanRowCountStaysUnknownWhenCatalogReportsZero() throws Exception
{
+ Catalog catalog = mock(Catalog.class);
+ Partition catalogPartition =
+ new Partition(partition("2025", "11").spec(), 0L, 1L, 1L, 0L,
-1, false);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(),
isNull()))
+ .thenReturn(new
PagedList<>(Collections.singletonList(catalogPartition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ 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(1);
+ assertThat(plan.rowCount()).isEqualTo(OptionalLong.empty());
+ }
+
+ @Test
+ void testPlanRowCountStaysUnknownWhenOneCatalogPartitionReportsZero()
throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition zero = new Partition(partition("2025", "10").spec(), 0L, 0L,
0L, 0L, -1, false);
+ Partition positive =
+ new Partition(partition("2025", "11").spec(), 3L, 1L, 1L, 0L,
-1, false);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(),
isNull()))
+ .thenReturn(new PagedList<>(Arrays.asList(zero, positive),
null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ 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(1);
+ assertThat(plan.rowCount()).isEqualTo(OptionalLong.empty());
+ }
+
@Test
void testPlanRowCountStaysUnknownWhenCatalogReportsNoStatistics() throws
Exception {
// Partitions as they come off the wire from a catalog that stores no
statistics. Summing
@@ -232,6 +269,21 @@ class CatalogManagedPartitionScanTest {
assertThat(plan.rowCount()).isEqualTo(OptionalLong.empty());
}
+ @Test
+ void testPlanRowCountIsZeroWhenCatalogReturnsNoPartitions() throws
Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(),
isNull()))
+ .thenReturn(new PagedList<>(Collections.emptyList(), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table = createTable(fileIO, tablePath,
partitionManager(catalog), false);
+
+ FormatTableScan.Plan plan = new FormatTableScan(table, null,
null).plan();
+
+ assertThat(plan.splits()).isEmpty();
+ assertThat(plan.rowCount()).isEqualTo(OptionalLong.of(0L));
+ }
+
@Test
void testUnderscoreInPartitionNameRemainsLiteralPrefix() {
LocalFileIO fileIO = LocalFileIO.create();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
index 534783fb02..dd4b7439e0 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
@@ -46,6 +46,7 @@ import
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.plan.stats.TableStats;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.slf4j.Logger;
@@ -56,6 +57,8 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.PrimitiveIterator;
+import java.util.stream.LongStream;
import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_PARTITIONS;
import static org.apache.paimon.options.OptionsUtils.PAIMON_PREFIX;
@@ -257,11 +260,12 @@ public abstract class FlinkTableSource
.newScan()
.listPartitionEntries();
long totalSize = 0;
- long rowCount = 0;
for (PartitionEntry entry : partitionEntries) {
totalSize += entry.fileSizeInBytes();
- rowCount += entry.recordCount();
}
+ long rowCount =
+ sumRowCounts(
+
partitionEntries.stream().mapToLong(PartitionEntry::recordCount));
long splitTargetSize = ((DataTable)
table).coreOptions().splitTargetSize();
splitStatistics =
new SplitStatistics((int) (totalSize / splitTargetSize
+ 1), rowCount);
@@ -277,11 +281,33 @@ public abstract class FlinkTableSource
.splits();
splitStatistics =
new SplitStatistics(
- splits.size(),
splits.stream().mapToLong(Split::rowCount).sum());
+ splits.size(),
+
sumRowCounts(splits.stream().mapToLong(Split::rowCount)));
}
}
}
+ /**
+ * Returns zero for an empty stream and Flink's unknown row-count sentinel
if any value is
+ * non-positive or the sum overflows.
+ */
+ static long sumRowCounts(LongStream rowCounts) {
+ PrimitiveIterator.OfLong iterator = rowCounts.iterator();
+ long totalRowCount = 0L;
+ while (iterator.hasNext()) {
+ long rowCount = iterator.nextLong();
+ if (rowCount <= 0) {
+ return TableStats.UNKNOWN.getRowCount();
+ }
+ try {
+ totalRowCount = Math.addExact(totalRowCount, rowCount);
+ } catch (ArithmeticException e) {
+ return TableStats.UNKNOWN.getRowCount();
+ }
+ }
+ return totalRowCount;
+ }
+
/** Split statistics for inferring row count and parallelism size. */
protected static class SplitStatistics {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceStatisticsTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceStatisticsTest.java
new file mode 100644
index 0000000000..7c73eceffa
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceStatisticsTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.paimon.flink.source;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.plan.stats.TableStats;
+import org.junit.jupiter.api.Test;
+import org.mockito.Answers;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.stream.LongStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for row count inference in {@link FlinkTableSource}. */
+class FlinkTableSourceStatisticsTest {
+
+ @Test
+ void testSumRowCounts() {
+ assertThat(FlinkTableSource.sumRowCounts(LongStream.empty())).isZero();
+ assertThat(FlinkTableSource.sumRowCounts(LongStream.of(2L,
3L))).isEqualTo(5L);
+
+ long unknown = TableStats.UNKNOWN.getRowCount();
+
assertThat(FlinkTableSource.sumRowCounts(LongStream.of(0L))).isEqualTo(unknown);
+ assertThat(FlinkTableSource.sumRowCounts(LongStream.of(3L,
0L))).isEqualTo(unknown);
+
assertThat(FlinkTableSource.sumRowCounts(LongStream.of(-1L))).isEqualTo(unknown);
+ assertThat(FlinkTableSource.sumRowCounts(LongStream.of(3L,
-2L))).isEqualTo(unknown);
+ assertThat(FlinkTableSource.sumRowCounts(LongStream.of(Long.MAX_VALUE,
1L)))
+ .isEqualTo(unknown);
+ }
+
+ @Test
+ void testNonDataTableInferenceReturnsUnknownForUnknownSplitRowCount() {
+ Table table = mock(Table.class);
+ when(table.options()).thenReturn(Collections.emptyMap());
+ when(table.statistics()).thenReturn(Optional.empty());
+ ReadBuilder readBuilder = mock(ReadBuilder.class,
Answers.RETURNS_SELF);
+ when(table.newReadBuilder()).thenReturn(readBuilder);
+ TableScan scan = mock(TableScan.class);
+ when(readBuilder.newScan()).thenReturn(scan);
+ TableScan.Plan plan = mock(TableScan.Plan.class);
+ when(scan.plan()).thenReturn(plan);
+ Split known = split(5L);
+ Split unknown = split(-1L);
+ when(plan.splits()).thenReturn(Arrays.asList(known, unknown));
+
+ DataTableSource source =
+ new DataTableSource(
+ ObjectIdentifier.of("catalog", "database", "table"),
table, false, null);
+
+ assertThat(source.reportStatistics().getRowCount())
+ .isEqualTo(TableStats.UNKNOWN.getRowCount());
+ }
+
+ @Test
+ void testDataTableInferenceReturnsUnknownForUnknownPartitionRowCount() {
+ DataTable table = mock(DataTable.class);
+ when(table.options()).thenReturn(Collections.emptyMap());
+ when(table.statistics()).thenReturn(Optional.empty());
+ CoreOptions coreOptions = mock(CoreOptions.class);
+ when(coreOptions.splitTargetSize()).thenReturn(128L);
+ when(table.coreOptions()).thenReturn(coreOptions);
+ ReadBuilder readBuilder = mock(ReadBuilder.class,
Answers.RETURNS_SELF);
+ when(table.newReadBuilder()).thenReturn(readBuilder);
+ TableScan scan = mock(TableScan.class);
+ when(readBuilder.newScan()).thenReturn(scan);
+ PartitionEntry known = mock(PartitionEntry.class);
+ when(known.recordCount()).thenReturn(5L);
+ PartitionEntry unknown = mock(PartitionEntry.class);
+ when(unknown.recordCount()).thenReturn(0L);
+ when(scan.listPartitionEntries()).thenReturn(Arrays.asList(known,
unknown));
+
+ DataTableSource source =
+ new DataTableSource(
+ ObjectIdentifier.of("catalog", "database", "table"),
table, false, null);
+
+ assertThat(source.reportStatistics().getRowCount())
+ .isEqualTo(TableStats.UNKNOWN.getRowCount());
+ }
+
+ private static Split split(long rowCount) {
+ Split split = mock(Split.class);
+ when(split.rowCount()).thenReturn(rowCount);
+ return split;
+ }
+}
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 9df3e683b4..f281c25ac9 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
@@ -47,18 +47,35 @@ case class PaimonStatistics(
lazy val numRows: OptionalLong = {
if (scanRowCount.isPresent) {
- // A catalog may report zero because it cannot tell "never measured"
from "measured, and
- // empty". Over real bytes leave it unknown; over zero bytes the scan
really is empty.
- if (scanRowCount.getAsLong > 0 || fileTotalSize == 0) {
+ val rowCount = scanRowCount.getAsLong
+ // A scan may use a non-positive row count for unknown. Only zero from a
scan with no splits
+ // proves that the result is empty.
+ if (rowCount > 0 || (rowCount == 0 && splits.isEmpty)) {
scanRowCount
} else {
OptionalLong.empty()
}
- } else if (splits.exists(_.rowCount() == -1)) {
- OptionalLong.empty()
} else {
- OptionalLong.of(splits.map(_.rowCount()).sum)
+ sumSplitRowCounts
+ }
+ }
+
+ private def sumSplitRowCounts: OptionalLong = {
+ var totalRowCount = 0L
+ var index = 0
+ while (index < splits.length) {
+ val rowCount = splits(index).rowCount()
+ if (rowCount <= 0) {
+ return OptionalLong.empty()
+ }
+ try {
+ totalRowCount = Math.addExact(totalRowCount, rowCount)
+ } catch {
+ case _: ArithmeticException => return OptionalLong.empty()
+ }
+ index += 1
}
+ OptionalLong.of(totalRowCount)
}
lazy val sizeInBytes: OptionalLong = {
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/read/PaimonStatisticsTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/read/PaimonStatisticsTest.scala
new file mode 100644
index 0000000000..837b917189
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/read/PaimonStatisticsTest.scala
@@ -0,0 +1,75 @@
+/*
+ * 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.paimon.spark.read
+
+import org.apache.paimon.table.source.Split
+import org.apache.paimon.types.{DataTypes, RowType}
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.{Optional, OptionalLong}
+
+class PaimonStatisticsTest extends AnyFunSuite {
+
+ private val rowType = RowType.of(DataTypes.INT())
+
+ test("non-positive split row counts are unknown") {
+ Seq(Seq(0L), Seq(-2L), Seq(5L, 0L), Seq(5L, -2L)).foreach {
+ rowCounts => assert(!statistics(rowCounts: _*).numRows.isPresent,
rowCounts.toString())
+ }
+
+ assert(statistics(2L, 3L).numRows.getAsLong == 5L)
+ }
+
+ test("non-positive scan row counts are unknown for a non-empty scan") {
+ Seq(0L, -1L, -2L).foreach {
+ rowCount =>
+ val result = statistics(OptionalLong.of(rowCount), -1L)
+ assert(!result.numRows.isPresent, rowCount.toString)
+ }
+ }
+
+ test("empty scans have an exact zero row count") {
+ assert(statistics().numRows.getAsLong == 0L)
+ assert(statistics(OptionalLong.of(0L)).numRows.getAsLong == 0L)
+ }
+
+ test("positive scan row count takes priority over split sentinels") {
+ assert(statistics(OptionalLong.of(4L), -1L).numRows.getAsLong == 4L)
+ }
+
+ test("overflowed split row count is unknown") {
+ assert(!statistics(Long.MaxValue, 1L).numRows.isPresent)
+ }
+
+ private def statistics(rowCounts: Long*): PaimonStatistics =
+ statistics(OptionalLong.empty(), rowCounts: _*)
+
+ private def statistics(scanRowCount: OptionalLong, rowCounts: Long*):
PaimonStatistics = {
+ val splits = rowCounts.map {
+ count =>
+ new Split {
+ override def rowCount(): Long = count
+
+ override def mergedRowCount(): OptionalLong = OptionalLong.empty()
+ }
+ }.toArray
+ PaimonStatistics(splits, rowType, rowType, Optional.empty(), scanRowCount)
+ }
+}
