This is an automated email from the ASF dual-hosted git repository.
zhangstar333 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ecdaf026564 [fix](paimon) Read row count estimates from snapshot
metadata (#68135)
ecdaf026564 is described below
commit ecdaf026564c6e6dc0a9e9af8944810450f66400
Author: zhangstar333 <[email protected]>
AuthorDate: Fri Sep 18 16:21:23 2026 +0800
[fix](paimon) Read row count estimates from snapshot metadata (#68135)
### 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.
---
.../doris/connector/paimon/PaimonCatalogOps.java | 63 +++-
.../connector/paimon/PaimonConnectorMetadata.java | 23 +-
.../paimon/PaimonCatalogRowCountTest.java | 345 +++++++++++++++++++++
.../PaimonConnectorMetadataStatisticsTest.java | 2 +-
4 files changed, 410 insertions(+), 23 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
index c60b10f0062..b52ffd8f360 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java
@@ -17,19 +17,23 @@
package org.apache.doris.connector.paimon;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogUtils;
import org.apache.paimon.catalog.Database;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.privilege.PrivilegedFileStoreTable;
import org.apache.paimon.rest.RESTCatalog;
import org.apache.paimon.schema.Schema;
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.tag.Tag;
import org.apache.paimon.types.DataField;
@@ -164,12 +168,9 @@ public interface PaimonCatalogOps {
boolean branchExists(Table table, String branchName);
/**
- * Returns the total row count of {@code table} = sum of {@code
split.rowCount()} over
- * {@code table.newReadBuilder().newScan().plan().splits()} (legacy
- * {@code PaimonExternalTable.fetchRowCount} / {@code
PaimonSysExternalTable.fetchRowCount}).
- * Returns a plain {@code long} (never a paimon {@code Split} list) so the
metadata layer's
- * >0-else-UNKNOWN logic is unit-testable offline with {@code
RecordingPaimonCatalogOps}
- * ({@code FakePaimonTable.newReadBuilder()} throws).
+ * Returns an optimizer estimate from the selected snapshot's unmerged
record count, or -1
+ * when snapshot metadata cannot describe the relation. Never plans splits
to obtain statistics.
+ * Like the former sum of split row counts, this is not an exact count for
primary-key tables.
*/
long rowCount(Table table);
@@ -422,13 +423,49 @@ public interface PaimonCatalogOps {
@Override
public long rowCount(Table table) {
- // Legacy PaimonExternalTable.fetchRowCount /
PaimonSysExternalTable.fetchRowCount: sum
- // the planned-split record counts.
- long rowCount = 0;
- for (Split split :
table.newReadBuilder().newScan().plan().splits()) {
- rowCount += split.rowCount();
+ // System/format tables have no data snapshot count. A fallback
pair combines two
+ // branches, so its main snapshot alone cannot estimate the
relation either.
+ if (!(table instanceof FileStoreTable)
+ ||
PaimonTableDecorators.unwrapToFallbackOrBase((FileStoreTable) table)
+ instanceof FallbackReadFileStoreTable) {
+ return -1;
}
- return rowCount;
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ CoreOptions options = fileStoreTable.coreOptions();
+ // Batch scans can exclude level-0 files or postponed buckets even
in full-snapshot mode.
+ // The snapshot counter includes those files; do not enumerate
manifests to correct it.
+ if ((!fileStoreTable.primaryKeys().isEmpty() &&
options.batchScanSkipLevel0()
+ &&
options.toConfiguration().get(CoreOptions.BATCH_SCAN_MODE) ==
CoreOptions.BatchScanMode.NONE)
+ || options.bucket() == BucketMode.POSTPONE_BUCKET) {
+ return -1;
+ }
+ switch (options.startupMode()) {
+ case LATEST:
+ case LATEST_FULL:
+ case FROM_TIMESTAMP:
+ case FROM_SNAPSHOT:
+ case FROM_SNAPSHOT_FULL:
+ break;
+ default:
+ // Incremental/file-creation-time scans and unresolved
compacted-full scans
+ // do not read the full snapshot selected by
TimeTravelUtil.
+ return -1;
+ }
+ if (fileStoreTable instanceof PrivilegedFileStoreTable) {
+ // Match the old scan's SELECT check without planning any
splits. TimeTravelUtil
+ // eagerly calls tagManager(), which requires INSERT on the
privilege wrapper.
+ fileStoreTable.newScan();
+ fileStoreTable =
PaimonTableDecorators.unwrapToFallbackOrBase(fileStoreTable);
+ }
+ // Catalog query authorization normally runs in scan.plan(),
independently of the
+ // privilege wrapper's SELECT check. Preserve it without planning
any file splits.
+ if (options.queryAuthEnabled()) {
+
fileStoreTable.catalogEnvironment().tableQueryAuth(options).auth(null);
+ }
+ Snapshot snapshot =
TimeTravelUtil.tryTravelOrLatest(fileStoreTable);
+ // Old snapshot versions can omit totalRecordCount; an empty table
has no snapshot.
+ return snapshot == null || snapshot.totalRecordCount() == null
+ ? -1 : snapshot.totalRecordCount();
}
@Override
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
index 4e69b899fb2..217f5cf5bf7 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
@@ -1595,13 +1595,10 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
}
/**
- * Returns the base-table row count = sum of planned-split row counts
(legacy
- * {@code PaimonExternalTable.fetchRowCount}: {@code rowCount > 0 ?
rowCount : UNKNOWN}). Shared
- * by normal AND system paimon tables: fe-core {@code
PluginDrivenSysExternalTable} inherits
- * {@code PluginDrivenExternalTable.fetchRowCount}, and {@link
#resolveTable} is sys-aware, so a
- * sys handle plans its OWN synthetic table's splits (closes Finding 5.1
with one override).
+ * Returns the base-table optimizer estimate from snapshot metadata
without planning splits.
+ * System tables and scan modes without a whole-snapshot estimate report
UNKNOWN.
* Returns {@code Optional.empty()} (→ fe-core -1 / UNKNOWN) when the
count is 0 (legacy parity)
- * or planning fails (best-effort, like the other connector read paths —
stats run in background
+ * or metadata loading fails (best-effort, like the other connector read
paths — stats run in background
* analysis / SHOW and must not surface a transient remote error as a
query-killing exception).
* {@code dataSize} is left UNKNOWN (-1): legacy computed no base-table
dataSize here.
*/
@@ -1611,6 +1608,9 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
PaimonTableHandle paimonHandle = (PaimonTableHandle) handle;
long rowCount;
try {
+ if
(PaimonScanParams.getPinnedFileCreationTime(paimonHandle.getScanOptions()).isPresent())
{
+ return Optional.empty();
+ }
Table table =
PaimonReaderOptions.runtimeSafeTable(resolveTable(paimonHandle));
table = runtimeSafeSystemTable(paimonHandle, table,
Collections.emptyMap());
PaimonReaderOptions.validateEffectiveTable(table);
@@ -1628,8 +1628,8 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
/**
* Row count AS OF the pinned snapshot, for a time-travel read. Applies
the snapshot to the handle (the
* SAME {@link #applySnapshot} the scan path uses) and copies its scan
options onto the resolved table,
- * so the summed split row counts reflect the pinned snapshot / branch /
tag — matching the rows
- * the scan reads instead of the latest count. Any failure degrades to
empty, and the caller then falls
+ * so the estimate reflects the pinned snapshot / branch / tag instead of
the latest count.
+ * Any failure degrades to empty, and the caller then falls
* back to the latest cached estimate (estimate-only, never a correctness
concern).
*/
@Override
@@ -1646,8 +1646,13 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
// first commit even though execution is still required to
scan zero rows.
return Optional.empty();
}
- Table table = resolveTable(pinned);
Map<String, String> scanOptions = pinned.getScanOptions();
+ // applyOptions removes this Doris-only marker, but execution
still filters files by it.
+ // This also covers scan.creation-time-millis when it resolves to
a file-creation scan.
+ if
(PaimonScanParams.getPinnedFileCreationTime(scanOptions).isPresent()) {
+ return Optional.empty();
+ }
+ Table table = resolveTable(pinned);
if (scanOptions != null && !scanOptions.isEmpty()) {
table = PaimonScanParams.isOptionsPin(scanOptions)
? PaimonScanParams.applyOptions(table, scanOptions)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogRowCountTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogRowCountTest.java
new file mode 100644
index 00000000000..d20a2c30911
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCatalogRowCountTest.java
@@ -0,0 +1,345 @@
+// 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.connector.paimon;
+
+import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.SeekableInputStream;
+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.TableSnapshot;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.system.SnapshotsTable;
+import org.apache.paimon.types.DataTypes;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.lang.reflect.Proxy;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/** Snapshot statistics must not open manifests or construct a split plan. */
+public class PaimonCatalogRowCountTest {
+ @TempDir
+ Path warehouse;
+
+ private final PaimonCatalogOps ops = new
PaimonCatalogOps.CatalogBackedPaimonCatalogOps(null);
+
+ @Test
+ public void latestAppendAndPrimaryKeyCountsNeedNoManifests() throws
Exception {
+ for (boolean primaryKey : new boolean[] {false, true}) {
+ FileStoreTable table = newTable("latest_" + primaryKey,
primaryKey);
+ snapshot(table, 1, 10L);
+ snapshot(table, 2, 25L);
+ Assertions.assertEquals(25L, ops.rowCount(table));
+ }
+ }
+
+ @Test
+ public void historicalSnapshotAndTimestampUseSelectedCount() throws
Exception {
+ FileStoreTable table = newTable("history", true);
+ snapshot(table, 1, 10L);
+ snapshot(table, 2, 25L);
+ Assertions.assertEquals(10L, ops.rowCount(table.copy(
+ Collections.singletonMap("scan.snapshot-id", "1"))));
+ Assertions.assertEquals(10L, ops.rowCount(table.copy(
+ Collections.singletonMap("scan.timestamp-millis", "1500"))));
+ }
+
+ @Test
+ public void tagSurvivesExpiredSnapshot() throws Exception {
+ FileStoreTable table = newTable("tag", false);
+ snapshot(table, 1, 10L);
+ table.createTag("retained", 1L);
+ snapshot(table, 2, 25L);
+ table.fileIO().delete(table.snapshotManager().snapshotPath(1), false);
+ Assertions.assertEquals(10L, ops.rowCount(table.copy(
+ Collections.singletonMap("scan.tag-name", "retained"))));
+ }
+
+ @Test
+ public void branchUsesItsOwnSnapshot() throws Exception {
+ FileStoreTable table = newTable("branch", false);
+ snapshot(table, 1, 10L);
+ table.createBranch("dev");
+ FileStoreTable branch = table.switchToBranch("dev");
+ snapshot(branch, 1, 7L);
+ snapshot(table, 2, 25L);
+ Assertions.assertEquals(7L, ops.rowCount(branch));
+ Assertions.assertEquals(25L, ops.rowCount(table));
+ }
+
+ @Test
+ public void emptyAndLegacySnapshotsDoNotFallBackToPlanning() throws
Exception {
+ FileStoreTable table = newTable("empty", false);
+ Assertions.assertEquals(-1L, ops.rowCount(table));
+ snapshot(table, 1, null);
+ Assertions.assertEquals(-1L, ops.rowCount(table));
+ snapshot(table, 2, 0L);
+ Assertions.assertEquals(0L, ops.rowCount(table));
+ }
+
+ @Test
+ public void partialScansDoNotUseWholeSnapshotCount() throws Exception {
+ FileStoreTable table = newTable("partial", false);
+ snapshot(table, 1, 10L);
+ snapshot(table, 2, 25L);
+ Assertions.assertEquals(-1L, ops.rowCount(table.copy(
+ Collections.singletonMap("incremental-between", "1,2"))));
+ Assertions.assertEquals(-1L, ops.rowCount(table.copy(
+ Collections.singletonMap("scan.file-creation-time-millis",
"1500"))));
+ Assertions.assertEquals(-1L, ops.rowCount(table.copy(
+ Collections.singletonMap("scan.mode", "compacted-full"))));
+ }
+
+ @Test
+ public void systemAndFallbackTablesDoNotUseBaseCount() throws Exception {
+ FileStoreTable table = newTable("system", false);
+ snapshot(table, 1, 10L);
+ Assertions.assertEquals(-1L, ops.rowCount(new SnapshotsTable(table)));
+ Assertions.assertEquals(-1L, ops.rowCount(new FakePaimonTable("format",
+ DataTypes.ROW(DataTypes.FIELD(0, "id", DataTypes.INT())),
+ Collections.emptyList(), Collections.emptyList())));
+ FileStoreTable fallback = newTable("fallback", false);
+ snapshot(fallback, 1, 20L);
+ Assertions.assertEquals(-1L, ops.rowCount(new
FallbackReadFileStoreTable(table, fallback)));
+ }
+
+ @Test
+ public void committedFilesExcludedByBatchScanReturnUnknown() throws
Exception {
+ for (Map<String, String> options : List.of(
+ Collections.singletonMap("deletion-vectors.enabled", "true"),
+ Collections.singletonMap("merge-engine", "first-row"),
+ Collections.singletonMap("bucket", "-2"))) {
+ ManifestGuardFileIO fileIO = new ManifestGuardFileIO();
+ fileIO.rejectManifests = false;
+ FileStoreTable table =
newTable(options.keySet().iterator().next(), true, options, fileIO);
+ append(table);
+ Assertions.assertEquals(1L,
table.latestSnapshot().get().totalRecordCount().longValue());
+ Assertions.assertEquals(0L,
table.newScan().plan().splits().stream().mapToLong(Split::rowCount).sum(),
+ "The committed file must be invisible to the ordinary
batch scan");
+
+ FileStoreTable compactScan =
table.copy(Collections.singletonMap("batch-scan-mode", "compact"));
+ fileIO.rejectManifests = true;
+ Assertions.assertEquals(-1L, ops.rowCount(table));
+ Assertions.assertEquals(options.containsKey("bucket") ? -1L : 1L,
ops.rowCount(compactScan));
+ }
+ }
+
+ @Test
+ public void committedOrdinaryTableKeepsSnapshotEstimate() throws Exception
{
+ ManifestGuardFileIO fileIO = new ManifestGuardFileIO();
+ fileIO.rejectManifests = false;
+ FileStoreTable table = newTable("committed", true,
Collections.emptyMap(), fileIO);
+ append(table);
+ Assertions.assertEquals(1L,
table.newScan().plan().splits().stream().mapToLong(Split::rowCount).sum());
+ fileIO.rejectManifests = true;
+ Assertions.assertEquals(1L, ops.rowCount(table));
+ }
+
+ @Test
+ public void preservedPrivilegeWrapperNeedsOnlySelect() throws Exception {
+ FileStoreTable table = newTable("privileged", false);
+ snapshot(table, 1, 10L);
+ AtomicInteger selectChecks = new AtomicInteger();
+ AtomicBoolean denySelect = new AtomicBoolean();
+ PrivilegeChecker checker = (PrivilegeChecker) Proxy.newProxyInstance(
+ PrivilegeChecker.class.getClassLoader(), new Class<?>[]
{PrivilegeChecker.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("assertCanInsert")) {
+ throw new SecurityException("INSERT denied");
+ }
+ if (method.getName().equals("assertCanSelect")
+ ||
method.getName().equals("assertCanSelectOrInsert")) {
+ selectChecks.incrementAndGet();
+ if (denySelect.get()) {
+ throw new SecurityException("SELECT denied");
+ }
+ }
+ return null;
+ });
+ FileStoreTable privileged = PrivilegedFileStoreTable.wrap(table,
checker, Identifier.create("db", "t"));
+ Assertions.assertSame(privileged,
PaimonReaderOptions.runtimeSafeTable(privileged));
+ PaimonConnectorMetadata metadata = metadata();
+ PaimonTableHandle handle = handle(privileged);
+ Assertions.assertEquals(10L, metadata.getTableStatistics(null,
handle).get().getRowCount());
+ Assertions.assertEquals(10L, metadata.getTableStatistics(null, handle,
+
ConnectorMvccSnapshot.builder().snapshotId(1L).property("scan.snapshot-id",
"1").build())
+ .get().getRowCount());
+ Assertions.assertTrue(selectChecks.get() >= 2);
+ denySelect.set(true);
+ Assertions.assertThrows(SecurityException.class, () ->
ops.rowCount(privileged));
+ }
+
+ @Test
+ public void catalogQueryAuthorizationControlsSnapshotStatistics() throws
Exception {
+ FileStoreTable base = newTable("query_auth", false,
+ Collections.singletonMap("query-auth.enabled", "true"), new
ManifestGuardFileIO());
+ snapshot(base, 1, 10L);
+ Identifier identifier = Identifier.create("db", "t");
+ AtomicInteger authCalls = new AtomicInteger();
+ AtomicBoolean denyQuery = new AtomicBoolean();
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
base.location()) {
+ @Override
+ public List<String> authTableQuery(Identifier requested,
List<String> select) {
+ Assertions.assertEquals(identifier, requested);
+ Assertions.assertNull(select, "Statistics must preserve the
old all-column authorization");
+ authCalls.incrementAndGet();
+ if (denyQuery.get()) {
+ throw new Catalog.TableNoPermissionException(requested);
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Optional<TableSnapshot> loadSnapshot(Identifier requested) {
+ return Optional.of(new
TableSnapshot(base.snapshotManager().snapshot(1L), 0L, 0L, 0L, 0L));
+ }
+ }) {
+ // Like a REST-loaded table, this has a catalog loader but no
privilege wrapper.
+ CatalogEnvironment environment = new CatalogEnvironment(
+ identifier, null, () -> catalog, null, null, false);
+ FileStoreTable table = FileStoreTableFactory.create(
+ base.fileIO(), base.location(), base.schema(),
environment);
+ PaimonConnectorMetadata metadata = metadata();
+ PaimonTableHandle handle = handle(table);
+ ConnectorMvccSnapshot pinned = ConnectorMvccSnapshot.builder()
+ .snapshotId(1L).property("scan.snapshot-id", "1").build();
+
+ Assertions.assertEquals(10L, metadata.getTableStatistics(null,
handle).get().getRowCount());
+ Assertions.assertEquals(10L, metadata.getTableStatistics(null,
handle, pinned).get().getRowCount());
+ Assertions.assertEquals(2, authCalls.get());
+
+ denyQuery.set(true);
+ RuntimeException denied =
Assertions.assertThrows(RuntimeException.class, () -> ops.rowCount(table));
+
Assertions.assertInstanceOf(Catalog.TableNoPermissionException.class,
denied.getCause());
+ Assertions.assertFalse(metadata.getTableStatistics(null,
handle).isPresent());
+ Assertions.assertFalse(metadata.getTableStatistics(null, handle,
pinned).isPresent());
+ Assertions.assertEquals(5, authCalls.get());
+
+ FileStoreTable disabled =
table.copy(Collections.singletonMap("query-auth.enabled", "false"));
+ Assertions.assertEquals(10L, ops.rowCount(disabled));
+ Assertions.assertEquals(5, authCalls.get(), "Disabled query
authorization must not contact the catalog");
+ }
+ }
+
+ @Test
+ public void normalizedFileCreationHandlesReturnUnknown() throws Exception {
+ ManifestGuardFileIO fileIO = new ManifestGuardFileIO();
+ fileIO.rejectManifests = false;
+ FileStoreTable table = newTable("creation", false,
Collections.emptyMap(), fileIO);
+ append(table);
+ fileIO.rejectManifests = true;
+ PaimonConnectorMetadata metadata = metadata();
+ for (String key : new String[] {"scan.file-creation-time-millis",
"scan.creation-time-millis"}) {
+ // A threshold before the first snapshot forces creation-time's
file-filter fallback.
+ Map<String, String> resolved =
PaimonScanParams.markAsOptions(PaimonScanParams.resolveOptions(
+ table, Collections.singletonMap(key, "1")));
+
Assertions.assertTrue(PaimonScanParams.getPinnedFileCreationTime(resolved).isPresent());
+ FileStoreTable selected = (FileStoreTable)
PaimonScanParams.applyOptions(table, resolved);
+ Assertions.assertEquals(CoreOptions.StartupMode.FROM_SNAPSHOT,
selected.coreOptions().startupMode());
+ Assertions.assertEquals(1L, ops.rowCount(selected), "The Table
alone has lost the file filter");
+ ConnectorMvccSnapshot snapshot =
ConnectorMvccSnapshot.builder().snapshotId(1L)
+ .properties(resolved).build();
+ Assertions.assertFalse(metadata.getTableStatistics(null,
handle(table), snapshot).isPresent());
+ Assertions.assertFalse(metadata.getTableStatistics(null,
+ handle(table).withScanOptions(resolved)).isPresent());
+ }
+ }
+
+ private PaimonConnectorMetadata metadata() {
+ return new PaimonConnectorMetadata(ops,
PaimonCatalogProperties.of(Collections.emptyMap()),
+ new RecordingConnectorContext());
+ }
+
+ private PaimonTableHandle handle(FileStoreTable table) {
+ PaimonTableHandle handle = new PaimonTableHandle("db", "t",
table.partitionKeys(), table.primaryKeys());
+ handle.setPaimonTable(table);
+ return handle;
+ }
+
+ private void append(FileStoreTable table) throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite(); BatchTableCommit
commit = builder.newCommit()) {
+ write.write(GenericRow.of(1));
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private FileStoreTable newTable(String name, boolean primaryKey) throws
Exception {
+ return newTable(name, primaryKey, Collections.emptyMap(), new
ManifestGuardFileIO());
+ }
+
+ private FileStoreTable newTable(String name, boolean primaryKey,
+ Map<String, String> options, LocalFileIO fileIO) throws Exception {
+ org.apache.paimon.fs.Path path = new
org.apache.paimon.fs.Path(warehouse.resolve(name).toUri());
+ Schema.Builder schema = Schema.newBuilder().column("id",
DataTypes.INT())
+ .option("file.format", "parquet").option("write-only", "true")
+ .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
IOException {
+ // Deliberately no manifest files: a return to split planning must
fail this test.
+ Snapshot snapshot = new Snapshot(id, 0L, "unused-base", null,
"unused-delta", null,
+ null, null, null, "test", id, Snapshot.CommitKind.APPEND, id *
1000,
+ Collections.emptyMap(), count, count, null, null, null, null,
null);
+
table.fileIO().mkdirs(table.snapshotManager().snapshotPath(id).getParent());
+
table.fileIO().overwriteFileUtf8(table.snapshotManager().snapshotPath(id),
snapshot.toJson());
+ table.snapshotManager().commitLatestHint(id);
+ }
+
+ private static class ManifestGuardFileIO extends LocalFileIO {
+ private boolean rejectManifests = true;
+
+ @Override
+ public SeekableInputStream newInputStream(org.apache.paimon.fs.Path
path) throws IOException {
+ Assertions.assertFalse(rejectManifests &&
path.toString().contains("/manifest/"),
+ "Row count estimation must not read manifests: " + path);
+ return super.newInputStream(path);
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataStatisticsTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataStatisticsTest.java
index 410cb35ff3e..c8c528b6bc2 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataStatisticsTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataStatisticsTest.java
@@ -43,7 +43,7 @@ import java.util.Optional;
* <p>Before the fix the connector inherited the default {@code
ConnectorStatisticsOps} (returns
* {@code Optional.empty()}), so every paimon table — normal AND system —
reported row count -1
* (UNKNOWN), degrading the Nereids cost model (join-reorder force-disabled)
and SHOW/info_schema.
- * The fix overrides it to sum {@code split.rowCount()} via the {@code
PaimonCatalogOps.rowCount}
+ * The connector obtains a snapshot estimate via the {@code
PaimonCatalogOps.rowCount}
* seam (faked here — {@code FakePaimonTable.newReadBuilder()} throws, the
whole reason for the
* seam). Each test FAILS before the fix (default empty) and PASSES after, and
encodes WHY.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
