This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 094b27b7c4d [fix](paimon) Read row count estimates from snapshot
metadata (#68155)
094b27b7c4d is described below
commit 094b27b7c4d92bc42ab80b326cf5e97b90966a9a
Author: zhangstar333 <[email protected]>
AuthorDate: Sat Sep 19 09:43:29 2026 +0800
[fix](paimon) Read row count estimates from snapshot metadata (#68155)
### What problem does this PR solve?
Problem Summary:
Ordinary Paimon queries request table cardinality during planning. On a
row-count cache miss, the connector planned every split and summed file
record counts, consuming CPU and memory proportional to the table's
manifests and files even when the query selects only a small partition.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../datasource/paimon/PaimonExternalTable.java | 53 ++++--
.../datasource/paimon/PaimonSysExternalTable.java | 14 +-
.../datasource/paimon/PaimonExternalTableTest.java | 8 +-
.../datasource/paimon/PaimonRowCountTest.java | 191 +++++++++++++++++++++
4 files changed, 240 insertions(+), 26 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
index e142d42c7f5..7996b06106d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
@@ -55,11 +55,14 @@ import org.apache.logging.log4j.Logger;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.privilege.PrivilegedFileStoreTable;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.DataTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypeRoot;
@@ -269,20 +272,48 @@ public class PaimonExternalTable extends ExternalTable
implements MTMVRelatedTab
@Override
public long fetchRowCount() {
makeSureInitialized();
- long rowCount = 0;
- // Row-count planning bypasses ScanNode, so build the same CPU-capped
disposable handle
- // here instead of validating the hardware-neutral catalog copy
directly.
+ // Keep the reader policy consistent with scan planning, including
privilege wrappers.
Table effectiveTable =
PaimonReaderOptions.runtimeSafeTable(getBasePaimonTable());
- // Statistics and row-count cache planning run before ScanNode and
must not reach an
- // unsafe manifest executor, even when the foreground relation later
supplies an override.
PaimonReaderOptions.validateEffectiveTable(effectiveTable);
- List<Split> splits =
effectiveTable.newReadBuilder().newScan().plan().splits();
- for (Split split : splits) {
- rowCount += split.rowCount();
+ if (!(effectiveTable instanceof FileStoreTable)
+ ||
PaimonTableDecorators.unwrapToFallbackOrBase((FileStoreTable) effectiveTable)
+ instanceof FallbackReadFileStoreTable) {
+ return UNKNOWN_ROW_COUNT;
}
- if (rowCount == 0) {
- LOG.info("Paimon table {} row count is 0, return -1", name);
+ FileStoreTable table = (FileStoreTable) effectiveTable;
+ CoreOptions options = table.coreOptions();
+ // These batch scans exclude some files from the snapshot. Do not plan
splits merely
+ // to refine an optimizer estimate.
+ if ((!table.primaryKeys().isEmpty() && options.batchScanSkipLevel0()
+ && options.toConfiguration().get(CoreOptions.BATCH_SCAN_MODE)
== CoreOptions.BatchScanMode.NONE)
+ || options.bucket() == BucketMode.POSTPONE_BUCKET) {
+ return UNKNOWN_ROW_COUNT;
}
+ switch (options.startupMode()) {
+ case LATEST:
+ case LATEST_FULL:
+ case FROM_TIMESTAMP:
+ case FROM_SNAPSHOT:
+ case FROM_SNAPSHOT_FULL:
+ break;
+ default:
+ // Incremental, file-creation-time and compacted scans do not
necessarily read
+ // the complete snapshot selected by TimeTravelUtil.
+ return UNKNOWN_ROW_COUNT;
+ }
+ if (table instanceof PrivilegedFileStoreTable) {
+ // Preserve SELECT authorization without planning. TimeTravelUtil
calls tagManager(),
+ // which would incorrectly require INSERT permission on the
privilege wrapper.
+ table.newScan();
+ table = PaimonTableDecorators.unwrapToFallbackOrBase(table);
+ }
+ if (options.queryAuthEnabled()) {
+ table.catalogEnvironment().tableQueryAuth(options).auth(null);
+ }
+ Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table);
+ // Read the snapshot counter without enumerating manifests/files or
materializing splits.
+ // For primary-key tables this is a physical record estimate, not an
exact logical count.
+ long rowCount = snapshot == null ? UNKNOWN_ROW_COUNT :
snapshot.totalRecordCount();
return rowCount > 0 ? rowCount : UNKNOWN_ROW_COUNT;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java
index 6e154d7e6d0..3286f941140 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java
@@ -39,7 +39,6 @@ import org.apache.logging.log4j.Logger;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypeRoot;
@@ -358,16 +357,9 @@ public class PaimonSysExternalTable extends ExternalTable {
@Override
public long fetchRowCount() {
- makeSureInitialized();
- long rowCount = 0;
- List<Split> splits =
getSysPaimonTable().newReadBuilder().newScan().plan().splits();
- for (Split split : splits) {
- rowCount += split.rowCount();
- }
- if (rowCount == 0) {
- LOG.info("Paimon system table {} row count is 0, return -1", name);
- }
- return rowCount > 0 ? rowCount : UNKNOWN_ROW_COUNT;
+ // System-table row counts cannot use the data snapshot's record
count. Planning their
+ // splits can enumerate all manifests (e.g. $files), so never do it
for an estimate.
+ return UNKNOWN_ROW_COUNT;
}
@Override
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
index ef0e3e0a42e..27f5557cf8f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalTableTest.java
@@ -551,12 +551,11 @@ public class PaimonExternalTableTest {
}
@Test
- public void
testFetchRowCountCapsAcceptedManifestParallelismBeforePlanning() {
+ public void
testFetchRowCountCapsAcceptedManifestParallelismWithoutPlanning() {
int localCapacity = Runtime.getRuntime().availableProcessors();
org.junit.Assume.assumeTrue(localCapacity <
PaimonReaderOptions.MAX_MANIFEST_PARALLELISM);
FileStoreTable rawTable = Mockito.mock(FileStoreTable.class);
FileStoreTable cappedTable = Mockito.mock(FileStoreTable.class);
- ReadBuilder readBuilder = Mockito.mock(ReadBuilder.class,
Mockito.RETURNS_DEEP_STUBS);
Mockito.when(rawTable.options()).thenReturn(ImmutableMap.of(
CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
String.valueOf(localCapacity + 1)));
Mockito.when(rawTable.copyWithoutTimeTravel(ArgumentMatchers.argThat(options ->
@@ -565,8 +564,8 @@ public class PaimonExternalTableTest {
.thenReturn(cappedTable);
Mockito.when(cappedTable.options()).thenReturn(ImmutableMap.of(
CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
String.valueOf(localCapacity)));
- Mockito.when(cappedTable.newReadBuilder()).thenReturn(readBuilder);
-
Mockito.when(readBuilder.newScan().plan().splits()).thenReturn(Collections.emptyList());
+ CoreOptions cappedOptions = CoreOptions.fromMap(cappedTable.options());
+ Mockito.when(cappedTable.coreOptions()).thenReturn(cappedOptions);
PaimonExternalTable externalTable = Mockito.mock(
PaimonExternalTable.class, Mockito.CALLS_REAL_METHODS);
Mockito.doNothing().when(externalTable).makeSureInitialized();
@@ -576,6 +575,7 @@ public class PaimonExternalTableTest {
Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT,
externalTable.fetchRowCount());
}
Mockito.verify(rawTable).copyWithoutTimeTravel(ArgumentMatchers.anyMap());
+ Mockito.verify(cappedTable, Mockito.never()).newReadBuilder();
}
@Test
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonRowCountTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonRowCountTest.java
new file mode 100644
index 00000000000..bba3e27f3ac
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonRowCountTest.java
@@ -0,0 +1,191 @@
+// 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.doris.datasource.paimon;
+
+import org.apache.doris.catalog.TableIf;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.privilege.PrivilegeChecker;
+import org.apache.paimon.privilege.PrivilegedFileStoreTable;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.DataTypes;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class PaimonRowCountTest {
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testSystemTableEstimateDoesNotLoadMetadataOrPlan() {
+ PaimonSysExternalTable table =
Mockito.mock(PaimonSysExternalTable.class, invocation -> {
+ throw new AssertionError("Row count estimation must not access
metadata: " + invocation.getMethod());
+ });
+ Mockito.doCallRealMethod().when(table).fetchRowCount();
+
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, table.fetchRowCount());
+ }
+
+ @Test
+ public void testLatestCountWithoutManifests() throws Exception {
+ for (boolean primaryKey : new boolean[] {false, true}) {
+ FileStoreTable table = newTable(primaryKey);
+ snapshot(table, 1, 10L);
+ snapshot(table, 2, 25L);
+ Assert.assertEquals(25L, rowCount(table));
+ }
+ }
+
+ @Test
+ public void testSnapshotTimestampAndTagCounts() throws Exception {
+ FileStoreTable table = newTable(false);
+ snapshot(table, 1, 10L);
+ table.createTag("retained", 1L);
+ snapshot(table, 2, 25L);
+ Assert.assertEquals(10L,
rowCount(table.copy(ImmutableMap.of("scan.snapshot-id", "1"))));
+ Assert.assertEquals(10L,
rowCount(table.copy(ImmutableMap.of("scan.timestamp-millis", "1500"))));
+ table.fileIO().delete(table.snapshotManager().snapshotPath(1), false);
+ Assert.assertEquals(10L,
rowCount(table.copy(ImmutableMap.of("scan.tag-name", "retained"))));
+ }
+
+ @Test
+ public void testEmptyAndNonpositiveCountsReturnUnknown() throws Exception {
+ FileStoreTable table = newTable(false);
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, rowCount(table));
+ snapshot(table, 1, 0L);
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, rowCount(table));
+ snapshot(table, 2, -1L);
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, rowCount(table));
+ }
+
+ @Test
+ public void testPartialScansReturnUnknownWithoutManifests() throws
Exception {
+ FileStoreTable table = newTable(false);
+ snapshot(table, 1, 10L);
+ snapshot(table, 2, 25L);
+ for (String[] option : new String[][] {
+ {"incremental-between", "1,2"},
+ {"scan.file-creation-time-millis", "1500"},
+ {"scan.creation-time-millis", "1500"},
+ {"scan.mode", "compacted-full"}}) {
+ Assert.assertEquals(option[0], TableIf.UNKNOWN_ROW_COUNT,
+ rowCount(table.copy(ImmutableMap.of(option[0],
option[1]))));
+ }
+ FileStoreTable deletionVectors = newTable(true,
ImmutableMap.of("deletion-vectors.enabled", "true"));
+ snapshot(deletionVectors, 1, 10L);
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT,
rowCount(deletionVectors));
+ FileStoreTable postponedBuckets = newTable(true,
ImmutableMap.of("bucket", "-2"));
+ snapshot(postponedBuckets, 1, 10L);
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT,
rowCount(postponedBuckets));
+ }
+
+ @Test
+ public void testUnsupportedTablesDoNotPlan() throws Exception {
+ Table formatTable = Mockito.mock(Table.class);
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, rowCount(formatTable));
+ Mockito.verify(formatTable, Mockito.never()).newReadBuilder();
+ FileStoreTable main = newTable(false);
+ FileStoreTable other = newTable(false);
+ snapshot(main, 1, 10L);
+ snapshot(other, 1, 20L);
+ Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT,
+ rowCount(new FallbackReadFileStoreTable(main, other, true)));
+ }
+
+ @Test
+ public void testSelectOnlyPrivilegeWrapper() throws Exception {
+ FileStoreTable table = newTable(false);
+ snapshot(table, 1, 10L);
+ PrivilegeChecker checker = Mockito.mock(PrivilegeChecker.class);
+ Identifier identifier = Identifier.create("db", "tbl");
+ Mockito.doThrow(new IllegalStateException("INSERT is not granted"))
+ .when(checker).assertCanInsert(identifier);
+ FileStoreTable privileged = PrivilegedFileStoreTable.wrap(table,
checker, identifier);
+ Assert.assertEquals(10L, rowCount(privileged));
+ Mockito.verify(checker).assertCanSelect(identifier);
+ Mockito.verify(checker, Mockito.never()).assertCanInsert(identifier);
+ Mockito.doThrow(new IllegalStateException("SELECT is not granted"))
+ .when(checker).assertCanSelect(identifier);
+ Assert.assertThrows(IllegalStateException.class, () ->
rowCount(privileged));
+ }
+
+ @Test
+ public void testCatalogQueryAuthorizationWithoutPlanning() throws
Exception {
+ FileStoreTable table =
Mockito.spy(newTable(false).copy(ImmutableMap.of("query-auth.enabled",
"true")));
+ snapshot(table, 1, 10L);
+ CatalogEnvironment environment =
Mockito.mock(CatalogEnvironment.class, Mockito.RETURNS_DEEP_STUBS);
+ Mockito.doReturn(environment).when(table).catalogEnvironment();
+ Assert.assertEquals(10L, rowCount(table));
+
Mockito.verify(environment.tableQueryAuth(Mockito.any(CoreOptions.class))).auth(null);
+
Mockito.when(environment.tableQueryAuth(Mockito.any(CoreOptions.class)).auth(null))
+ .thenThrow(new IllegalStateException("Query is not
authorized"));
+ Assert.assertThrows(IllegalStateException.class, () ->
rowCount(table));
+ Mockito.verify(table, Mockito.never()).newReadBuilder();
+ }
+
+ private long rowCount(Table table) {
+ PaimonExternalTable external = Mockito.mock(PaimonExternalTable.class,
Mockito.CALLS_REAL_METHODS);
+ Mockito.doNothing().when(external).makeSureInitialized();
+ Mockito.doReturn(table).when(external).getBasePaimonTable();
+ return external.fetchRowCount();
+ }
+
+ private FileStoreTable newTable(boolean primaryKey) throws Exception {
+ return newTable(primaryKey, Collections.emptyMap());
+ }
+
+ private FileStoreTable newTable(boolean primaryKey, Map<String, String>
options) throws Exception {
+ Path path = new Path(temporaryFolder.newFolder().toURI());
+ LocalFileIO fileIO = LocalFileIO.create();
+ Schema.Builder schema = Schema.newBuilder().column("id",
DataTypes.INT())
+ .option("scan.manifest.parallelism", "1");
+ if (primaryKey) {
+ schema.primaryKey("id").option("bucket", "1");
+ }
+ options.forEach(schema::option);
+ new SchemaManager(fileIO, path).createTable(schema.build());
+ return FileStoreTableFactory.create(fileIO, path);
+ }
+
+ private void snapshot(FileStoreTable table, long id, long count) throws
Exception {
+ // No manifest files exist: accidentally returning to split planning
must fail.
+ Snapshot snapshot = new Snapshot(id, 0L, "unused-base", null,
"unused-delta", null,
+ null, null, null, "test", id, Snapshot.CommitKind.APPEND, id *
1000,
+ count, count, null, null, null, Collections.emptyMap(), null);
+
table.fileIO().mkdirs(table.snapshotManager().snapshotPath(id).getParent());
+
table.fileIO().overwriteFileUtf8(table.snapshotManager().snapshotPath(id),
snapshot.toJson());
+ table.snapshotManager().commitLatestHint(id);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
