From 669e18030f18e466a08eba18887af2905cfd6296 Mon Sep 17 00:00:00 2001 From: yangtao555 Date: Thu, 1 Oct 2026 22:00:23 +0800 Subject: [PATCH 1/2] feature(stats) Estimate scan cardinality from selected partitions for OLAP and external tables --- .../connector/hive/HiveConnectorMetadata.java | 77 ++++ ...iveConnectorMetadataFileListStatsTest.java | 103 +++++ ...onnectorMetadataSiblingDelegationTest.java | 16 +- .../connector/spi/ConnectorStatisticsOps.java | 17 + .../spi/ConnectorPluginSurfaceTest.java | 6 +- .../resources/connector-metadata-methods.txt | 1 + fe/fe-connector/pom.xml | 2 +- .../org/apache/doris/catalog/OlapTable.java | 30 ++ .../doris/datasource/ExternalTable.java | 6 + .../plugin/PluginDrivenExternalTable.java | 67 +++ .../post/PrunePartitionPredicate.java | 76 +--- .../mv/SyncMaterializationContext.java | 10 +- .../expression/rules/PartitionPruner.java | 5 +- .../rules/rewrite/PruneFileScanPartition.java | 14 +- .../rules/rewrite/PruneOlapScanPartition.java | 4 +- .../nereids/rules/rewrite/StatsDerive.java | 12 +- .../doris/nereids/stats/FilterEstimation.java | 8 + .../doris/nereids/stats/StatsCalculator.java | 204 +++++++-- .../trees/copier/LogicalPlanDeepCopier.java | 14 +- .../plans/PartitionPrunablePredicate.java | 58 ++- .../nereids/trees/plans/algebra/OlapScan.java | 4 + .../trees/plans/logical/LogicalFileScan.java | 115 ++++- .../trees/plans/logical/LogicalOlapScan.java | 82 +++- .../logical/LogicalOlapTableStreamScan.java | 6 +- .../PhysicalLazyMaterializeOlapScan.java | 6 +- .../plans/physical/PhysicalOlapScan.java | 5 + .../doris/statistics/model/Statistics.java | 25 +- .../statistics/model/StatisticsBuilder.java | 12 +- .../apache/doris/catalog/OlapTableTest.java | 19 + ...PluginDrivenExternalTableRowCountTest.java | 125 ++++++ .../rules/rewrite/PartitionPrunerTest.java | 18 +- .../rewrite/PullUpJoinFromUnionAllTest.java | 39 ++ .../rules/rewrite/StatsDeriveTest.java | 70 ++++ .../nereids/stats/FilterEstimationTest.java | 12 + .../nereids/stats/StatsCalculatorTest.java | 393 +++++++++++++++++- .../plans/logical/LogicalFileScanTest.java | 75 ++++ .../plans/logical/LogicalOlapScanTest.java | 135 ++++++ 37 files changed, 1703 insertions(+), 168 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/StatsDeriveTest.java diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java index 402a69bcd85f57..324bcce2716112 100644 --- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java @@ -1000,6 +1000,40 @@ public long estimateDataSizeByListingFiles(ConnectorSession session, ConnectorTa } } + @Override + public long estimateDataSizeByListingFiles(ConnectorSession session, ConnectorTableHandle handle, + List selectedPartitionNames) { + if (!(handle instanceof HiveTableHandle)) { + return siblingMetadata(session, handle) + .estimateDataSizeByListingFiles(session, handle, selectedPartitionNames); + } + HiveTableHandle hiveHandle = (HiveTableHandle) handle; + if (hiveHandle.getTableType() != HiveTableType.HIVE) { + return -1; + } + if (selectedPartitionNames.isEmpty()) { + return 0; + } + if (hiveHandle.isTransactional()) { + return -1; + } + + ClassLoader previous = Thread.currentThread().getContextClassLoader(); + try { + Thread.currentThread().setContextClassLoader(getClass().getClassLoader()); + FileSystem fs = storage().getFileSystem(session); + return estimateSelectedPartitionsDataSize(hiveHandle, selectedPartitionNames, + STATS_PARTITION_SAMPLE_SIZE, + (location, values) -> sumCachedFileSizes(hiveHandle, location, values, fs)); + } catch (RuntimeException e) { + LOG.warn("Failed to estimate selected hive partition data size for {}.{} from file list", + hiveHandle.getDbName(), hiveHandle.getTableName(), e); + return -1; + } finally { + Thread.currentThread().setContextClassLoader(previous); + } + } + /** * Returns the raw byte length of every data file across ALL partitions (not sampled, not summed), a port of * legacy {@code HMSExternalTable.getChunkSizes} for {@code ANALYZE ... WITH SAMPLE}. Only plain-hive tables @@ -1079,6 +1113,49 @@ long estimateDataSize(HiveTableHandle handle, int sampleSize, ToLongBiFunction selectedPartitionNames, + int sampleSize, ToLongBiFunction> sizeOf) { + try { + if (selectedPartitionNames.isEmpty()) { + return 0; + } + int selectedPartitionCount = selectedPartitionNames.size(); + boolean sampled = sampleSize > 0 && sampleSize < selectedPartitionCount; + List chosenPartitionNames = selectedPartitionNames; + if (sampled) { + List shuffled = new ArrayList<>(selectedPartitionNames); + Collections.shuffle(shuffled); + chosenPartitionNames = shuffled.subList(0, sampleSize); + } + List partitions = hmsClient.getExistingPartitions( + handle.getDbName(), handle.getTableName(), chosenPartitionNames); + if (partitions.size() != chosenPartitionNames.size()) { + return -1; + } + List refs = new ArrayList<>(partitions.size()); + for (HmsPartitionInfo partition : partitions) { + String location = partition.getLocation(); + if (location == null || location.isEmpty()) { + return -1; + } + refs.add(new PartitionRef(location, partition.getValues())); + } + + long selectedSize = 0; + for (PartitionRef ref : refs) { + selectedSize += Math.max(0, sizeOf.applyAsLong(ref.location, ref.partitionValues)); + } + if (sampled && selectedSize == 0) { + return -1; + } + return sampled ? scaleSampledSize(selectedSize, selectedPartitionCount, refs.size()) : selectedSize; + } catch (RuntimeException e) { + LOG.warn("Failed to estimate selected hive partition data size for {}.{} from file list", + handle.getDbName(), handle.getTableName(), e); + return -1; + } + } + /** * Scales a sampled data size up to the whole table: {@code sampledSize * totalPartitions / * sampledPartitions} (legacy {@code HMSExternalTable.getRowCountFromFileList}). Multiplies BEFORE dividing diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataFileListStatsTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataFileListStatsTest.java index c015e40b466df4..b2d0592d1828d4 100644 --- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataFileListStatsTest.java +++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataFileListStatsTest.java @@ -21,6 +21,7 @@ import org.apache.doris.connector.hms.HmsDatabaseInfo; import org.apache.doris.connector.hms.HmsPartitionInfo; import org.apache.doris.connector.hms.HmsTableInfo; +import org.apache.doris.connector.spi.ConnectorSession; import org.apache.doris.filesystem.FileSystem; import org.junit.jupiter.api.Assertions; @@ -95,6 +96,68 @@ public void sampledPartitionsAreScaledUpToTheWholeTable() { Assertions.assertEquals(400L, metadata(client).estimateDataSize(partitioned(), 2, (loc, vals) -> 100)); } + @Test + public void selectedPartitionEstimateOnlyListsTheSelectedRange() { + PartitionFakeHmsClient client = new PartitionFakeHmsClient(Arrays.asList("p0", "p1", "p2")); + long size = metadata(client).estimateSelectedPartitionsDataSize( + partitioned(), Collections.singletonList("p1"), 30, + (loc, vals) -> loc.endsWith("p1") ? 200 : 10_000); + + Assertions.assertEquals(200L, size); + } + + @Test + public void selectedPartitionSamplingScalesWithinTheSelectedRange() { + PartitionFakeHmsClient client = new PartitionFakeHmsClient( + Arrays.asList("p0", "p1", "p2", "p3", "p4")); + long size = metadata(client).estimateSelectedPartitionsDataSize( + partitioned(), Arrays.asList("p1", "p2", "p3", "p4"), 2, + (loc, vals) -> 100); + + Assertions.assertEquals(400L, size); + Assertions.assertEquals(2, client.lastRequestedPartitionNames.size()); + Assertions.assertTrue(Arrays.asList("p1", "p2", "p3", "p4") + .containsAll(client.lastRequestedPartitionNames)); + } + + @Test + public void zeroSizedSampleDoesNotProveTheWholeSelectedRangeIsEmpty() { + PartitionFakeHmsClient client = new PartitionFakeHmsClient( + Arrays.asList("p0", "p1", "p2", "p3")); + long size = metadata(client).estimateSelectedPartitionsDataSize( + partitioned(), Arrays.asList("p0", "p1", "p2", "p3"), 2, + (loc, vals) -> 0); + + Assertions.assertEquals(-1L, size); + } + + @Test + public void fullyInspectedEmptySelectedPartitionsHaveZeroDataSize() { + PartitionFakeHmsClient client = new PartitionFakeHmsClient(Arrays.asList("p0", "p1")); + long size = metadata(client).estimateSelectedPartitionsDataSize( + partitioned(), Arrays.asList("p0", "p1"), 30, + (loc, vals) -> 0); + + Assertions.assertEquals(0L, size); + } + + @Test + public void emptySelectedPartitionRangeHasZeroDataSize() { + PartitionFakeHmsClient client = new PartitionFakeHmsClient(Collections.emptyList()); + Assertions.assertEquals(0L, metadata(client).estimateSelectedPartitionsDataSize( + partitioned(), Collections.emptyList(), 30, (loc, vals) -> 100)); + Assertions.assertTrue(client.lastRequestedPartitionNames.isEmpty()); + } + + @Test + public void missingSelectedPartitionMakesTheEstimateUnknown() { + PartitionFakeHmsClient client = new PartitionFakeHmsClient(Arrays.asList("p0", "p1")); + client.removeExistingPartition("p1"); + + Assertions.assertEquals(-1L, metadata(client).estimateSelectedPartitionsDataSize( + partitioned(), Arrays.asList("p0", "p1"), 30, (loc, vals) -> 100)); + } + @Test public void zeroTotalSizeReturnsMinusOne() { PartitionFakeHmsClient client = new PartitionFakeHmsClient(Arrays.asList("p0", "p1")); @@ -193,6 +256,27 @@ public void nonHiveTableTypeIsNotEstimated() { .estimateDataSizeByListingFiles(null, hudiHandle)); } + @Test + public void transactionalHiveSelectedPartitionsAreNotEstimated() { + HiveTableHandle transactionalHandle = new HiveTableHandle.Builder("db", "t", HiveTableType.HIVE) + .partitionKeyNames(Collections.singletonList("dt")) + .tableParameters(Collections.singletonMap("transactional", "true")) + .build(); + PartitionFakeHmsClient client = new PartitionFakeHmsClient(Collections.singletonList("p0")); + FakeConnectorContext context = new FakeConnectorContext() { + @Override + public FileSystem getFileSystem(ConnectorSession session) { + return new FakeFileSystem(); + } + }; + HiveConnectorMetadata metadata = new HiveConnectorMetadata( + client, HiveTestProperties.minimal(), context); + + Assertions.assertEquals(-1L, metadata.estimateDataSizeByListingFiles( + null, transactionalHandle, Collections.singletonList("p0"))); + Assertions.assertTrue(client.lastRequestedPartitionNames.isEmpty()); + } + /** A {@link HiveFileListingCache} whose listing always fails, to prove listFileSizes propagates (not swallows). */ private static final class ThrowingFileListingCache extends HiveFileListingCache { ThrowingFileListingCache() { @@ -214,6 +298,8 @@ public List listDataFiles(String dbName, String tableName, Strin private static final class PartitionFakeHmsClient implements HmsClient { private final List partitionNames; private final java.util.Set withoutLocation = new java.util.HashSet<>(); + private final java.util.Set missingPartitions = new java.util.HashSet<>(); + private List lastRequestedPartitionNames = Collections.emptyList(); PartitionFakeHmsClient(List partitionNames) { this.partitionNames = partitionNames; @@ -223,6 +309,10 @@ void dropLocationFor(String name) { withoutLocation.add(name); } + void removeExistingPartition(String name) { + missingPartitions.add(name); + } + @Override public List listPartitionNames(String dbName, String tableName, int maxParts) { return partitionNames; @@ -239,6 +329,19 @@ public List getPartitions(String dbName, String tableName, Lis return result; } + @Override + public List getExistingPartitions( + String dbName, String tableName, List partNames) { + lastRequestedPartitionNames = new ArrayList<>(partNames); + List existingNames = new ArrayList<>(); + for (String name : partNames) { + if (!missingPartitions.contains(name)) { + existingNames.add(name); + } + } + return getPartitions(dbName, tableName, existingNames); + } + @Override public HmsTableInfo getTable(String dbName, String tableName) { throw new UnsupportedOperationException(); diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java index f086b010fc6352..b789664ba74114 100644 --- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java +++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java @@ -241,6 +241,8 @@ public void everyPerHandleMethodForwardsAForeignHandleToTheSibling() { md.getTableStatistics(session, foreignHandle); md.getColumnStatistics(session, foreignHandle, "c"); long size = md.estimateDataSizeByListingFiles(session, foreignHandle); + long selectedSize = md.estimateDataSizeByListingFiles( + session, foreignHandle, Collections.singletonList("p")); Optional> filter = md.applyFilter(session, foreignHandle, null); List partNames = md.listPartitionNames(session, foreignHandle); md.listPartitions(session, foreignHandle, Optional.empty()); @@ -278,6 +280,9 @@ public void everyPerHandleMethodForwardsAForeignHandleToTheSibling() { // A few return values prove the ANSWER is the sibling's, not hive's default. Assertions.assertEquals(RecordingSiblingMetadata.SENTINEL_SIZE, size, "estimateDataSize must return the sibling's value, not hive's -1"); + Assertions.assertEquals(RecordingSiblingMetadata.SENTINEL_SIZE, selectedSize, + "selected-partition estimateDataSize must return the sibling's value, not hive's -1"); + Assertions.assertEquals(Collections.singletonList("p"), siblingMetadata.selectedPartitionNames); Assertions.assertEquals(RecordingSiblingMetadata.SENTINEL_SNAPSHOT_ID, pin.getSnapshotId(), "beginQuerySnapshot must return the sibling's snapshot-id pin, not hive's -1 last-modified pin"); Assertions.assertEquals(Collections.singletonMap("p", 55L), partitionFreshness, @@ -726,7 +731,7 @@ private static final class RecordingSiblingMetadata implements ConnectorMetadata // dropping a guard, or adding one that should not forward, changes this list and fails the test). static final List EXPECTED_METHODS = Collections.unmodifiableList(Arrays.asList( "getTableSchema", "getColumnHandles", "getTableStatistics", "getColumnStatistics", - "estimateDataSizeByListingFiles", + "estimateDataSizeByListingFiles", "estimateSelectedDataSizeByListingFiles", "applyFilter", "listPartitionNames", "listPartitions", "beginQuerySnapshot", "getTableFreshness", "getPartitionFreshnessMillis", "getPartitionsFreshnessMillis", "dropTable", @@ -747,6 +752,7 @@ private static final class RecordingSiblingMetadata implements ConnectorMetadata "validateRowLevelDmlMode", "validateStaticPartitionColumns", "validateWritePartitionNames")); final List calls = new ArrayList<>(); + List selectedPartitionNames = Collections.emptyList(); final Optional> filterResult = Optional.of(new FilterApplicationResult<>(SIBLING_HANDLE, null, false)); @@ -792,6 +798,14 @@ public long estimateDataSizeByListingFiles(ConnectorSession session, ConnectorTa return SENTINEL_SIZE; } + @Override + public long estimateDataSizeByListingFiles(ConnectorSession session, ConnectorTableHandle handle, + List selectedPartitionNames) { + calls.add("estimateSelectedDataSizeByListingFiles"); + this.selectedPartitionNames = new ArrayList<>(selectedPartitionNames); + return SENTINEL_SIZE; + } + @Override public Optional> applyFilter(ConnectorSession session, diff --git a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatisticsOps.java b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatisticsOps.java index d602504afaa577..ec1cba8a0af07a 100644 --- a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatisticsOps.java +++ b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatisticsOps.java @@ -98,6 +98,23 @@ default long estimateDataSizeByListingFiles(ConnectorSession session, ConnectorT return -1; } + /** + * Estimates the on-disk data size in bytes for exactly the named partitions. The names are connector + * partition identifiers returned to the engine; an empty list means that pruning selected no partitions. + * Returns -1 when the connector cannot estimate this exact partition set. Zero is a valid result and means + * the selected partitions contain no data files. The supplied handle already carries any snapshot pin for + * this scan; an implementation that cannot honor that exact scope must return -1. + * + *

The default must not delegate to the whole-table overload because that would silently substitute an + * incompatible row-count scope.

+ */ + default long estimateDataSizeByListingFiles( + ConnectorSession session, + ConnectorTableHandle handle, + List selectedPartitionNames) { + return -1; + } + /** * Returns the RAW byte length of every data file across ALL partitions of the table (not sampled, not summed), * for {@code ANALYZE ... WITH SAMPLE}: fe-core seed-shuffles and cumulates these sizes to a sample scale diff --git a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java index a2b20ae311d973..35ed8d90a83794 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java +++ b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java @@ -87,9 +87,9 @@ public void connectorApiMajorTracksTheRecordedSurfaceChange() throws IOException Assertions.assertNotNull(in, "missing connector plugin API version resource"); version.load(in); } - // Major 12 adds the SUPPORTS_FIELD_ID_ACCESS_PATH and SUPPORTS_SYS_TABLE_NESTED_COLUMN_PRUNE - // capabilities: a plugin naming either constant cannot link against an older FE. - Assertions.assertEquals("12.0", version.getProperty("api.version")); + // Major 13 在 major 12 的接口基础上新增所选分区的数据量估算重载。 + // 旧 FE 必须在链接不兼容字节码前拒绝使用该重载的插件。 + Assertions.assertEquals("13.0", version.getProperty("api.version")); } /** Root entry points plus provider/handle types returned to connector plugins. */ diff --git a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt index b69475f19975ec..38e12f954b5d10 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt +++ b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt @@ -26,6 +26,7 @@ dropTable(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.conne dropTag(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.handle.ConnectorTableHandle,org.apache.doris.connector.spi.ddl.DropRefChange) dropView(org.apache.doris.connector.spi.ConnectorSession,java.lang.String,java.lang.String) estimateDataSizeByListingFiles(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.handle.ConnectorTableHandle) +estimateDataSizeByListingFiles(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.handle.ConnectorTableHandle,java.util.List) fromRemoteColumnName(org.apache.doris.connector.spi.ConnectorSession,java.lang.String,java.lang.String,java.lang.String) fromRemoteDatabaseName(org.apache.doris.connector.spi.ConnectorSession,java.lang.String) fromRemoteTableName(org.apache.doris.connector.spi.ConnectorSession,java.lang.String,java.lang.String) diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml index 121868e3dd830f..9bf53c1050bfa3 100644 --- a/fe/fe-connector/pom.xml +++ b/fe/fe-connector/pom.xml @@ -55,7 +55,7 @@ under the License. of the latter two means bumping this property as well (and fe-extension-spi means bumping all five families). --> - 12.0 + 13.0 diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java index 5229c6fc55d39a..2e747fa2ea4ba3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java @@ -1949,6 +1949,36 @@ public long getRowCountForPartitionIndex(long partitionId, long indexId, boolean return index.getRowCount() == -1 ? 0 : index.getRowCount(); } + /** + * Returns the row count of the selected partitions for the specified index. If some selected + * partitions have not reported row counts, distribute the table row count not accounted for by + * known selected partitions across the remaining partitions. + */ + public double getRowCountForSelectedPartitions(Collection selectedPartitionIds, + long indexId, double tableRowCount) { + int unknownSelectedPartitionCount = 0; + double knownSelectedPartitionRowCount = 0; + for (long partitionId : selectedPartitionIds) { + long partitionRowCount = getRowCountForPartitionIndex(partitionId, indexId, true); + if (partitionRowCount == UNKNOWN_ROW_COUNT) { + unknownSelectedPartitionCount++; + } else { + knownSelectedPartitionRowCount += partitionRowCount; + } + } + if (unknownSelectedPartitionCount == 0) { + return knownSelectedPartitionRowCount; + } + + int knownSelectedPartitionCount = selectedPartitionIds.size() - unknownSelectedPartitionCount; + int remainingPartitionCount = getPartitionNum() - knownSelectedPartitionCount; + Preconditions.checkArgument(remainingPartitionCount > 0, + "selected partitions with unknown row count should not cover all table partitions"); + double remainingRowCount = Math.max(0, tableRowCount - knownSelectedPartitionRowCount); + return knownSelectedPartitionRowCount + + remainingRowCount * unknownSelectedPartitionCount / remainingPartitionCount; + } + @Override public long getAvgRowLength() { return getTableStatusStats().getAvgRowLength(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java index 11798d13bae629..a828f547cea0c9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java @@ -276,6 +276,12 @@ public long getRowCount() { .getCachedRowCount(catalog.getId(), dbId, id, true); } + /** Returns the row count for a pruned external partition set, or UNKNOWN when unsupported. */ + public long getRowCountForSelectedPartitions(SelectedPartitions selectedPartitions, + Optional snapshot) { + return UNKNOWN_ROW_COUNT; + } + @Override public long getCachedRowCount() { // Return -1 if uninitialized. diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java index da7a7f8c0e35eb..c417160cf17861 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java @@ -55,6 +55,7 @@ import org.apache.doris.datasource.systable.PartitionsSysTable; import org.apache.doris.datasource.systable.PluginDrivenSysTable; import org.apache.doris.datasource.systable.SysTable; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.GlobalVariable; import org.apache.doris.statistics.analysis.AnalysisInfo; @@ -1446,6 +1447,56 @@ public long fetchRowCount() { } } + @Override + public long getRowCountForSelectedPartitions(SelectedPartitions selectedPartitions, + Optional snapshot) { + ConnectContext context = ConnectContext.get(); + if (!GlobalVariable.enable_get_row_count_from_file_list + || !selectedPartitions.isPruned + || selectedPartitions.totalPartitionNum == 0 + || selectedPartitions.selectedPartitions.size() >= selectedPartitions.totalPartitionNum + || context != null && context.getStatementContext() != null + && context.getStatementContext().hasAnyPlanReadLockTable()) { + // Selected-partition estimation can synchronously load HMS metadata and file listings. A mixed + // internal/external query derives statistics while holding its internal-table read locks, so it must + // fall back to the existing table-level estimate rather than add selected-partition remote I/O + // in that critical section. + return UNKNOWN_ROW_COUNT; + } + + try { + makeSureInitialized(); + PluginDrivenExternalCatalog pluginCatalog = (PluginDrivenExternalCatalog) catalog; + Connector connector = pluginCatalog.getConnector(); + ConnectorSession session = pluginCatalog.buildConnectorSession(); + ConnectorMetadata metadata = PluginDrivenMetadata.get(session, connector); + Optional handleOpt = resolveConnectorTableHandle(session, metadata); + if (!handleOpt.isPresent()) { + return UNKNOWN_ROW_COUNT; + } + + ConnectorTableHandle handle = handleOpt.get(); + if (snapshot.isPresent() && snapshot.get() instanceof PluginDrivenMvccSnapshot) { + ConnectorMvccSnapshot connectorSnapshot = + ((PluginDrivenMvccSnapshot) snapshot.get()).getConnectorSnapshot(); + handle = metadata.applySnapshot(session, handle, connectorSnapshot); + } + long dataSize = metadata.estimateDataSizeByListingFiles( + session, handle, new ArrayList<>(selectedPartitions.selectedPartitions.keySet())); + if (dataSize < 0) { + return UNKNOWN_ROW_COUNT; + } + if (dataSize == 0) { + return 0; + } + long rowWidth = estimatedRowWidth(true, snapshot); + return rowWidth > 0 ? dataSize / rowWidth : UNKNOWN_ROW_COUNT; + } catch (RuntimeException e) { + LOG.debug("Failed to estimate selected partition row count for table {}", name, e); + return UNKNOWN_ROW_COUNT; + } + } + private long fetchRowCount(ConnectorSession session, ConnectorMetadata metadata) { Optional handleOpt = resolveConnectorTableHandle(session, metadata); if (!handleOpt.isPresent()) { @@ -1579,6 +1630,22 @@ private long estimatedRowWidth(boolean excludePartitionColumns) { return rowWidth; } + private long estimatedRowWidth(boolean excludePartitionColumns, Optional snapshot) { + List schema = getFullSchema(snapshot); + if (schema == null) { + return 0; + } + List partitionColumns = excludePartitionColumns ? getPartitionColumns(snapshot) : null; + long rowWidth = 0; + for (Column column : schema) { + if (partitionColumns != null && partitionColumns.contains(column)) { + continue; + } + rowWidth += column.getDataType().getSlotSize(); + } + return rowWidth; + } + /** * The engine name shown in the {@code ENGINE} column of {@code SHOW TABLE STATUS} and * {@code information_schema.tables} (and through the REST metadata API). Named by the connector, which diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PrunePartitionPredicate.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PrunePartitionPredicate.java index 01ecf88e595ac4..bda887f6f7fde9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PrunePartitionPredicate.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PrunePartitionPredicate.java @@ -17,26 +17,15 @@ package org.apache.doris.nereids.processor.post; -import org.apache.doris.analysis.Expr; -import org.apache.doris.analysis.SlotRef; -import org.apache.doris.catalog.Column; -import org.apache.doris.catalog.OlapTable; import org.apache.doris.nereids.CascadesContext; import org.apache.doris.nereids.trees.expressions.Expression; -import org.apache.doris.nereids.trees.expressions.Slot; -import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; import org.apache.doris.nereids.trees.plans.Plan; import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan; import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter; import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan; -import org.apache.doris.nereids.util.ExpressionUtils; -import java.util.HashMap; -import java.util.HashSet; import java.util.LinkedHashSet; -import java.util.List; -import java.util.Map; import java.util.Optional; import java.util.Set; @@ -67,25 +56,9 @@ public Plan visitPhysicalFilter(PhysicalFilter filter, CascadesC if (skipPrunePredicate) { return filter; } - Set scanPartitions = new HashSet<>(scan.getSelectedPartitionIds()); - Map nameToOutputSlot = buildNameToSlotMap(scan); - Set remaining = new LinkedHashSet<>(filter.getConjuncts()); - boolean changed = false; PartitionPrunablePredicate entry = entryOpt.get(); - if (entry.getSelectedPartitionIds().containsAll(scanPartitions)) { - Map slotReplaceMap = - buildSlotReplaceMap(entry.getSnapshotPartitionSlots(), nameToOutputSlot); - if (slotReplaceMap != null) { - for (Expression conjunct : entry.getPrunableConjuncts()) { - Expression rewritten = slotReplaceMap.isEmpty() - ? conjunct : ExpressionUtils.replace(conjunct, slotReplaceMap); - if (remaining.remove(rewritten)) { - changed = true; - } - } - } - } + boolean changed = remaining.removeAll(entry.getPrunableConjuncts()); if (!changed) { return filter; } @@ -96,51 +69,4 @@ public Plan visitPhysicalFilter(PhysicalFilter filter, CascadesC .copyStatsAndGroupIdFrom((AbstractPhysicalPlan) filter); } - private static Map buildNameToSlotMap(PhysicalOlapScan scan) { - OlapTable table = scan.getTable(); - List slots = scan.getOutput(); - Map map = new HashMap<>(slots.size()); - if (scan.getSelectedIndexId() == table.getBaseIndexId()) { - for (Slot slot : slots) { - map.put(slot.getName().toLowerCase(), slot); - } - } else { - for (Slot slot : slots) { - if (!(slot instanceof SlotReference)) { - continue; - } - SlotReference slotReference = (SlotReference) slot; - Optional columnOptional = slotReference.getOriginalColumn(); - if (!columnOptional.isPresent()) { - continue; - } - Expr expr = columnOptional.get().getDefineExpr(); - if (!(expr instanceof SlotRef)) { - continue; - } - map.put(((SlotRef) expr).getColumnName().toLowerCase(), slot); - } - } - return map; - } - - /** - * Map each recorded snapshot slot to the scan's current output slot of the - * same column name. Returns null when any snapshot slot cannot be located, - * so the caller can skip the entry. - */ - private static Map buildSlotReplaceMap( - List snapshotSlots, Map nameToOutputSlot) { - Map replaceMap = new HashMap<>(snapshotSlots.size()); - for (Slot snapshot : snapshotSlots) { - Slot current = nameToOutputSlot.get(snapshot.getName().toLowerCase()); - if (current == null) { - return null; - } - if (!snapshot.equals(current)) { - replaceMap.put(snapshot, current); - } - } - return replaceMap; - } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/SyncMaterializationContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/SyncMaterializationContext.java index 22d91344bfde6c..df2339d936223d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/SyncMaterializationContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/SyncMaterializationContext.java @@ -125,15 +125,11 @@ public Plan visitLogicalOlapScan(LogicalOlapScan olapScan, Void context) { LogicalOlapScan queryScan = (LogicalOlapScan) queryStructInfoRelations.get(0); if (!queryScan.getTable().getFullQualifiers().equals( olapScan.getTable().getFullQualifiers())) { - // Only the same table, we can do partition prue + // Only scans of the same table can share a partition-pruning result. return olapScan; } - // Carry partition-prunable predicates from the original query scan onto - // the rewritten MV scan so the post-processor can still drop the - // predicates that have already been enforced by partition pruning. - return olapScan - .withSelectedPartitionIds(queryScan.getSelectedPartitionIds()) - .withPartitionPrunablePredicates(queryScan.getPartitionPrunablePredicates()); + // Copy the selected partitions and rebind the pruning proof to the rewritten scan's slots. + return olapScan.withPartitionPruningFrom(queryScan); } }, null); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/PartitionPruner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/PartitionPruner.java index 09fbca12c73270..9cfcc6c8757c93 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/PartitionPruner.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/PartitionPruner.java @@ -196,9 +196,10 @@ private static > PartitionPruneResult pruneInternal( partitionPredicate = OrToIn.EXTRACT_MODE_INSTANCE.rewriteTree( partitionPredicate, new ExpressionRewriteContext(cascadesContext)); if (BooleanLiteral.TRUE.equals(partitionPredicate)) { - // The partition column predicate is always true and can be deleted, the partition cannot be pruned + // No effective partition predicate is available. Keep all partitions and do not record + // an applied predicate. return new PartitionPruneResult<>(Utils.fastToImmutableList(idToPartitions.keySet()), - Optional.of(originalPartitionPredicate), false); + Optional.empty(), false); } else if (BooleanLiteral.FALSE.equals(partitionPredicate) || partitionPredicate.isNullLiteral()) { // The partition column predicate is always false, and all partitions can be pruned. return new PartitionPruneResult<>(ImmutableList.of(), Optional.empty(), true); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java index b3b121053c7913..62c24b82cbe496 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java @@ -30,9 +30,12 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.util.ExpressionUtils; import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; import com.google.common.collect.Maps; import com.google.common.collect.Sets; import org.apache.commons.collections4.CollectionUtils; @@ -84,7 +87,7 @@ private SelectedPartitions pruneExternalPartitions(ExternalTable externalTable, ctx.getStatementContext().getSnapshot(externalTable, scan.getTableSnapshot(), scan.getScanParams())))) { // non partitioned table, return NOT_PRUNED. - // non partition table will be handled in HiveScanNode. + // non partition table will be handled in PluginDrivenScanNode. return SelectedPartitions.NOT_PRUNED; } Map scanOutput = scan.getOutput() @@ -139,7 +142,12 @@ private SelectedPartitions pruneExternalPartitions(ExternalTable externalTable, "pruned partition %s is missing in the selected partitions snapshot", name); selectedPartitionItems.put(name, item); } - return new SelectedPartitions(nameToPartitionItem.size(), selectedPartitionItems, true, - result.hasPartitionPredicate); + return scan.getSelectedPartitions().withPruneResult( + selectedPartitionItems, + result.hasPartitionPredicate, + result.prunedPartitionPredicate.isPresent() ? partitionSlots : ImmutableList.of(), + result.prunedPartitionPredicate + .map(ExpressionUtils::extractConjunctionToSet) + .orElse(ImmutableSet.of())); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartition.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartition.java index c89234b92fe3f8..64399d32aa4d14 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartition.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartition.java @@ -96,9 +96,7 @@ public List buildRules() { if (rewrittenLogicalRelation instanceof LogicalEmptyRelation) { return rewrittenLogicalRelation; } - boolean skipPrunePredicate = ctx.connectContext.getSessionVariable().skipPrunePredicate - || ctx.statementContext.isDelete(); - if (!skipPrunePredicate && prunedRes.second.isPresent()) { + if (prunedRes.second.isPresent()) { // Defer the predicate removal to PlanPostProcessor so that materialized-view // rewrite still sees the original predicates. Otherwise, partition predicates // that are equivalent to the surviving partition list would be silently diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/StatsDerive.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/StatsDerive.java index 4decb15d6e32fa..61c8361164777e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/StatsDerive.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/StatsDerive.java @@ -31,6 +31,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation; import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation; import org.apache.doris.nereids.trees.plans.logical.LogicalExcept; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; import org.apache.doris.nereids.trees.plans.logical.LogicalIntersect; @@ -179,6 +180,16 @@ public Statistics visitLogicalOlapScan(LogicalOlapScan olapScan, DeriveContext c return stats; } + @Override + public Statistics visitLogicalFileScan(LogicalFileScan fileScan, DeriveContext context) { + Statistics stats = fileScan.getStats(); + if (stats == null || deepDerive) { + stats = context.calculator.computeFileScan(fileScan); + fileScan.setStatistics(stats); + } + return stats; + } + @Override public Statistics visitLogicalCatalogRelation(LogicalCatalogRelation relation, DeriveContext context) { Statistics stats = relation.getStats(); @@ -393,4 +404,3 @@ public Statistics visitLogicalRelation(LogicalRelation relation, DeriveContext c } - diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/FilterEstimation.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/FilterEstimation.java index 5507d42c3186de..49625b4e1c6d34 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/FilterEstimation.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/FilterEstimation.java @@ -41,6 +41,7 @@ import org.apache.doris.nereids.trees.expressions.Slot; import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.trees.expressions.functions.Function; +import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral; import org.apache.doris.nereids.trees.expressions.literal.ComparableLiteral; import org.apache.doris.nereids.trees.expressions.literal.DateTimeLiteral; import org.apache.doris.nereids.trees.expressions.literal.Literal; @@ -126,6 +127,13 @@ public Statistics visit(Expression expr, EstimationContext context) { return context.statistics.withSel(DEFAULT_INEQUALITY_COEFFICIENT); } + @Override + public Statistics visitBooleanLiteral(BooleanLiteral booleanLiteral, EstimationContext context) { + return new StatisticsBuilder(context.statistics) + .setRowCount(booleanLiteral.getValue() ? context.statistics.getRowCount() : 0) + .build(); + } + @Override public Statistics visitAnd(And and, EstimationContext context) { List children = and.children(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java index e3071cfd25fbc3..ef4ee2e7a99049 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java @@ -32,6 +32,9 @@ import org.apache.doris.catalog.TableIf; import org.apache.doris.common.FeConstants; import org.apache.doris.common.Pair; +import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.datasource.mvcc.MvccSnapshot; +import org.apache.doris.datasource.mvcc.MvccUtil; import org.apache.doris.nereids.CascadesContext; import org.apache.doris.nereids.memo.Group; import org.apache.doris.nereids.memo.GroupExpression; @@ -49,6 +52,7 @@ import org.apache.doris.nereids.trees.expressions.functions.agg.Min; import org.apache.doris.nereids.trees.expressions.literal.Literal; import org.apache.doris.nereids.trees.plans.GroupPlan; +import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; import org.apache.doris.nereids.trees.plans.Plan; import org.apache.doris.nereids.trees.plans.algebra.Aggregate; import org.apache.doris.nereids.trees.plans.algebra.CatalogRelation; @@ -76,6 +80,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation; import org.apache.doris.nereids.trees.plans.logical.LogicalExcept; import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; import org.apache.doris.nereids.trees.plans.logical.LogicalIntersect; @@ -135,6 +140,7 @@ import org.apache.doris.nereids.trees.plans.physical.PhysicalWorkTableReference; import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor; import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.util.ExpressionUtils; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.SessionVariable; import org.apache.doris.statistics.StatisticConstants; @@ -162,6 +168,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -401,28 +408,6 @@ private ColumnStatistic getColumnStatsFromTableCache(CatalogRelation catalogRela return getColumnStatistic(catalogRelation.getTable(), slot.getName(), idxId); } - private double getSelectedPartitionRowCount(OlapScan olapScan, double tableRowCount) { - // the number of partitions whose row count is not available - double unknownPartitionCount = 0; - double partRowCountSum = 0; - for (long id : olapScan.getSelectedPartitionIds()) { - long partRowCount = olapScan.getTable() - .getRowCountForPartitionIndex(id, olapScan.getSelectedIndexId(), true); - if (partRowCount == -1) { - unknownPartitionCount++; - } else { - partRowCountSum += partRowCount; - } - } - // estimate row count for unknownPartitionCount - if (unknownPartitionCount > 0) { - // each selected partition has at least one row - partRowCountSum += Math.max(unknownPartitionCount, - tableRowCount * unknownPartitionCount / olapScan.getTable().getPartitionNum()); - } - return partRowCountSum; - } - private void setHasUnknownColStatsInStatementContext() { if (ConnectContext.get() != null && ConnectContext.get().getStatementContext() != null) { ConnectContext.get().getStatementContext().setHasUnknownColStats(true); @@ -512,7 +497,8 @@ public Statistics computeOlapScan(OlapScan olapScan) { LOG.info("computeOlapScan optStats isPresent {}, tableRowCount is {}, table name is {}", optStats.isPresent(), tableRowCount, olapTable.getQualifiedName()); if (optStats.isPresent()) { - double selectedPartitionsRowCount = getSelectedPartitionRowCount(olapScan, tableRowCount); + double selectedPartitionsRowCount = olapTable.getRowCountForSelectedPartitions( + olapScan.getSelectedPartitionIds(), olapScan.getSelectedIndexId(), tableRowCount); if (isRegisteredRowCount(olapScan)) { // If a row count is injected for the materialized view, use it to fix the issue where // the materialized view cannot be selected by cbo stable due to selectedPartitionsRowCount being 0, @@ -579,7 +565,8 @@ public Statistics computeOlapScan(OlapScan olapScan) { && olapScan.getSelectedPartitionIds().size() < olapScan.getTable().getPartitionNum()) { // partition pruned // try to use selected partition stats, if failed, fall back to table stats - double selectedPartitionsRowCount = getSelectedPartitionRowCount(olapScan, tableRowCount); + double selectedPartitionsRowCount = olapTable.getRowCountForSelectedPartitions( + olapScan.getSelectedPartitionIds(), olapScan.getSelectedIndexId(), tableRowCount); List selectedPartitionNames = new ArrayList<>(olapScan.getSelectedPartitionIds().size()); olapScan.getSelectedPartitionIds().forEach(id -> { selectedPartitionNames.add(olapScan.getTable().getPartition(id).getName()); @@ -606,6 +593,7 @@ public Statistics computeOlapScan(OlapScan olapScan) { } checkIfUnknownStatsUsedAsKey(builder); builder.setRowCount(selectedPartitionsRowCount); + builder.setConjunctsAppliedToRowCount(getPrunableConjuncts(olapScan)); } else { // get table level stats for (SlotReference slot : visibleOutputSlots) { @@ -616,10 +604,20 @@ public Statistics computeOlapScan(OlapScan olapScan) { } checkIfUnknownStatsUsedAsKey(builder); builder.setRowCount(tableRowCount); + if (olapScan.getSelectedPartitionIds().size() >= olapScan.getTable().getPartitionNum()) { + builder.setConjunctsAppliedToRowCount(getPrunableConjuncts(olapScan)); + } } return computeVirtualColumnStats(olapScan, builder.build()); } + private Set getPrunableConjuncts(OlapScan olapScan) { + Optional prunablePredicate = olapScan.getPartitionPrunablePredicates(); + return prunablePredicate + .map(predicate -> predicate.getPrunableConjuncts()) + .orElse(ImmutableSet.of()); + } + private Statistics computeVirtualColumnStats(OlapScan relation, Statistics stats) { List virtualColumns; if (relation instanceof LogicalOlapScan) { @@ -816,7 +814,7 @@ public Statistics visitLogicalSchemaScan(LogicalSchemaScan schemaScan, Void cont @Override public Statistics visitLogicalFileScan(LogicalFileScan fileScan, Void context) { - return computeCatalogRelation(fileScan); + return computeFileScan(fileScan); } @Override @@ -1020,7 +1018,7 @@ public Statistics visitPhysicalWorkTableReference(PhysicalWorkTableReference wor @Override public Statistics visitPhysicalFileScan(PhysicalFileScan fileScan, Void context) { - return computeCatalogRelation(fileScan); + return computeFileScan(fileScan); } @Override @@ -1171,7 +1169,90 @@ public Statistics computeAssertNumRows(AssertNumRowsElement assertNumRowsElement * computeFilter */ public Statistics computeFilter(Filter filter, Statistics inputStats) { - return new FilterEstimation().estimate(filter.getPredicate(), inputStats); + Set appliedConjuncts = new LinkedHashSet<>(filter.getConjuncts()); + appliedConjuncts.retainAll(inputStats.getConjunctsAppliedToRowCount()); + Statistics inputWithConstrainedColumns = constrainColumnsWithoutReducingRows(appliedConjuncts, inputStats); + + Set remainingConjuncts = new LinkedHashSet<>(filter.getConjuncts()); + remainingConjuncts.removeAll(appliedConjuncts); + return new FilterEstimation().estimate(ExpressionUtils.and(remainingConjuncts), + inputWithConstrainedColumns); + } + + private Statistics constrainColumnsWithoutReducingRows(Set conjuncts, Statistics inputStats) { + Map> conjunctsBySlot = new LinkedHashMap<>(); + for (Expression conjunct : conjuncts) { + Set inputSlots = conjunct.getInputSlots(); + if (inputSlots.size() != 1) { + // A relationship between multiple partition columns (for example a = b) does not bound either + // column independently. FilterEstimation derives a reduced row count for such predicates and then + // caps both NDVs to it, which is not valid after restoring the scan cardinality. + continue; + } + Slot slot = inputSlots.iterator().next(); + conjunctsBySlot.computeIfAbsent(slot, ignored -> new ArrayList<>()).add(conjunct); + } + + StatisticsBuilder constrainedBuilder = new StatisticsBuilder(inputStats); + for (Map.Entry> entry : conjunctsBySlot.entrySet()) { + Slot slot = entry.getKey(); + ColumnStatistic currentColumnStats = inputStats.findColumnStatistics(slot); + if (currentColumnStats == null) { + continue; + } + + // The scan row count may already be much smaller than the table-level NDV. Estimate all constraints + // on this slot against one stable base, then restore the selected row count once. When OLAP partition + // metadata has already narrowed min/max, retain those proven bounds while borrowing the uncapped NDV + // and row count from the original table statistic. + ColumnStatistic estimationColumnStats = currentColumnStats; + double estimationRowCount = inputStats.getRowCount(); + if (currentColumnStats.getOriginal() != null) { + ColumnStatisticBuilder estimationColumnBuilder = + new ColumnStatisticBuilder(currentColumnStats.getOriginal()); + if (Double.isFinite(currentColumnStats.minValue)) { + estimationColumnBuilder.setMinValue(currentColumnStats.minValue) + .setMinExpr(currentColumnStats.minExpr); + } + if (Double.isFinite(currentColumnStats.maxValue)) { + estimationColumnBuilder.setMaxValue(currentColumnStats.maxValue) + .setMaxExpr(currentColumnStats.maxExpr); + } + estimationColumnStats = estimationColumnBuilder.build(); + estimationRowCount = Math.max(estimationRowCount, estimationColumnStats.count); + } + Statistics estimated = new StatisticsBuilder() + .setRowCount(estimationRowCount) + .putColumnStatistics(slot, estimationColumnStats) + .build(); + for (Expression conjunct : entry.getValue()) { + estimated = new FilterEstimation().estimate(conjunct, estimated); + } + if (estimated.getRowCount() <= 0) { + continue; + } + ColumnStatistic columnStatistic = estimated.findColumnStatistics(slot); + if (columnStatistic == null) { + continue; + } + + ColumnStatisticBuilder columnBuilder = new ColumnStatisticBuilder( + columnStatistic, inputStats.getRowCount()); + if (!columnStatistic.isUnKnown) { + double restoredNdv = Math.min(columnStatistic.ndv, inputStats.getRowCount()); + columnBuilder.setNdv(restoredNdv); + if (columnStatistic.numNulls >= 0) { + double restoredNulls = columnStatistic.numNulls; + if (estimated.getRowCount() > 0) { + restoredNulls *= inputStats.getRowCount() / estimated.getRowCount(); + } + columnBuilder.setNumNulls(Math.max(0, Math.min( + restoredNulls, inputStats.getRowCount() - restoredNdv))); + } + } + constrainedBuilder.putColumnStatistics(slot, columnBuilder.build()); + } + return constrainedBuilder.build(); } private ColumnStatistic getColumnStatistic(TableIf table, String colName, long idxId) { @@ -1248,17 +1329,55 @@ private ColumnStatistic getColumnStatistic( return olapTableStatistics.getColumnStatistics(colName, connectContext); } + public Statistics computeFileScan(LogicalFileScan fileScan) { + Optional snapshot = MvccUtil.getSnapshotFromContext( + fileScan.getTable(), fileScan.getTableSnapshot(), fileScan.getScanParams()); + return buildFileScanStats(fileScan, fileScan.getTable(), fileScan.getSelectedPartitions(), snapshot); + } + + public Statistics computeFileScan(PhysicalFileScan fileScan) { + Optional snapshot = MvccUtil.getSnapshotFromContext( + fileScan.getTable(), fileScan.getTableSnapshot(), fileScan.getScanParams()); + return buildFileScanStats(fileScan, fileScan.getTable(), fileScan.getSelectedPartitions(), snapshot); + } + + private Statistics buildFileScanStats(CatalogRelation fileScan, ExternalTable table, + SelectedPartitions selectedPartitions, Optional snapshot) { + long selectedPartitionRowCount = table.getRowCountForSelectedPartitions(selectedPartitions, snapshot); + if (selectedPartitionRowCount != TableIf.UNKNOWN_ROW_COUNT) { + return buildCatalogRelationStats(fileScan, selectedPartitionRowCount, + selectedPartitions.getPrunableConjuncts(), true); + } + + boolean tableRowCountMatchesSelection = selectedPartitions.totalPartitionNum > 0 + && selectedPartitions.selectedPartitions.size() == selectedPartitions.totalPartitionNum; + return buildCatalogRelationStats(fileScan, table.getRowCount(), + tableRowCountMatchesSelection + ? selectedPartitions.getPrunableConjuncts() : ImmutableSet.of(), + false); + } + /** * compute stats for catalogRelations except OlapScan */ public Statistics computeCatalogRelation(CatalogRelation catalogRelation) { + return buildCatalogRelationStats(catalogRelation, catalogRelation.getTable().getRowCount(), + ImmutableSet.of(), false); + } + + private Statistics buildCatalogRelationStats(CatalogRelation catalogRelation, double scanRowCount, + Set conjunctsAppliedToRowCount, boolean scaleColumnStatsToScanRowCount) { StatisticsBuilder builder = new StatisticsBuilder(); - double tableRowCount = catalogRelation.getTable().getRowCount(); + boolean hasKnownScanRowCount = scanRowCount != TableIf.UNKNOWN_ROW_COUNT; + double estimatedScanRowCount = scanRowCount; // for FeUt, use ColumnStatistic.UNKNOWN if (!FeConstants.enableInternalSchemaDb || ConnectContext.get() == null || ConnectContext.get().getState().isInternal()) { - builder.setRowCount(Math.max(1, tableRowCount)); + builder.setRowCount(hasKnownScanRowCount ? Math.max(1, scanRowCount) : 1); + if (hasKnownScanRowCount) { + builder.setConjunctsAppliedToRowCount(conjunctsAppliedToRowCount); + } for (Slot slot : catalogRelation.getOutput()) { builder.putColumnStatistics(slot, ColumnStatistic.UNKNOWN); } @@ -1274,12 +1393,14 @@ public Statistics computeCatalogRelation(CatalogRelation catalogRelation) { } } Set slotSet = slotSetBuilder.build(); - if (tableRowCount <= 0) { - tableRowCount = 1; + if (hasKnownScanRowCount) { + scanRowCount = Math.max(1, scanRowCount); + } else { + scanRowCount = 1; // try to get row count from col stats for (SlotReference slot : slotSet) { ColumnStatistic cache = getColumnStatsFromTableCache(catalogRelation, slot); - tableRowCount = Math.max(cache.count, tableRowCount); + scanRowCount = Math.max(cache.count, scanRowCount); } } for (SlotReference slot : slotSet) { @@ -1289,11 +1410,26 @@ public Statistics computeCatalogRelation(CatalogRelation catalogRelation) { } else { cache = getColumnStatsFromTableCache(catalogRelation, slot); } - ColumnStatisticBuilder colStatsBuilder = new ColumnStatisticBuilder(cache, tableRowCount); + ColumnStatisticBuilder colStatsBuilder = new ColumnStatisticBuilder(cache, scanRowCount); + if (scaleColumnStatsToScanRowCount && !cache.isUnKnown) { + // External catalogs currently expose table-level column statistics. Keep their ratios when the + // scan cardinality has been narrowed to selected partitions. + double scaledNulls = 0; + if (cache.count > 0 && cache.numNulls >= 0) { + double scale = Math.min(1, Math.max(0, estimatedScanRowCount) / cache.count); + scaledNulls = Math.min(cache.numNulls * scale, scanRowCount); + colStatsBuilder.setNumNulls(scaledNulls); + } + colStatsBuilder.setNdv(Math.min(cache.ndv, scanRowCount - scaledNulls)); + } builder.putColumnStatistics(slot, colStatsBuilder.build()); } checkIfUnknownStatsUsedAsKey(builder); - return builder.setRowCount(tableRowCount).build(); + builder.setRowCount(scanRowCount); + if (hasKnownScanRowCount) { + builder.setConjunctsAppliedToRowCount(conjunctsAppliedToRowCount); + } + return builder.build(); } /** diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/copier/LogicalPlanDeepCopier.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/copier/LogicalPlanDeepCopier.java index abb9e015a278d9..7b3c0e66d9eda2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/copier/LogicalPlanDeepCopier.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/copier/LogicalPlanDeepCopier.java @@ -40,6 +40,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation; import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation; import org.apache.doris.nereids.trees.plans.logical.LogicalExcept; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate; import org.apache.doris.nereids.trees.plans.logical.LogicalHaving; @@ -109,8 +110,17 @@ public Plan visitLogicalCatalogRelation(LogicalCatalogRelation catalogRelation, } LogicalCatalogRelation newRelation = catalogRelation.withRelationId(StatementScopeIdGenerator.newRelationId()); - if (context.shouldInvalidatePartitionPruning() && newRelation instanceof LogicalOlapScan) { - newRelation = ((LogicalOlapScan) newRelation).withPartitionPruned(false); + if (newRelation instanceof LogicalOlapScan) { + if (context.shouldInvalidatePartitionPruning()) { + newRelation = ((LogicalOlapScan) newRelation).withPartitionPruned(false); + } else if (((LogicalOlapScan) catalogRelation).getPartitionPrunablePredicates().isPresent()) { + newRelation = ((LogicalOlapScan) newRelation) + .withReboundPartitionPruningProofFrom((LogicalOlapScan) catalogRelation); + } + } else if (newRelation instanceof LogicalFileScan + && ((LogicalFileScan) catalogRelation).getSelectedPartitions().hasPruningProof()) { + newRelation = ((LogicalFileScan) newRelation) + .withReboundPartitionPruningProofFrom((LogicalFileScan) catalogRelation); } updateReplaceMapWithOutput(catalogRelation, newRelation, context.exprIdReplaceMap); List virtualColumns = catalogRelation.getVirtualColumns().stream() diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PartitionPrunablePredicate.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PartitionPrunablePredicate.java index 54ba19493b86bc..6b241cda0f509b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PartitionPrunablePredicate.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PartitionPrunablePredicate.java @@ -19,11 +19,14 @@ import org.apache.doris.nereids.trees.expressions.Expression; import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.util.ExpressionUtils; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Set; @@ -43,22 +46,22 @@ * *

The predicate lives on the scan itself (see {@code LogicalOlapScan} and * {@code PhysicalOlapScan}) so we no longer need to match it back to its scan - * via a table identifier. Because rewrites between recording and removal may - * rebuild the scan with fresh slot ids, {@link #snapshotPartitionSlots} - * captures the slots that appear in the recorded conjuncts. The post-processor - * maps them onto the actual scan's output slots by column name before - * performing the conjunct removal. + * via a table identifier. {@link #partitionSlots} always belong to the scan + * state carrying this proof. A scan transformation that changes its output + * slots must explicitly call {@link #rebindSlots(Map)} or discard the + * proof, so statistics derivation and physical post-processing never need to + * infer slot lineage on the read path. */ public class PartitionPrunablePredicate { private final Set selectedPartitionIds; - private final List snapshotPartitionSlots; + private final List partitionSlots; private final Set prunableConjuncts; public PartitionPrunablePredicate(Set selectedPartitionIds, - List snapshotPartitionSlots, + List partitionSlots, Set prunableConjuncts) { this.selectedPartitionIds = ImmutableSet.copyOf(selectedPartitionIds); - this.snapshotPartitionSlots = ImmutableList.copyOf(snapshotPartitionSlots); + this.partitionSlots = ImmutableList.copyOf(partitionSlots); this.prunableConjuncts = ImmutableSet.copyOf(prunableConjuncts); } @@ -72,24 +75,51 @@ public boolean equals(Object o) { } PartitionPrunablePredicate that = (PartitionPrunablePredicate) o; return selectedPartitionIds.equals(that.selectedPartitionIds) - && snapshotPartitionSlots.equals(that.snapshotPartitionSlots) + && partitionSlots.equals(that.partitionSlots) && prunableConjuncts.equals(that.prunableConjuncts); } @Override public int hashCode() { - return Objects.hash(selectedPartitionIds, snapshotPartitionSlots, prunableConjuncts); + return Objects.hash(selectedPartitionIds, partitionSlots, prunableConjuncts); } - public Set getSelectedPartitionIds() { - return selectedPartitionIds; + public List getPartitionSlots() { + return partitionSlots; } - public List getSnapshotPartitionSlots() { - return snapshotPartitionSlots; + public boolean covers(List currentSelectedPartitionIds) { + return selectedPartitionIds.containsAll(currentSelectedPartitionIds); } public Set getPrunableConjuncts() { return prunableConjuncts; } + + /** + * Rebind this proof to slots from another scan state of the same table. The caller owns the scan-specific + * lineage rules and must provide a mapping for every recorded partition slot. + */ + public PartitionPrunablePredicate rebindSlots(Map slotMapping) { + Map replacements = new HashMap<>(partitionSlots.size()); + ImmutableList.Builder reboundSlots = + ImmutableList.builderWithExpectedSize(partitionSlots.size()); + for (Slot partitionSlot : partitionSlots) { + Slot reboundSlot = Objects.requireNonNull(slotMapping.get(partitionSlot), + "missing rebound slot for partition slot: " + partitionSlot); + reboundSlots.add(reboundSlot); + if (!partitionSlot.equals(reboundSlot)) { + replacements.put(partitionSlot, reboundSlot); + } + } + if (replacements.isEmpty()) { + return this; + } + ImmutableSet.Builder reboundConjuncts = + ImmutableSet.builderWithExpectedSize(prunableConjuncts.size()); + for (Expression conjunct : prunableConjuncts) { + reboundConjuncts.add(ExpressionUtils.replace(conjunct, replacements)); + } + return new PartitionPrunablePredicate(selectedPartitionIds, reboundSlots.build(), reboundConjuncts.build()); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/OlapScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/OlapScan.java index d5fe7c2341369d..f3b2dccabadbcc 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/OlapScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/OlapScan.java @@ -18,8 +18,10 @@ package org.apache.doris.nereids.trees.plans.algebra; import org.apache.doris.catalog.OlapTable; +import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; import java.util.List; +import java.util.Optional; /** OlapScan */ public interface OlapScan { @@ -32,6 +34,8 @@ public interface OlapScan { List getSelectedTabletIds(); + Optional getPartitionPrunablePredicates(); + /** getScanTabletNum */ default int getScanTabletNum() { List selectedTabletIds = getSelectedTabletIds(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java index 09b5e549b5a473..266aaa99ec50f2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java @@ -31,6 +31,7 @@ import org.apache.doris.nereids.rules.expression.rules.SortedPartitionRanges; import org.apache.doris.nereids.trees.TableSample; import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; import org.apache.doris.nereids.trees.expressions.NamedExpression; import org.apache.doris.nereids.trees.expressions.Slot; import org.apache.doris.nereids.trees.expressions.SlotReference; @@ -40,18 +41,23 @@ import org.apache.doris.nereids.trees.plans.PlanType; import org.apache.doris.nereids.trees.plans.RelationId; import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.nereids.util.ExpressionUtils; import org.apache.doris.nereids.util.Utils; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableList.Builder; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; import java.util.Collection; +import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.Set; /** * Logical file scan for external catalog. @@ -204,7 +210,7 @@ protected boolean hasSameScanState(LogicalCatalogRelation other) { return false; } LogicalFileScan that = (LogicalFileScan) other; - return Objects.equals(selectedPartitions, that.selectedPartitions) + return selectedPartitions.hasSameSelection(that.selectedPartitions) && Objects.equals(tableSample, that.tableSample) && hasSameSnapshot(tableSnapshot, that.tableSnapshot) && hasSameScanParams(scanParams, that.scanParams); @@ -331,6 +337,9 @@ public static class SelectedPartitions { */ public final Optional> sortedPartitionRanges; + private final List partitionSlots; + private final Set prunableConjuncts; + /** * Constructor for SelectedPartitions. */ @@ -353,12 +362,92 @@ public SelectedPartitions(long totalPartitionNum, Map sel public SelectedPartitions(long totalPartitionNum, Map selectedPartitions, boolean isPruned, boolean hasPartitionPredicate, Optional> sortedPartitionRanges) { + this(totalPartitionNum, selectedPartitions, isPruned, hasPartitionPredicate, sortedPartitionRanges, + ImmutableList.of(), ImmutableSet.of()); + } + + private SelectedPartitions(long totalPartitionNum, Map selectedPartitions, + boolean isPruned, boolean hasPartitionPredicate, + Optional> sortedPartitionRanges, + List partitionSlots, Set prunableConjuncts) { this.totalPartitionNum = totalPartitionNum; this.selectedPartitions = ImmutableMap.copyOf(Objects.requireNonNull(selectedPartitions, "selectedPartitions is null")); this.isPruned = isPruned; this.hasPartitionPredicate = hasPartitionPredicate; - this.sortedPartitionRanges = sortedPartitionRanges; + this.sortedPartitionRanges = Objects.requireNonNull(sortedPartitionRanges, + "sortedPartitionRanges is null"); + this.partitionSlots = ImmutableList.copyOf(Objects.requireNonNull(partitionSlots, + "partitionSlots is null")); + this.prunableConjuncts = ImmutableSet.copyOf(Objects.requireNonNull(prunableConjuncts, + "prunableConjuncts is null")); + Preconditions.checkArgument(isPruned || prunableConjuncts.isEmpty(), + "prunable conjuncts require a pruned partition state"); + } + + /** + * Returns a fresh partition-pruning result derived from this selection. The sorted ranges are cleared + * because they describe the pre-pruning partition map, while the predicate proof is frozen with the + * surviving partition set. + */ + public SelectedPartitions withPruneResult(Map selectedPartitions, + boolean hasPartitionPredicate, List partitionSlots, + Set prunableConjuncts) { + Preconditions.checkState(!isPruned, "partition pruning has already been applied"); + Preconditions.checkArgument(this.selectedPartitions.keySet().containsAll(selectedPartitions.keySet()), + "selected partitions must be a subset of the current partition snapshot"); + return new SelectedPartitions(totalPartitionNum, selectedPartitions, true, hasPartitionPredicate, + Optional.empty(), partitionSlots, prunableConjuncts); + } + + public boolean hasPruningProof() { + return !prunableConjuncts.isEmpty(); + } + + /** Rebind the pruning proof to a new output namespace. */ + public SelectedPartitions rebindPruningProof(List output) { + if (!hasPruningProof()) { + return this; + } + Map outputSlotsByName = new HashMap<>(output.size()); + for (Slot slot : output) { + outputSlotsByName.put(slot.getName().toLowerCase(Locale.ROOT), slot); + } + Map replacements = new HashMap<>(partitionSlots.size()); + ImmutableList.Builder reboundSlots = + ImmutableList.builderWithExpectedSize(partitionSlots.size()); + for (Slot partitionSlot : partitionSlots) { + Slot reboundSlot = outputSlotsByName.get(partitionSlot.getName().toLowerCase(Locale.ROOT)); + Preconditions.checkState(reboundSlot != null, + "Can not find output slot for prunable partition slot: %s", partitionSlot.getName()); + reboundSlots.add(reboundSlot); + if (!partitionSlot.equals(reboundSlot)) { + replacements.put(partitionSlot, reboundSlot); + } + } + if (replacements.isEmpty()) { + return this; + } + ImmutableSet.Builder reboundConjuncts = + ImmutableSet.builderWithExpectedSize(prunableConjuncts.size()); + for (Expression conjunct : prunableConjuncts) { + reboundConjuncts.add(ExpressionUtils.replace(conjunct, replacements)); + } + return new SelectedPartitions(totalPartitionNum, selectedPartitions, isPruned, hasPartitionPredicate, + sortedPartitionRanges, reboundSlots.build(), reboundConjuncts.build()); + } + + public Set getPrunableConjuncts() { + return prunableConjuncts; + } + + /** Compare partition-selection state independently of the pruning proof's output-slot namespace. */ + private boolean hasSameSelection(SelectedPartitions other) { + return totalPartitionNum == other.totalPartitionNum + && isPruned == other.isPruned + && hasPartitionPredicate == other.hasPartitionPredicate + && selectedPartitions.keySet().equals(other.selectedPartitions.keySet()) + && sortedPartitionRanges.isPresent() == other.sortedPartitionRanges.isPresent(); } @Override @@ -370,17 +459,15 @@ public boolean equals(Object o) { return false; } SelectedPartitions that = (SelectedPartitions) o; - return isPruned == that.isPruned - && hasPartitionPredicate == that.hasPartitionPredicate - && Objects.equals( - selectedPartitions.keySet(), that.selectedPartitions.keySet()) - && Objects.equals( - sortedPartitionRanges.isPresent(), that.sortedPartitionRanges.isPresent()); + return hasSameSelection(that) + && Objects.equals(partitionSlots, that.partitionSlots) + && Objects.equals(prunableConjuncts, that.prunableConjuncts); } @Override public int hashCode() { - return Objects.hash(selectedPartitions, isPruned, hasPartitionPredicate, sortedPartitionRanges.isPresent()); + return Objects.hash(totalPartitionNum, selectedPartitions.keySet(), isPruned, hasPartitionPredicate, + sortedPartitionRanges.isPresent(), partitionSlots, prunableConjuncts); } } @@ -393,12 +480,20 @@ public LogicalFileScan withOperativeSlots(Collection operativeSlots) { } public LogicalFileScan withCachedOutput(List cachedOutputs) { + SelectedPartitions reboundPartitions = selectedPartitions.rebindPruningProof(cachedOutputs); return AbstractPlan.copyWithSameId(this, () -> new LogicalFileScan(relationId, (ExternalTable) table, qualifier, - selectedPartitions, operativeSlots, virtualColumns, tableSample, tableSnapshot, + reboundPartitions, operativeSlots, virtualColumns, tableSample, tableSnapshot, scanParams, groupExpression, Optional.empty(), tableAlias, Optional.of(cachedOutputs))); } + /** Rebind a copied scan's partition-pruning proof to this scan's output slots. */ + public LogicalFileScan withReboundPartitionPruningProofFrom(LogicalFileScan source) { + Preconditions.checkArgument(getTable().getId() == source.getTable().getId(), + "partition-pruning proof can only be rebound between scans of the same table"); + return withSelectedPartitions(source.selectedPartitions.rebindPruningProof(getOutput())); + } + @Override public List getOperativeSlots() { return operativeSlots; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java index 608a79a75be4d2..9f5c8c403604bf 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java @@ -17,6 +17,8 @@ package org.apache.doris.nereids.trees.plans.logical; +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.SlotRef; import org.apache.doris.analysis.TableScanParams; import org.apache.doris.catalog.Column; import org.apache.doris.catalog.KeysType; @@ -386,6 +388,11 @@ public LogicalOlapScan(RelationId id, Table table, List qualifier, this.partitionPrunablePredicates = partitionPrunablePredicates == null ? Optional.empty() : partitionPrunablePredicates; + Preconditions.checkArgument(partitionPruned || !this.partitionPrunablePredicates.isPresent(), + "partition prunable predicates require a pruned partition state"); + Preconditions.checkArgument(!this.partitionPrunablePredicates.isPresent() + || this.partitionPrunablePredicates.get().covers(this.selectedPartitionIds), + "partition prunable predicates must cover the selected partitions"); this.scanParams = scanParams; } @@ -435,6 +442,7 @@ public boolean hasPartitionPredicate() { return hasPartitionPredicate; } + @Override public Optional getPartitionPrunablePredicates() { return partitionPrunablePredicates; } @@ -458,6 +466,71 @@ public LogicalOlapScan withPartitionPrunablePredicates( partitionPrunablePredicates, scanParams)); } + /** + * Copy a partition-pruning result from another scan of the same table and bind its proof to this scan's + * output slots. This keeps slot lineage repair at the scan-state transition instead of deferring it to + * statistics derivation or physical post-processing. + */ + public LogicalOlapScan withPartitionPruningFrom(LogicalOlapScan source) { + return withSelectedPartitionIds(source.selectedPartitionIds) + .withReboundPartitionPruningProofFrom(source); + } + + /** Rebind only the partition-pruning proof while preserving this scan's own pruning state. */ + public LogicalOlapScan withReboundPartitionPruningProofFrom(LogicalOlapScan source) { + Preconditions.checkArgument(getTable().getId() == source.getTable().getId(), + "partition-pruning proof can only be rebound between scans of the same table"); + return withPartitionPrunablePredicates( + source.rebindPartitionPrunablePredicates(selectedIndexId, getOutput())); + } + + private Optional rebindPartitionPrunablePredicates( + long targetIndexId, List targetOutput) { + if (!partitionPrunablePredicates.isPresent()) { + return Optional.empty(); + } + Map targetSlotsByBaseColumn = directBaseColumnSlots(targetIndexId, targetOutput); + Map slotMapping = new HashMap<>(); + for (Slot partitionSlot : partitionPrunablePredicates.get().getPartitionSlots()) { + Optional baseColumnName = directBaseColumnName(selectedIndexId, partitionSlot); + if (!baseColumnName.isPresent()) { + return Optional.empty(); + } + Slot targetSlot = targetSlotsByBaseColumn.get(baseColumnName.get()); + if (targetSlot == null) { + return Optional.empty(); + } + slotMapping.put(partitionSlot, targetSlot); + } + return Optional.of(partitionPrunablePredicates.get().rebindSlots(slotMapping)); + } + + private Map directBaseColumnSlots(long indexId, List output) { + Map slotsByBaseColumn = new HashMap<>(output.size()); + for (Slot slot : output) { + directBaseColumnName(indexId, slot).ifPresent(name -> slotsByBaseColumn.putIfAbsent(name, slot)); + } + return slotsByBaseColumn; + } + + private Optional directBaseColumnName(long indexId, Slot slot) { + if (!(slot instanceof SlotReference)) { + return Optional.empty(); + } + Optional column = ((SlotReference) slot).getOriginalColumn(); + if (!column.isPresent()) { + return Optional.empty(); + } + if (indexId == getTable().getBaseIndexId()) { + return Optional.of(column.get().getName().toLowerCase(Locale.ROOT)); + } + Expr defineExpr = column.get().getDefineExpr(); + if (!(defineExpr instanceof SlotRef)) { + return Optional.empty(); + } + return Optional.of(((SlotRef) defineExpr).getColumnName().toLowerCase(Locale.ROOT)); + } + @Override public String getFingerprint() { String partitions = ""; @@ -573,6 +646,8 @@ public LogicalOlapScan withSelectedPartitionIds(List selectedPartitionIds) */ public LogicalOlapScan withSelectedPartitionIds(List selectedPartitionIds, boolean hasPartitionPredicate) { + Optional retainedProof = partitionPrunablePredicates + .filter(proof -> proof.covers(selectedPartitionIds)); return AbstractPlan.copyWithSameId(this, () -> new LogicalOlapScan(relationId, (Table) table, qualifier, Optional.empty(), Optional.of(getLogicalProperties()), @@ -581,7 +656,7 @@ public LogicalOlapScan withSelectedPartitionIds(List selectedPartitionIds, hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams)); + retainedProof, scanParams)); } /** @@ -769,7 +844,8 @@ public boolean isPartitionPruned() { } /** - * Return a new scan with the specified partition pruning state. + * Return a new scan with the specified partition pruning state. Resetting the state invalidates the + * predicate proof recorded for the previous pruning result. */ public LogicalOlapScan withPartitionPruned(boolean partitionPruned) { return AbstractPlan.copyWithSameId(this, () -> @@ -780,7 +856,7 @@ public LogicalOlapScan withPartitionPruned(boolean partitionPruned) { hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams)); + partitionPruned ? partitionPrunablePredicates : Optional.empty(), scanParams)); } public List getSelectedTabletIds() { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java index 4630d4236ed40f..d8cbc14b0a54eb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java @@ -324,6 +324,8 @@ public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPa @Override public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPartitionIds, boolean hasPartitionPredicate) { + Optional retainedProof = partitionPrunablePredicates + .filter(proof -> proof.covers(selectedPartitionIds)); return AbstractPlan.copyWithSameId(this, () -> new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, groupExpression, Optional.of(getLogicalProperties()), @@ -332,7 +334,7 @@ public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPa hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, readMode)); + retainedProof, scanParams, readMode)); } @Override @@ -345,7 +347,7 @@ public LogicalOlapTableStreamScan withPartitionPruned(boolean partitionPruned) { hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, readMode)); + partitionPruned ? partitionPrunablePredicates : Optional.empty(), scanParams, readMode)); } /** diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeOlapScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeOlapScan.java index 7b36d1b70c89c6..c811ba74115f06 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeOlapScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalLazyMaterializeOlapScan.java @@ -48,6 +48,7 @@ public PhysicalLazyMaterializeOlapScan(PhysicalOlapScan physicalOlapScan, physicalOlapScan.getSelectedIndexId(), physicalOlapScan.getSelectedTabletIds(), physicalOlapScan.getSelectedPartitionIds(), + physicalOlapScan.hasPartitionPredicate(), physicalOlapScan.getDistributionSpec(), physicalOlapScan.getPreAggStatus(), physicalOlapScan.getBaseOutputs(), @@ -62,7 +63,10 @@ public PhysicalLazyMaterializeOlapScan(PhysicalOlapScan physicalOlapScan, physicalOlapScan.getScoreLimit(), physicalOlapScan.getScoreRangeInfo(), physicalOlapScan.getAnnOrderKeys(), - physicalOlapScan.getAnnLimit() + physicalOlapScan.getAnnLimit(), + physicalOlapScan.getTableAlias(), + physicalOlapScan.getPartitionPrunablePredicates(), + physicalOlapScan.getScanParams() ); this.scan = physicalOlapScan; this.rowId = rowId; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapScan.java index 8fd94f8d47430a..92e570fd5f3a69 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapScan.java @@ -43,6 +43,7 @@ import org.apache.doris.rpc.RpcException; import org.apache.doris.statistics.model.Statistics; +import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -197,6 +198,9 @@ public PhysicalOlapScan(RelationId id, OlapTable olapTable, List qualifi this.partitionPrunablePredicates = partitionPrunablePredicates == null ? Optional.empty() : partitionPrunablePredicates; + Preconditions.checkArgument(!this.partitionPrunablePredicates.isPresent() + || this.partitionPrunablePredicates.get().covers(this.selectedPartitionIds), + "partition prunable predicates must cover the selected partitions"); this.scanParams = scanParams == null ? Optional.empty() : scanParams; } @@ -218,6 +222,7 @@ public boolean hasPartitionPredicate() { return hasPartitionPredicate; } + @Override public Optional getPartitionPrunablePredicates() { return partitionPrunablePredicates; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/model/Statistics.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/model/Statistics.java index 50e1057f29fb60..c4284c43245ab1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/statistics/model/Statistics.java +++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/model/Statistics.java @@ -24,6 +24,8 @@ import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.types.coercion.CharacterType; +import com.google.common.collect.ImmutableSet; + import java.text.DecimalFormat; import java.util.Comparator; import java.util.HashMap; @@ -49,6 +51,9 @@ public class Statistics { private long actualRowCount = -1L; private boolean isFromHbo = false; + // Predicates whose row-count effect is already included by the scan, but whose column-domain effect + // still needs to be applied when the corresponding Filter is estimated. + private final Set conjunctsAppliedToRowCount; public Statistics(Statistics another) { this.rowCount = another.rowCount; @@ -57,15 +62,23 @@ public Statistics(Statistics another) { this.tupleSize = another.tupleSize; this.deltaRowCount = another.getDeltaRowCount(); this.isFromHbo = another.isFromHbo; + this.conjunctsAppliedToRowCount = another.conjunctsAppliedToRowCount; } public Statistics(double rowCount, int widthInJoinCluster, Map expressionToColumnStats, double deltaRowCount, boolean isFromHbo) { + this(rowCount, widthInJoinCluster, expressionToColumnStats, deltaRowCount, isFromHbo, ImmutableSet.of()); + } + + public Statistics(double rowCount, int widthInJoinCluster, + Map expressionToColumnStats, double deltaRowCount, boolean isFromHbo, + Set conjunctsAppliedToRowCount) { this.rowCount = rowCount; this.widthInJoinCluster = widthInJoinCluster; this.expressionToColumnStats = expressionToColumnStats; this.deltaRowCount = deltaRowCount; this.isFromHbo = isFromHbo; + this.conjunctsAppliedToRowCount = ImmutableSet.copyOf(conjunctsAppliedToRowCount); } public Statistics(double rowCount, Map expressionToColumnStats) { @@ -91,10 +104,12 @@ public double getRowCount() { public Statistics withRowCount(double rowCount) { return new Statistics(rowCount, widthInJoinCluster, new HashMap<>(expressionToColumnStats), - 0, isFromHbo); + 0, isFromHbo, conjunctsAppliedToRowCount); } public Statistics withExpressionToColumnStats(Map expressionToColumnStats) { + // The replacement map may use a different expression/ExprId namespace (for example MV statistics + // normalization), so predicate provenance tied to the previous expressions is no longer reusable. return new Statistics(rowCount, widthInJoinCluster, expressionToColumnStats, 0, isFromHbo); } @@ -103,7 +118,7 @@ public Statistics withExpressionToColumnStats(Map e */ public Statistics withRowCountAndEnforceValid(double rowCount) { Statistics statistics = new Statistics(rowCount, widthInJoinCluster, - expressionToColumnStats, 0, isFromHbo); + expressionToColumnStats, 0, isFromHbo, conjunctsAppliedToRowCount); statistics.normalizeColumnStatistics(this.rowCount, false); return statistics; } @@ -162,7 +177,7 @@ public Statistics withSel(double notNullSel, double numNull) { } double newCount = rowCount * notNullSel + numNull; return new Statistics(newCount, widthInJoinCluster, new HashMap<>(expressionToColumnStats), - 0, isFromHbo); + 0, isFromHbo, conjunctsAppliedToRowCount); } public Statistics addColumnStats(Expression expression, ColumnStatistic columnStatistic) { @@ -335,6 +350,10 @@ public boolean isFromHbo() { return this.isFromHbo; } + public Set getConjunctsAppliedToRowCount() { + return conjunctsAppliedToRowCount; + } + public StatisticsBuilder cleanHotValues() { StatisticsBuilder builder = new StatisticsBuilder(this); for (Map.Entry entry : columnStatistics().entrySet()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/model/StatisticsBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/model/StatisticsBuilder.java index 765dad10772a2a..7b618cf0db32b1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/statistics/model/StatisticsBuilder.java +++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/model/StatisticsBuilder.java @@ -20,6 +20,7 @@ import org.apache.doris.nereids.trees.expressions.Expression; import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.Map; import java.util.Set; @@ -32,6 +33,7 @@ public class StatisticsBuilder { private double deltaRowCount = 0.0; private boolean isFromHbo = false; + private final Set conjunctsAppliedToRowCount = new LinkedHashSet<>(); public StatisticsBuilder() { this.expressionToColumnStats = new HashMap<>(); @@ -44,6 +46,7 @@ public StatisticsBuilder(Statistics statistics) { this.expressionToColumnStats = new HashMap<>(); this.expressionToColumnStats.putAll(statistics.columnStatistics()); this.isFromHbo = statistics.isFromHbo(); + this.conjunctsAppliedToRowCount.addAll(statistics.getConjunctsAppliedToRowCount()); } public StatisticsBuilder setRowCount(double rowCount) { @@ -61,6 +64,12 @@ public StatisticsBuilder setDeltaRowCount(double deltaRowCount) { return this; } + public StatisticsBuilder setConjunctsAppliedToRowCount(Set conjunctsAppliedToRowCount) { + this.conjunctsAppliedToRowCount.clear(); + this.conjunctsAppliedToRowCount.addAll(conjunctsAppliedToRowCount); + return this; + } + public StatisticsBuilder putColumnStatistics( Map expressionToColumnStats) { this.expressionToColumnStats.putAll(expressionToColumnStats); @@ -77,6 +86,7 @@ public Set> getExpressionColumnStatsEntri } public Statistics build() { - return new Statistics(rowCount, widthInJoinCluster, expressionToColumnStats, deltaRowCount, isFromHbo); + return new Statistics(rowCount, widthInJoinCluster, expressionToColumnStats, deltaRowCount, isFromHbo, + conjunctsAppliedToRowCount); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java index 2671b51b3796ff..c5294ac2aeacca 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java @@ -498,6 +498,25 @@ public void testGetPartitionRowCount() { olapTable.getRowCountForPartitionIndex(11, 10, true); } + @Test + public void testGetSelectedPartitionRowCountEstimatesUnknownPartitionsFromRemainingRows() { + OlapTable olapTable = new OlapTable(); + MaterializedIndex knownIndex = new MaterializedIndex(10, MaterializedIndex.IndexState.NORMAL); + knownIndex.setRowCountReported(true); + knownIndex.setRowCount(30); + MaterializedIndex selectedUnknownIndex = new MaterializedIndex(10, MaterializedIndex.IndexState.NORMAL); + selectedUnknownIndex.setRowCountReported(false); + MaterializedIndex otherUnknownIndex = new MaterializedIndex(10, MaterializedIndex.IndexState.NORMAL); + otherUnknownIndex.setRowCountReported(false); + olapTable.addPartition(new Partition(1, "p1", knownIndex, null)); + olapTable.addPartition(new Partition(2, "p2", selectedUnknownIndex, null)); + olapTable.addPartition(new Partition(3, "p3", otherUnknownIndex, null)); + + double rowCount = olapTable.getRowCountForSelectedPartitions(Arrays.asList(1L, 2L), 10, 100); + + Assertions.assertEquals(65, rowCount, 0.001); + } + @Test public void testGetSchemaAllIndexes() { OlapTable table = Mockito.spy(new OlapTable()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableRowCountTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableRowCountTest.java index c17ecbeababe8d..e5f7ecdbb2cc50 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableRowCountTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableRowCountTest.java @@ -18,6 +18,7 @@ package org.apache.doris.datasource.plugin; import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.PartitionItem; import org.apache.doris.catalog.PrimitiveType; import org.apache.doris.catalog.TableIf; import org.apache.doris.connector.spi.Connector; @@ -26,13 +27,21 @@ import org.apache.doris.connector.spi.ConnectorStatementScope; import org.apache.doris.connector.spi.ConnectorTableStatistics; import org.apache.doris.connector.spi.handle.ConnectorTableHandle; +import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; import org.apache.doris.datasource.ExternalDatabase; import org.apache.doris.datasource.SchemaCacheValue; import org.apache.doris.datasource.SessionContext; +import org.apache.doris.datasource.mvcc.PluginDrivenMvccSnapshot; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; +import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.GlobalVariable; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; import org.mockito.Mockito; import java.util.Arrays; @@ -199,6 +208,122 @@ public void layer3ZeroQuotientYieldsUnknownNotEmptyTable() { }); } + @Test + public void selectedPartitionFileListEstimateUsesOnlySelectedPartitionNames() { + withFileListGate(true, () -> { + ConnectorMetadata metadata = Mockito.mock(ConnectorMetadata.class); + ConnectorSession session = Mockito.mock(ConnectorSession.class); + Mockito.when(session.getStatementScope()).thenReturn(ConnectorStatementScope.NONE); + ConnectorTableHandle handle = Mockito.mock(ConnectorTableHandle.class); + Mockito.when(metadata.getTableHandle(session, "REMOTE_DB", "REMOTE_TBL")) + .thenReturn(Optional.of(handle)); + Mockito.when(metadata.estimateDataSizeByListingFiles( + session, handle, Collections.singletonList("p1"))).thenReturn(400L); + TestablePluginCatalog catalog = new TestablePluginCatalog(metadata, session); + + @SuppressWarnings("unchecked") + ExternalDatabase db = Mockito.mock(ExternalDatabase.class); + Mockito.when(db.getRemoteName()).thenReturn("REMOTE_DB"); + List schema = Arrays.asList(intCol("v"), bigintCol("dt")); + PluginDrivenSchemaCacheValue cacheValue = new PluginDrivenSchemaCacheValue( + schema, Collections.singletonList(schema.get(1)), Collections.singletonList("dt")); + PluginDrivenExternalTable table = new PluginDrivenExternalTable( + 1L, "tbl", "REMOTE_TBL", catalog, db) { + @Override + protected synchronized void makeSureInitialized() { + } + + @Override + public Optional getSchemaCacheValue() { + return Optional.of(cacheValue); + } + }; + + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions selectedPartitions = new SelectedPartitions( + 2, ImmutableMap.of("p1", p1, "p2", p2), false) + .withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(), Collections.emptySet()); + + Assertions.assertEquals(100L, + table.getRowCountForSelectedPartitions(selectedPartitions, Optional.empty())); + Mockito.verify(metadata, Mockito.never()).estimateDataSizeByListingFiles(session, handle); + }); + } + + @Test + public void selectedPartitionEstimateSkipsRemoteListingWhenInternalTablesNeedPlanLocks() { + withFileListGate(true, () -> { + PluginDrivenExternalTable table = tableForFileList( + 400L, Arrays.asList(intCol("v"), bigintCol("dt")), Collections.singletonList(1)); + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions selectedPartitions = new SelectedPartitions( + 2, ImmutableMap.of("p1", p1, "p2", p2), false) + .withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(), Collections.emptySet()); + ConnectContext context = Mockito.mock(ConnectContext.class); + StatementContext statementContext = Mockito.mock(StatementContext.class); + Mockito.when(context.getStatementContext()).thenReturn(statementContext); + Mockito.when(statementContext.hasAnyPlanReadLockTable()).thenReturn(true); + + try (MockedStatic mockedContext = Mockito.mockStatic(ConnectContext.class)) { + mockedContext.when(ConnectContext::get).thenReturn(context); + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, + table.getRowCountForSelectedPartitions(selectedPartitions, Optional.empty())); + } + }); + } + + @Test + public void selectedPartitionEstimateUsesThePinnedSnapshotHandle() { + withFileListGate(true, () -> { + ConnectorMetadata metadata = Mockito.mock(ConnectorMetadata.class); + ConnectorSession session = Mockito.mock(ConnectorSession.class); + Mockito.when(session.getStatementScope()).thenReturn(ConnectorStatementScope.NONE); + ConnectorTableHandle handle = Mockito.mock(ConnectorTableHandle.class); + ConnectorTableHandle pinnedHandle = Mockito.mock(ConnectorTableHandle.class); + ConnectorMvccSnapshot connectorSnapshot = ConnectorMvccSnapshot.builder().snapshotId(7).build(); + PluginDrivenMvccSnapshot snapshot = new PluginDrivenMvccSnapshot( + connectorSnapshot, Collections.emptyMap(), Collections.emptyMap()); + Mockito.when(metadata.getTableHandle(session, "REMOTE_DB", "REMOTE_TBL")) + .thenReturn(Optional.of(handle)); + Mockito.when(metadata.applySnapshot(session, handle, connectorSnapshot)).thenReturn(pinnedHandle); + Mockito.when(metadata.estimateDataSizeByListingFiles( + session, pinnedHandle, Collections.singletonList("p1"))).thenReturn(400L); + TestablePluginCatalog catalog = new TestablePluginCatalog(metadata, session); + + @SuppressWarnings("unchecked") + ExternalDatabase db = Mockito.mock(ExternalDatabase.class); + Mockito.when(db.getRemoteName()).thenReturn("REMOTE_DB"); + List schema = Arrays.asList(intCol("v"), bigintCol("dt")); + PluginDrivenSchemaCacheValue cacheValue = new PluginDrivenSchemaCacheValue( + schema, Collections.singletonList(schema.get(1)), Collections.singletonList("dt")); + PluginDrivenExternalTable table = new PluginDrivenExternalTable( + 1L, "tbl", "REMOTE_TBL", catalog, db) { + @Override + protected synchronized void makeSureInitialized() { + } + + @Override + public Optional getSchemaCacheValue() { + return Optional.of(cacheValue); + } + }; + + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions selectedPartitions = new SelectedPartitions( + 2, ImmutableMap.of("p1", p1, "p2", p2), false) + .withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(), Collections.emptySet()); + + Assertions.assertEquals(100L, + table.getRowCountForSelectedPartitions(selectedPartitions, Optional.of(snapshot))); + }); + } + private static void withFileListGate(boolean enabled, Runnable body) { boolean previous = GlobalVariable.enable_get_row_count_from_file_list; GlobalVariable.enable_get_row_count_from_file_list = enabled; diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PartitionPrunerTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PartitionPrunerTest.java index f26c4701ce711b..30de9e06789eff 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PartitionPrunerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PartitionPrunerTest.java @@ -327,7 +327,22 @@ public void testComplexNestedPredicate() } @Test - public void testPruneWithResultIgnoresNonPruningPartitionPredicate() throws AnalysisException { + public void testPruneWithResultIgnoresNonPartitionPredicate() throws AnalysisException { + Map idToPartitions = ImmutableMap.of( + "p1", createListPartitionItem("1"), + "p2", createListPartitionItem("2")); + + PartitionPruneResult result = PartitionPruner.pruneWithResult( + ImmutableList.of(slotA), new GreaterThan(slotB, Literal.of(0)), idToPartitions, cascadesContext, + PartitionTableType.OLAP, Optional.empty()); + + Assertions.assertEquals(2, result.partitions.size()); + Assertions.assertFalse(result.hasPartitionPredicate); + Assertions.assertFalse(result.prunedPartitionPredicate.isPresent()); + } + + @Test + public void testPruneWithResultKeepsProvableNonFilteringPartitionPredicate() throws AnalysisException { Map idToPartitions = ImmutableMap.of( "p1", createListPartitionItem("1"), "p2", createListPartitionItem("2")); @@ -338,6 +353,7 @@ public void testPruneWithResultIgnoresNonPruningPartitionPredicate() throws Anal Assertions.assertEquals(2, result.partitions.size()); Assertions.assertFalse(result.hasPartitionPredicate); + Assertions.assertTrue(result.prunedPartitionPredicate.isPresent()); } @Test diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PullUpJoinFromUnionAllTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PullUpJoinFromUnionAllTest.java index bf224a763409a0..90dd7bebeaa2f3 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PullUpJoinFromUnionAllTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PullUpJoinFromUnionAllTest.java @@ -24,6 +24,7 @@ import org.apache.doris.catalog.OdbcTable; import org.apache.doris.catalog.OlapTable; import org.apache.doris.catalog.Partition; +import org.apache.doris.catalog.PartitionItem; import org.apache.doris.catalog.Type; import org.apache.doris.catalog.stream.OlapTableStreamWrapper; import org.apache.doris.catalog.stream.StreamReadMode; @@ -35,11 +36,13 @@ import org.apache.doris.nereids.trees.expressions.NamedExpression; import org.apache.doris.nereids.trees.expressions.Slot; import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; import org.apache.doris.nereids.trees.plans.JoinType; import org.apache.doris.nereids.trees.plans.Plan; import org.apache.doris.nereids.trees.plans.RelationId; import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier; import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; import org.apache.doris.nereids.trees.plans.logical.LogicalJoin; import org.apache.doris.nereids.trees.plans.logical.LogicalOdbcScan; @@ -152,6 +155,42 @@ void comparatorRejectsDifferentSelectedPartitions() { Assertions.assertFalse(comparator.isLogicalEqual(firstPartitionScan, secondPartitionScan)); } + @Test + void comparatorIgnoresPruningProofExprIdsButKeepsPartitionScope() { + ExternalTable table = Mockito.mock(ExternalTable.class); + Mockito.when(table.getId()).thenReturn(48L); + Mockito.when(table.getName()).thenReturn("common_partitioned_file"); + Mockito.when(table.getDatabase()).thenReturn(null); + Mockito.when(table.getBaseSchema()).thenReturn(ImmutableList.of(new Column("dt", Type.INT, true))); + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions initial = new SelectedPartitions(2, ImmutableMap.of("p1", p1, "p2", p2), false); + + LogicalFileScan left = newFileScan(table, Optional.empty()); + Slot leftSlot = left.getOutput().get(0); + EqualTo leftPredicate = new EqualTo(leftSlot, new IntegerLiteral(1)); + left = left.withSelectedPartitions(initial.withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(leftSlot), ImmutableSet.of(leftPredicate))); + + LogicalFileScan right = newFileScan(table, Optional.empty()); + Slot rightSlot = right.getOutput().get(0); + EqualTo rightPredicate = new EqualTo(rightSlot, new IntegerLiteral(1)); + right = right.withSelectedPartitions(initial.withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(rightSlot), ImmutableSet.of(rightPredicate))); + + PullUpJoinFromUnionAll.LogicalPlanComparator comparator = + new PullUpJoinFromUnionAll().new LogicalPlanComparator(); + Assertions.assertNotEquals(left.getSelectedPartitions(), right.getSelectedPartitions()); + Assertions.assertTrue(comparator.isLogicalEqual( + new LogicalFilter<>(ImmutableSet.of(leftPredicate), left), + new LogicalFilter<>(ImmutableSet.of(rightPredicate), right))); + + LogicalFileScan differentPartition = left.withSelectedPartitions(initial.withPruneResult( + ImmutableMap.of("p2", p2), true, ImmutableList.of(leftSlot), + ImmutableSet.of(new EqualTo(leftSlot, new IntegerLiteral(2))))); + Assertions.assertFalse(comparator.isLogicalEqual(left, differentPartition)); + } + @Test void comparatorRejectsDifferentFileScanParams() { // Post-migration, Hudi (and every external file table) binds to LogicalFileScan, and the Hudi diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/StatsDeriveTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/StatsDeriveTest.java new file mode 100644 index 00000000000000..dae35b175f31a6 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/StatsDeriveTest.java @@ -0,0 +1,70 @@ +// 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.nereids.rules.rewrite; + +import org.apache.doris.catalog.PartitionItem; +import org.apache.doris.common.FeConstants; +import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.plans.RelationId; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; +import org.apache.doris.nereids.types.IntegerType; +import org.apache.doris.statistics.model.Statistics; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.Optional; + +public class StatsDeriveTest { + + @Test + public void logicalFileScanUsesSelectedPartitionRowCount() { + boolean previous = FeConstants.enableInternalSchemaDb; + try { + FeConstants.enableInternalSchemaDb = false; + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions selectedPartitions = new SelectedPartitions( + 2, ImmutableMap.of("p1", p1, "p2", p2), false) + .withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(), Collections.emptySet()); + ExternalTable table = Mockito.mock(ExternalTable.class); + Mockito.when(table.initSelectedPartitions(Mockito.any())).thenReturn(selectedPartitions); + Mockito.when(table.getRowCountForSelectedPartitions( + Mockito.eq(selectedPartitions), Mockito.any())).thenReturn(7L); + SlotReference output = new SlotReference("v", IntegerType.INSTANCE); + LogicalFileScan scan = new LogicalFileScan(new RelationId(1), table, + ImmutableList.of("db"), ImmutableList.of(), Optional.empty(), Optional.empty(), + Optional.empty(), Optional.of(ImmutableList.of(output))); + + Statistics statistics = scan.accept(new StatsDerive(false), new StatsDerive.DeriveContext()); + + Assertions.assertEquals(7, statistics.getRowCount(), 0.001); + Assertions.assertSame(statistics, scan.getStats()); + } finally { + FeConstants.enableInternalSchemaDb = previous; + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/FilterEstimationTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/FilterEstimationTest.java index 115f6786a28e54..77e573edfb4231 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/FilterEstimationTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/FilterEstimationTest.java @@ -35,6 +35,7 @@ import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.trees.expressions.functions.scalar.Left; import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral; +import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral; import org.apache.doris.nereids.trees.expressions.literal.DateLiteral; import org.apache.doris.nereids.trees.expressions.literal.DateTimeLiteral; import org.apache.doris.nereids.trees.expressions.literal.DoubleLiteral; @@ -61,6 +62,17 @@ import java.util.Map; class FilterEstimationTest { + @Test + public void testBooleanLiteral() { + Statistics statistics = new StatisticsBuilder().setRowCount(100).build(); + FilterEstimation estimation = new FilterEstimation(); + + Assertions.assertEquals(100, + estimation.estimate(BooleanLiteral.TRUE, statistics).getRowCount(), 0.001); + Assertions.assertEquals(0, + estimation.estimate(BooleanLiteral.FALSE, statistics).getRowCount(), 0.001); + } + // a > 500 or b < 100 // b isNaN @Test diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java index 078368cb71a06e..6ce3bf81046372 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/stats/StatsCalculatorTest.java @@ -22,8 +22,12 @@ import org.apache.doris.catalog.Env; import org.apache.doris.catalog.OlapTable; import org.apache.doris.catalog.Partition; +import org.apache.doris.catalog.PartitionItem; import org.apache.doris.catalog.PrimitiveType; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.FeConstants; import org.apache.doris.common.Pair; +import org.apache.doris.datasource.ExternalTable; import org.apache.doris.nereids.CascadesContext; import org.apache.doris.nereids.memo.Group; import org.apache.doris.nereids.memo.GroupExpression; @@ -33,6 +37,11 @@ import org.apache.doris.nereids.trees.expressions.EqualTo; import org.apache.doris.nereids.trees.expressions.ExprId; import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Not; import org.apache.doris.nereids.trees.expressions.Or; import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; @@ -40,11 +49,14 @@ import org.apache.doris.nereids.trees.plans.GroupPlan; import org.apache.doris.nereids.trees.plans.JoinType; import org.apache.doris.nereids.trees.plans.LimitPhase; +import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; import org.apache.doris.nereids.trees.plans.Plan; import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier; import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate; import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation; import org.apache.doris.nereids.trees.plans.logical.LogicalExcept; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; import org.apache.doris.nereids.trees.plans.logical.LogicalJoin; import org.apache.doris.nereids.trees.plans.logical.LogicalLimit; @@ -62,6 +74,7 @@ import org.apache.doris.statistics.model.ColumnStatistic; import org.apache.doris.statistics.model.ColumnStatisticBuilder; import org.apache.doris.statistics.model.Statistics; +import org.apache.doris.statistics.model.StatisticsBuilder; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -69,6 +82,8 @@ import com.google.common.collect.Lists; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.mockito.MockedStatic; import org.mockito.Mockito; @@ -93,6 +108,15 @@ private Group newFakeGroup() { return group; } + private SelectedPartitions selectedPartitions(SlotReference partitionSlot, Expression partitionPredicate) { + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions initial = new SelectedPartitions( + 2, ImmutableMap.of("p1", p1, "p2", p2), false); + return initial.withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(partitionSlot), ImmutableSet.of(partitionPredicate)); + } + @Test public void testFilter() { List qualifier = Lists.newArrayList(); @@ -141,6 +165,366 @@ public void testFilter() { ownerGroupOr.getStatistics().getRowCount(), 0.1); } + @Test + public void testFilterSkipsConjunctAlreadyAppliedToScanRowCount() { + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + SlotReference valueSlot = new SlotReference("v", IntegerType.INSTANCE); + EqualTo partitionPredicate = new EqualTo(partitionSlot, new IntegerLiteral(1)); + EqualTo valuePredicate = new EqualTo(valueSlot, new IntegerLiteral(2)); + Statistics input = new StatisticsBuilder() + .setRowCount(1000) + .putColumnStatistics(partitionSlot, new ColumnStatisticBuilder(1000).setNdv(100).build()) + .putColumnStatistics(valueSlot, new ColumnStatisticBuilder(1000).setNdv(10).build()) + .setConjunctsAppliedToRowCount(ImmutableSet.of(partitionPredicate)) + .build(); + StatsCalculator calculator = new StatsCalculator((CascadesContext) null); + + LogicalFilter filter = new LogicalFilter<>( + ImmutableSet.of(partitionPredicate, valuePredicate), scan1); + Statistics filtered = calculator.computeFilter(filter, input); + Assertions.assertEquals(100, filtered.getRowCount(), 0.001); + Assertions.assertEquals(1, filtered.findColumnStatistics(partitionSlot).ndv, 0.001); + Assertions.assertEquals(1, filtered.findColumnStatistics(valueSlot).ndv, 0.001); + } + + @Test + public void testAppliedConjunctsDoNotCapUnrelatedPartitionColumnNdv() { + SlotReference firstPartitionSlot = new SlotReference("p1", IntegerType.INSTANCE); + SlotReference secondPartitionSlot = new SlotReference("p2", IntegerType.INSTANCE); + EqualTo firstPredicate = new EqualTo(firstPartitionSlot, new IntegerLiteral(1)); + GreaterThanEqual secondPredicate = new GreaterThanEqual(secondPartitionSlot, new IntegerLiteral(1)); + ColumnStatistic partitionColumnStats = new ColumnStatisticBuilder(1000) + .setNdv(100) + .setMinValue(1) + .setMinExpr(new IntLiteral(1)) + .setMaxValue(100) + .setMaxExpr(new IntLiteral(100)) + .build(); + Statistics input = new StatisticsBuilder() + .setRowCount(1000) + .putColumnStatistics(firstPartitionSlot, partitionColumnStats) + .putColumnStatistics(secondPartitionSlot, partitionColumnStats) + .setConjunctsAppliedToRowCount(ImmutableSet.of(firstPredicate, secondPredicate)) + .build(); + LogicalFilter filter = new LogicalFilter<>( + ImmutableSet.of(firstPredicate, secondPredicate), scan1); + + Statistics filtered = new StatsCalculator((CascadesContext) null).computeFilter(filter, input); + + Assertions.assertEquals(1000, filtered.getRowCount(), 0.001); + Assertions.assertEquals(1, filtered.findColumnStatistics(firstPartitionSlot).ndv, 0.001); + Assertions.assertEquals(100, filtered.findColumnStatistics(secondPartitionSlot).ndv, 0.001); + } + + @Test + public void testAppliedBoundsOnSameSlotShareOneEstimationBasis() { + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + GreaterThanEqual lowerBound = new GreaterThanEqual(partitionSlot, new IntegerLiteral(4501)); + LessThanEqual upperBound = new LessThanEqual(partitionSlot, new IntegerLiteral(4600)); + ColumnStatistic tableColumnStats = new ColumnStatisticBuilder(100_000) + .setNdv(10_000) + .setMinValue(1) + .setMinExpr(new IntLiteral(1)) + .setMaxValue(10_000) + .setMaxExpr(new IntLiteral(10_000)) + .build(); + ColumnStatistic selectedColumnStats = new ColumnStatisticBuilder(tableColumnStats, 1000) + .setNdv(1000) + .build(); + Statistics input = new StatisticsBuilder() + .setRowCount(1000) + .putColumnStatistics(partitionSlot, selectedColumnStats) + .setConjunctsAppliedToRowCount(ImmutableSet.of(lowerBound, upperBound)) + .build(); + LogicalFilter filter = new LogicalFilter<>( + ImmutableSet.of(lowerBound, upperBound), scan1); + + Statistics filtered = new StatsCalculator((CascadesContext) null).computeFilter(filter, input); + + Assertions.assertEquals(1000, filtered.getRowCount(), 0.001); + double ndv = filtered.findColumnStatistics(partitionSlot).ndv; + Assertions.assertTrue(ndv > 90 && ndv <= 100); + Assertions.assertEquals(4501, filtered.findColumnStatistics(partitionSlot).minValue, 0.001); + Assertions.assertEquals(4600, filtered.findColumnStatistics(partitionSlot).maxValue, 0.001); + } + + @Test + public void testAppliedPredicateKeepsExistingPartitionBoundsForRemainingPredicate() { + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + SlotReference valueSlot = new SlotReference("v", IntegerType.INSTANCE); + GreaterThanEqual appliedPredicate = new GreaterThanEqual(partitionSlot, new IntegerLiteral(1)); + LessThan remainingPredicate = new LessThan(partitionSlot, valueSlot); + ColumnStatistic tablePartitionStats = new ColumnStatisticBuilder(100_000) + .setNdv(10_000) + .setMinValue(1) + .setMinExpr(new IntLiteral(1)) + .setMaxValue(10_000) + .setMaxExpr(new IntLiteral(10_000)) + .build(); + ColumnStatistic selectedPartitionStats = new ColumnStatisticBuilder(tablePartitionStats, 1000) + .setNdv(1000) + .setMinValue(1) + .setMinExpr(new IntLiteral(1)) + .setMaxValue(2) + .setMaxExpr(new IntLiteral(2)) + .build(); + ColumnStatistic valueStats = new ColumnStatisticBuilder(1000) + .setNdv(100) + .setMinValue(100) + .setMinExpr(new IntLiteral(100)) + .setMaxValue(200) + .setMaxExpr(new IntLiteral(200)) + .build(); + Statistics input = new StatisticsBuilder() + .setRowCount(1000) + .putColumnStatistics(partitionSlot, selectedPartitionStats) + .putColumnStatistics(valueSlot, valueStats) + .setConjunctsAppliedToRowCount(ImmutableSet.of(appliedPredicate)) + .build(); + LogicalFilter filter = new LogicalFilter<>( + ImmutableSet.of(appliedPredicate, remainingPredicate), scan1); + + Statistics filtered = new StatsCalculator((CascadesContext) null).computeFilter(filter, input); + + Assertions.assertEquals(1000, filtered.getRowCount(), 0.001); + Assertions.assertEquals(1, filtered.findColumnStatistics(partitionSlot).minValue, 0.001); + Assertions.assertEquals(2, filtered.findColumnStatistics(partitionSlot).maxValue, 0.001); + } + + @Test + public void testRestoredAppliedConjunctStatsKeepNullCountNonNegative() { + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + GreaterThanEqual appliedPredicate = new GreaterThanEqual(partitionSlot, new IntegerLiteral(9901)); + Not remainingPredicate = new Not(new EqualTo(partitionSlot, new IntegerLiteral(9999))); + ColumnStatistic tableColumnStats = new ColumnStatisticBuilder(100_000) + .setNdv(10_000) + .setNumNulls(0) + .setMinValue(1) + .setMinExpr(new IntLiteral(1)) + .setMaxValue(10_000) + .setMaxExpr(new IntLiteral(10_000)) + .build(); + ColumnStatistic selectedColumnStats = new ColumnStatisticBuilder(tableColumnStats, 1000) + .setNdv(1000) + .setMinValue(9901) + .setMinExpr(new IntLiteral(9901)) + .setMaxValue(10_000) + .setMaxExpr(new IntLiteral(10_000)) + .build(); + Statistics input = new StatisticsBuilder() + .setRowCount(1000) + .putColumnStatistics(partitionSlot, selectedColumnStats) + .setConjunctsAppliedToRowCount(ImmutableSet.of(appliedPredicate)) + .build(); + LogicalFilter filter = new LogicalFilter<>( + ImmutableSet.of(appliedPredicate, remainingPredicate), scan1); + + Statistics filtered = new StatsCalculator((CascadesContext) null).computeFilter(filter, input); + + Assertions.assertTrue(filtered.getRowCount() <= 1000); + Assertions.assertEquals(0, filtered.findColumnStatistics(partitionSlot).numNulls, 0.001); + } + + @Test + public void testAppliedMultiColumnConjunctDoesNotCapColumnNdv() { + SlotReference firstPartitionSlot = new SlotReference("p1", IntegerType.INSTANCE); + SlotReference secondPartitionSlot = new SlotReference("p2", IntegerType.INSTANCE); + EqualTo partitionPredicate = new EqualTo(firstPartitionSlot, secondPartitionSlot); + ColumnStatistic partitionColumnStats = new ColumnStatisticBuilder(1000) + .setNdv(100) + .setMinValue(1) + .setMinExpr(new IntLiteral(1)) + .setMaxValue(100) + .setMaxExpr(new IntLiteral(100)) + .build(); + Statistics input = new StatisticsBuilder() + .setRowCount(1000) + .putColumnStatistics(firstPartitionSlot, partitionColumnStats) + .putColumnStatistics(secondPartitionSlot, partitionColumnStats) + .setConjunctsAppliedToRowCount(ImmutableSet.of(partitionPredicate)) + .build(); + LogicalFilter filter = new LogicalFilter<>( + ImmutableSet.of(partitionPredicate), scan1); + + Statistics filtered = new StatsCalculator((CascadesContext) null).computeFilter(filter, input); + + Assertions.assertEquals(1000, filtered.getRowCount(), 0.001); + Assertions.assertEquals(100, filtered.findColumnStatistics(firstPartitionSlot).ndv, 0.001); + Assertions.assertEquals(100, filtered.findColumnStatistics(secondPartitionSlot).ndv, 0.001); + } + + @Test + public void testAppliedIsNullConjunctRestoresSelectedRowNullCount() { + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + IsNull partitionPredicate = new IsNull(partitionSlot); + Statistics input = new StatisticsBuilder() + .setRowCount(1000) + .putColumnStatistics(partitionSlot, new ColumnStatisticBuilder(1000) + .setNdv(100) + .setNumNulls(100) + .build()) + .setConjunctsAppliedToRowCount(ImmutableSet.of(partitionPredicate)) + .build(); + LogicalFilter filter = new LogicalFilter<>( + ImmutableSet.of(partitionPredicate), scan1); + + Statistics filtered = new StatsCalculator((CascadesContext) null).computeFilter(filter, input); + + Assertions.assertEquals(1000, filtered.getRowCount(), 0.001); + Assertions.assertEquals(0, filtered.findColumnStatistics(partitionSlot).ndv, 0.001); + Assertions.assertEquals(1000, filtered.findColumnStatistics(partitionSlot).numNulls, 0.001); + } + + @ParameterizedTest(name = "selected row count {0} produces scan row count {1}") + @CsvSource({"7, 7", "0, 1"}) + public void testFileScanUsesKnownSelectedPartitionRowCount(long selectedRowCount, double expectedRowCount) { + boolean previous = FeConstants.enableInternalSchemaDb; + try { + FeConstants.enableInternalSchemaDb = false; + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + EqualTo partitionPredicate = new EqualTo(partitionSlot, new IntegerLiteral(1)); + SelectedPartitions selectedPartitions = selectedPartitions(partitionSlot, partitionPredicate); + ExternalTable table = Mockito.mock(ExternalTable.class); + LogicalFileScan scan = Mockito.mock(LogicalFileScan.class); + Mockito.when(scan.getTable()).thenReturn(table); + Mockito.when(scan.getSelectedPartitions()).thenReturn(selectedPartitions); + Mockito.when(scan.getTableSnapshot()).thenReturn(Optional.empty()); + Mockito.when(scan.getScanParams()).thenReturn(Optional.empty()); + Mockito.when(scan.getOutput()).thenReturn(ImmutableList.of(partitionSlot)); + Mockito.when(table.getRowCountForSelectedPartitions( + Mockito.eq(selectedPartitions), Mockito.any())).thenReturn(selectedRowCount); + + Statistics statistics = new StatsCalculator((CascadesContext) null).computeFileScan(scan); + + Assertions.assertEquals(expectedRowCount, statistics.getRowCount(), 0.001); + Assertions.assertEquals(ImmutableSet.of(partitionPredicate), + statistics.getConjunctsAppliedToRowCount()); + Mockito.verify(table, Mockito.never()).getRowCount(); + } finally { + FeConstants.enableInternalSchemaDb = previous; + } + } + + @Test + public void testFileScanKeepsPartitionPredicateWhenSelectedRowCountIsUnknown() { + boolean previous = FeConstants.enableInternalSchemaDb; + try { + FeConstants.enableInternalSchemaDb = false; + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + EqualTo partitionPredicate = new EqualTo(partitionSlot, new IntegerLiteral(1)); + SelectedPartitions selectedPartitions = selectedPartitions(partitionSlot, partitionPredicate); + ExternalTable table = Mockito.mock(ExternalTable.class); + LogicalFileScan scan = Mockito.mock(LogicalFileScan.class); + Mockito.when(scan.getTable()).thenReturn(table); + Mockito.when(scan.getSelectedPartitions()).thenReturn(selectedPartitions); + Mockito.when(scan.getTableSnapshot()).thenReturn(Optional.empty()); + Mockito.when(scan.getScanParams()).thenReturn(Optional.empty()); + Mockito.when(scan.getOutput()).thenReturn(ImmutableList.of(partitionSlot)); + Mockito.when(table.getRowCountForSelectedPartitions( + Mockito.eq(selectedPartitions), Mockito.any())).thenReturn(TableIf.UNKNOWN_ROW_COUNT); + Mockito.when(table.getRowCount()).thenReturn(1000L); + + StatsCalculator calculator = new StatsCalculator((CascadesContext) null); + Statistics scanStatistics = calculator.computeFileScan(scan); + + Assertions.assertEquals(1000, scanStatistics.getRowCount(), 0.001); + Assertions.assertTrue(scanStatistics.getConjunctsAppliedToRowCount().isEmpty()); + + Statistics input = new StatisticsBuilder(scanStatistics) + .putColumnStatistics(partitionSlot, + new ColumnStatisticBuilder(1000).setNdv(100).build()) + .build(); + LogicalFilter filter = + new LogicalFilter<>(ImmutableSet.of(partitionPredicate), scan1); + Assertions.assertEquals(10, calculator.computeFilter(filter, input).getRowCount(), 0.001); + } finally { + FeConstants.enableInternalSchemaDb = previous; + } + } + + @Test + public void testFileScanDoesNotApplyPartitionProofForDeferredEmptyPartitionUniverse() { + boolean previous = FeConstants.enableInternalSchemaDb; + try { + FeConstants.enableInternalSchemaDb = false; + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + EqualTo partitionPredicate = new EqualTo(partitionSlot, new IntegerLiteral(1)); + SelectedPartitions selectedPartitions = new SelectedPartitions(0, ImmutableMap.of(), false) + .withPruneResult(ImmutableMap.of(), false, + ImmutableList.of(partitionSlot), ImmutableSet.of(partitionPredicate)); + ExternalTable table = Mockito.mock(ExternalTable.class); + LogicalFileScan scan = Mockito.mock(LogicalFileScan.class); + Mockito.when(scan.getTable()).thenReturn(table); + Mockito.when(scan.getSelectedPartitions()).thenReturn(selectedPartitions); + Mockito.when(scan.getTableSnapshot()).thenReturn(Optional.empty()); + Mockito.when(scan.getScanParams()).thenReturn(Optional.empty()); + Mockito.when(scan.getOutput()).thenReturn(ImmutableList.of(partitionSlot)); + Mockito.when(table.getRowCountForSelectedPartitions( + Mockito.eq(selectedPartitions), Mockito.any())).thenReturn(TableIf.UNKNOWN_ROW_COUNT); + Mockito.when(table.getRowCount()).thenReturn(1000L); + + Statistics statistics = new StatsCalculator((CascadesContext) null).computeFileScan(scan); + + Assertions.assertEquals(1000, statistics.getRowCount(), 0.001); + Assertions.assertTrue(statistics.getConjunctsAppliedToRowCount().isEmpty()); + } finally { + FeConstants.enableInternalSchemaDb = previous; + } + } + + @Test + public void testFileScanScalesNullCountToSelectedPartitionRows() { + ConnectContext previousContext = ConnectContext.get(); + ConnectContext connectContext = new ConnectContext(); + connectContext.setThreadLocalInfo(); + boolean previous = FeConstants.enableInternalSchemaDb; + FeConstants.enableInternalSchemaDb = true; + + Env env = Mockito.mock(Env.class); + StatisticsCache statisticsCache = Mockito.mock(StatisticsCache.class); + ExternalTable table = Mockito.mock(ExternalTable.class); + LogicalFileScan scan = Mockito.mock(LogicalFileScan.class); + SlotReference slot = new SlotReference("v", IntegerType.INSTANCE); + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions selectedPartitions = new SelectedPartitions( + 2, ImmutableMap.of("p1", p1, "p2", p2), false) + .withPruneResult(ImmutableMap.of("p1", p1), true, + ImmutableList.of(), ImmutableSet.of()); + ColumnStatistic columnStatistic = new ColumnStatisticBuilder(1_000_000) + .setNdv(500_000) + .setNumNulls(500_000) + .build(); + + Mockito.when(env.getStatisticsCache()).thenReturn(statisticsCache); + Mockito.when(statisticsCache.getColumnStatistics( + -1, -1, 1, -1, "v", connectContext)).thenReturn(columnStatistic); + Mockito.when(table.getId()).thenReturn(1L); + Mockito.when(table.getRowCountForSelectedPartitions( + Mockito.eq(selectedPartitions), Mockito.any())).thenReturn(1000L); + Mockito.when(scan.getTable()).thenReturn(table); + Mockito.when(scan.getSelectedPartitions()).thenReturn(selectedPartitions); + Mockito.when(scan.getTableSnapshot()).thenReturn(Optional.empty()); + Mockito.when(scan.getScanParams()).thenReturn(Optional.empty()); + Mockito.when(scan.getOutput()).thenReturn(ImmutableList.of(slot)); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + + Statistics statistics = new StatsCalculator((CascadesContext) null).computeFileScan(scan); + + Assertions.assertEquals(1000, statistics.getRowCount(), 0.001); + Assertions.assertEquals(500, statistics.findColumnStatistics(slot).ndv, 0.001); + Assertions.assertEquals(500, statistics.findColumnStatistics(slot).numNulls, 0.001); + } finally { + FeConstants.enableInternalSchemaDb = previous; + ConnectContext.remove(); + if (previousContext != null) { + previousContext.setThreadLocalInfo(); + } + } + } + // a, b are in (0,100) // a=200 and b=300 => output: 0 rows @Test @@ -250,9 +634,14 @@ public void testComputeOlapScanScalesNumNullsForSelectedPartitions() { Mockito.when(scan.getSelectedPartitionIds()).thenReturn(ImmutableList.of(selectedPartitionId)); Mockito.when(scan.getOutput()).thenReturn(ImmutableList.of(slot)); Mockito.when(scan.getVirtualColumns()).thenReturn(ImmutableList.of()); + EqualTo partitionPredicate = new EqualTo(slot, new IntegerLiteral(2)); + Mockito.when(scan.getPartitionPrunablePredicates()).thenReturn(Optional.of( + new PartitionPrunablePredicate(ImmutableSet.of(selectedPartitionId), + ImmutableList.of(slot), ImmutableSet.of(partitionPredicate)))); Mockito.when(table.getBaseIndexId()).thenReturn(baseIndexId); Mockito.when(table.getRowCountForIndex(baseIndexId, true)).thenReturn(12L); - Mockito.when(table.getRowCountForPartitionIndex(selectedPartitionId, baseIndexId, true)).thenReturn(4L); + Mockito.when(table.getRowCountForSelectedPartitions( + ImmutableList.of(selectedPartitionId), baseIndexId, 12D)).thenReturn(4D); Mockito.when(table.getPartitionNum()).thenReturn(3); Mockito.when(table.getPartition(selectedPartitionId)).thenReturn(selectedPartition); Mockito.when(table.getQualifiedDbName()).thenReturn("test"); @@ -273,6 +662,8 @@ public void testComputeOlapScanScalesNumNullsForSelectedPartitions() { Assertions.assertEquals(1, result.numNulls, 0.001); Assertions.assertEquals("2", result.minExpr.getStringValue()); Assertions.assertEquals("2", result.maxExpr.getStringValue()); + Assertions.assertEquals(ImmutableSet.of(partitionPredicate), + statistics.getConjunctsAppliedToRowCount()); } finally { ConnectContext.remove(); if (previousContext != null) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java index d49f3d97d3de7e..18f75240f92c10 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java @@ -19,6 +19,7 @@ import org.apache.doris.analysis.TableScanParams; import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.PartitionItem; import org.apache.doris.catalog.Type; import org.apache.doris.datasource.CatalogIf; import org.apache.doris.datasource.ExternalDatabase; @@ -26,13 +27,22 @@ import org.apache.doris.datasource.mvcc.PluginDrivenMvccExternalTable; import org.apache.doris.datasource.plugin.PluginDrivenExternalTable; import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.rules.expression.rules.SortedPartitionRanges; +import org.apache.doris.nereids.trees.copier.DeepCopierContext; +import org.apache.doris.nereids.trees.copier.LogicalPlanDeepCopier; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.ExprId; import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; import org.apache.doris.nereids.trees.plans.RelationId; import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; +import org.apache.doris.nereids.types.IntegerType; import org.apache.doris.qe.ConnectContext; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -45,6 +55,71 @@ public class LogicalFileScanTest { + @Test + public void withCachedOutputRebindsPartitionProofAndDiscardsStaleSortedRanges() { + SlotReference partitionSlot = new SlotReference(new ExprId(1), "dt", IntegerType.INSTANCE, true, + ImmutableList.of("db", "t")); + SlotReference outputSlot = new SlotReference(new ExprId(2), "dt", IntegerType.INSTANCE, true, + ImmutableList.of("db", "t")); + EqualTo partitionPredicate = new EqualTo(partitionSlot, new IntegerLiteral(1)); + EqualTo outputPredicate = new EqualTo(outputSlot, new IntegerLiteral(1)); + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions initial = new SelectedPartitions(2, ImmutableMap.of("p1", p1, "p2", p2), false, + false, Optional.of(Mockito.mock(SortedPartitionRanges.class))); + SelectedPartitions pruned = initial.withPruneResult( + ImmutableMap.of("p1", p1), true, ImmutableList.of(partitionSlot), + ImmutableSet.of(partitionPredicate)); + PluginDrivenExternalTable table = Mockito.mock(PluginDrivenExternalTable.class); + LogicalFileScan scan = new LogicalFileScan(new RelationId(1), table, ImmutableList.of("db"), pruned, + ImmutableList.of(), ImmutableList.of(), Optional.empty(), Optional.empty(), Optional.empty(), + Optional.empty(), Optional.empty(), "", Optional.of(ImmutableList.of(partitionSlot))); + + LogicalFileScan reboundScan = scan.withCachedOutput(ImmutableList.of(outputSlot)); + + Assertions.assertFalse(reboundScan.getSelectedPartitions().sortedPartitionRanges.isPresent()); + Assertions.assertEquals(ImmutableSet.of(outputPredicate), + reboundScan.getSelectedPartitions().getPrunableConjuncts()); + } + + @Test + public void deepCopyRebindsFileScanPartitionProof() { + PartitionItem p1 = Mockito.mock(PartitionItem.class); + PartitionItem p2 = Mockito.mock(PartitionItem.class); + SelectedPartitions initial = new SelectedPartitions( + 2, ImmutableMap.of("p1", p1, "p2", p2), false); + PluginDrivenExternalTable table = Mockito.mock(PluginDrivenExternalTable.class); + Mockito.when(table.getId()).thenReturn(1L); + Mockito.when(table.getName()).thenReturn("t"); + Mockito.when(table.initSelectedPartitions(Mockito.any())).thenReturn(initial); + Mockito.when(table.getFullSchema(Mockito.any())).thenReturn( + ImmutableList.of(new Column("dt", Type.INT, true))); + LogicalFileScan scan = new LogicalFileScan(new RelationId(1), table, ImmutableList.of("db"), + ImmutableList.of(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty()); + Slot partitionSlot = scan.getOutput().get(0); + SelectedPartitions pruned = initial.withPruneResult( + ImmutableMap.of("p1", p1), true, ImmutableList.of(partitionSlot), + ImmutableSet.of(new EqualTo(partitionSlot, new IntegerLiteral(1)))); + scan = scan.withSelectedPartitions(pruned); + + LogicalFileScan copied = (LogicalFileScan) LogicalPlanDeepCopier.INSTANCE.deepCopy( + scan, new DeepCopierContext()); + Slot copiedSlot = copied.getOutput().get(0); + + Assertions.assertNotEquals(partitionSlot.getExprId(), copiedSlot.getExprId()); + Assertions.assertEquals(ImmutableSet.of(new EqualTo(copiedSlot, new IntegerLiteral(1))), + copied.getSelectedPartitions().getPrunableConjuncts()); + } + + @Test + public void selectedPartitionsEqualityIncludesTotalPartitionCount() { + PartitionItem p1 = Mockito.mock(PartitionItem.class); + SelectedPartitions oneOfOne = new SelectedPartitions(1, ImmutableMap.of("p1", p1), true); + SelectedPartitions oneOfTwo = new SelectedPartitions(2, ImmutableMap.of("p1", p1), true); + + Assertions.assertNotEquals(oneOfOne, oneOfTwo); + } + @Test public void testComputeOutputIncludesInvisibleRowLineageColumnsForIcebergTable() { // Post-cutover a native iceberg table is a PluginDrivenExternalTable, so computeOutput() flows through diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java index 0195f136968418..2dcac5ba75a9cb 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java @@ -17,20 +17,31 @@ package org.apache.doris.nereids.trees.plans.logical; +import org.apache.doris.analysis.SlotRef; import org.apache.doris.catalog.Column; import org.apache.doris.catalog.MTMV; +import org.apache.doris.catalog.MaterializedIndexMeta; import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; import org.apache.doris.catalog.PrimitiveType; import org.apache.doris.catalog.stream.OlapTableStream; import org.apache.doris.mtmv.MTMVCache; +import org.apache.doris.nereids.trees.copier.DeepCopierContext; +import org.apache.doris.nereids.trees.copier.LogicalPlanDeepCopier; +import org.apache.doris.nereids.trees.expressions.EqualTo; import org.apache.doris.nereids.trees.expressions.Slot; import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; +import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.PreAggStatus; import org.apache.doris.nereids.trees.plans.RelationId; +import org.apache.doris.nereids.types.IntegerType; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.SessionVariable; import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; @@ -263,4 +274,128 @@ public void testWithSelectedPartitionIdsPreservesPartitionPredicateFlagForAllSca Assertions.assertTrue(copiedStreamScan.hasPartitionPredicate()); } + @Test + public void copyingPartitionPruningToRollupRebindsThePredicateProof() { + OlapTable table = Mockito.mock(OlapTable.class); + Mockito.when(table.getId()).thenReturn(1L); + Mockito.when(table.getName()).thenReturn("test_table"); + Mockito.when(table.getPartitionIds()).thenReturn(ImmutableList.of(1L)); + Mockito.when(table.getPartition(1L)).thenReturn(Mockito.mock(Partition.class)); + Mockito.when(table.getBaseIndexId()).thenReturn(10L); + + Column baseColumn = createColumn("p"); + Mockito.when(table.getBaseSchema(true)).thenReturn(ImmutableList.of(baseColumn)); + Column rollupColumn = createColumn("mv_p"); + SlotRef defineExpr = Mockito.mock(SlotRef.class); + Mockito.when(defineExpr.getColumnName()).thenReturn("p"); + rollupColumn.setDefineExpr(defineExpr); + MaterializedIndexMeta rollupMeta = Mockito.mock(MaterializedIndexMeta.class); + Mockito.when(rollupMeta.getSchema()).thenReturn(ImmutableList.of(rollupColumn)); + Mockito.when(table.getIndexMetaByIndexId(20L)).thenReturn(rollupMeta); + + LogicalOlapScan sourceScan = new LogicalOlapScan(new RelationId(1), table, ImmutableList.of("db")); + Slot baseSlot = sourceScan.getOutput().get(0); + PartitionPrunablePredicate proof = new PartitionPrunablePredicate( + ImmutableSet.of(1L), ImmutableList.of(baseSlot), + ImmutableSet.of(new EqualTo(baseSlot, new IntegerLiteral(1)))); + sourceScan = sourceScan.withSelectedPartitionIds(ImmutableList.of(1L), true) + .withPartitionPrunablePredicates(Optional.of(proof)); + LogicalOlapScan rollupScan = new LogicalOlapScan( + new RelationId(2), table, ImmutableList.of("db"), ImmutableList.of(), ImmutableList.of(1L), + 20L, PreAggStatus.unset(), ImmutableList.of(), ImmutableList.of(), Optional.empty(), + ImmutableList.of()); + + LogicalOlapScan reboundScan = rollupScan.withPartitionPruningFrom(sourceScan); + Slot rollupSlot = reboundScan.getOutput().get(0); + PartitionPrunablePredicate reboundProof = reboundScan.getPartitionPrunablePredicates().orElseThrow(); + + Assertions.assertNotEquals(baseSlot.getExprId(), rollupSlot.getExprId()); + Assertions.assertEquals(ImmutableList.of(rollupSlot), reboundProof.getPartitionSlots()); + Assertions.assertEquals(ImmutableSet.of(new EqualTo(rollupSlot, new IntegerLiteral(1))), + reboundProof.getPrunableConjuncts()); + } + + @Test + public void deepCopyRebindsPartitionProofToCopiedOutput() { + OlapTable table = Mockito.mock(OlapTable.class); + Mockito.when(table.getId()).thenReturn(1L); + Mockito.when(table.getName()).thenReturn("test_table"); + Mockito.when(table.getPartitionIds()).thenReturn(ImmutableList.of(1L)); + Mockito.when(table.getPartition(1L)).thenReturn(Mockito.mock(Partition.class)); + Mockito.when(table.getBaseIndexId()).thenReturn(10L); + Mockito.when(table.getBaseSchema(true)).thenReturn(ImmutableList.of(createColumn("p"))); + + LogicalOlapScan scan = new LogicalOlapScan(new RelationId(1), table, ImmutableList.of("db")); + Slot partitionSlot = scan.getOutput().get(0); + PartitionPrunablePredicate proof = new PartitionPrunablePredicate( + ImmutableSet.of(1L), ImmutableList.of(partitionSlot), + ImmutableSet.of(new EqualTo(partitionSlot, new IntegerLiteral(1)))); + scan = scan.withSelectedPartitionIds(ImmutableList.of(1L), true) + .withPartitionPrunablePredicates(Optional.of(proof)); + + LogicalOlapScan copied = (LogicalOlapScan) LogicalPlanDeepCopier.INSTANCE.deepCopy( + scan, new DeepCopierContext()); + Slot copiedSlot = copied.getOutput().get(0); + PartitionPrunablePredicate copiedProof = copied.getPartitionPrunablePredicates().orElseThrow(); + + Assertions.assertNotEquals(partitionSlot.getExprId(), copiedSlot.getExprId()); + Assertions.assertEquals(ImmutableList.of(copiedSlot), copiedProof.getPartitionSlots()); + Assertions.assertEquals(ImmutableSet.of(new EqualTo(copiedSlot, new IntegerLiteral(1))), + copiedProof.getPrunableConjuncts()); + } + + @Test + public void selectedPartitionUpdatesKeepOnlyCoveringProofs() { + OlapTable table = Mockito.mock(OlapTable.class); + Mockito.when(table.getId()).thenReturn(1L); + Mockito.when(table.getName()).thenReturn("test_table"); + Mockito.when(table.getPartitionIds()).thenReturn(ImmutableList.of(1L, 2L)); + Mockito.when(table.getPartition(1L)).thenReturn(Mockito.mock(Partition.class)); + Mockito.when(table.getPartition(2L)).thenReturn(Mockito.mock(Partition.class)); + Mockito.when(table.getBaseIndexId()).thenReturn(10L); + Mockito.when(table.getBaseSchema(true)).thenReturn(ImmutableList.of(createColumn("p"))); + LogicalOlapScan scan = new LogicalOlapScan(new RelationId(1), table, ImmutableList.of("db")) + .withSelectedPartitionIds(ImmutableList.of(1L), true); + Slot partitionSlot = scan.getOutput().get(0); + PartitionPrunablePredicate proof = new PartitionPrunablePredicate( + ImmutableSet.of(1L), ImmutableList.of(partitionSlot), + ImmutableSet.of(new EqualTo(partitionSlot, new IntegerLiteral(1)))); + scan = scan.withPartitionPrunablePredicates(Optional.of(proof)); + + Assertions.assertTrue(scan.withSelectedPartitionIds(ImmutableList.of()) + .getPartitionPrunablePredicates().isPresent()); + Assertions.assertFalse(scan.withSelectedPartitionIds(ImmutableList.of(1L, 2L)) + .getPartitionPrunablePredicates().isPresent()); + + OlapTableStream stream = Mockito.mock(OlapTableStream.class); + LogicalOlapTableStreamScan streamScan = (LogicalOlapTableStreamScan) scan.withPreSnapshot(Optional.of(stream)); + streamScan = streamScan.withSelectedPartitionIds(ImmutableList.of(1L), true) + .withPartitionPrunablePredicates(Optional.of(proof)); + Assertions.assertFalse(streamScan.withSelectedPartitionIds(ImmutableList.of(1L, 2L)) + .getPartitionPrunablePredicates().isPresent()); + } + + @Test + public void invalidatingPartitionPruningClearsTheRecordedPredicateProof() { + SlotReference partitionSlot = new SlotReference("p", IntegerType.INSTANCE); + PartitionPrunablePredicate proof = new PartitionPrunablePredicate( + ImmutableSet.of(), ImmutableList.of(partitionSlot), + ImmutableSet.of(new EqualTo(partitionSlot, new IntegerLiteral(1)))); + LogicalOlapScan scan = createMockScan(ImmutableList.of(partitionSlot)) + .withSelectedPartitionIds(ImmutableList.of(), true) + .withPartitionPrunablePredicates(Optional.of(proof)); + + Assertions.assertFalse(scan.withPartitionPruned(false).getPartitionPrunablePredicates().isPresent()); + + OlapTableStream stream = Mockito.mock(OlapTableStream.class); + LogicalOlapTableStreamScan streamScan = (LogicalOlapTableStreamScan) scan.withPreSnapshot(Optional.of(stream)); + streamScan = streamScan.withSelectedPartitionIds(ImmutableList.of(), true) + .withPartitionPrunablePredicates(Optional.of(proof)); + Assertions.assertTrue(streamScan.getPartitionPrunablePredicates().isPresent()); + + LogicalOlapTableStreamScan invalidated = streamScan.withPartitionPruned(false); + Assertions.assertFalse(invalidated.getPartitionPrunablePredicates().isPresent()); + Assertions.assertTrue(invalidated.isSnapshot()); + } + } From 38cb69462436fe3687c00a7a6700639151022b38 Mon Sep 17 00:00:00 2001 From: yangtao555 Date: Fri, 2 Oct 2026 19:02:23 +0800 Subject: [PATCH 2/2] fix test --- .../connector/iceberg/IcebergWritePlanProviderTest.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java index 17af0421186d70..11668b34450da6 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java @@ -1724,7 +1724,7 @@ public void planWriteThreadsPinnedReadSnapshotFromHandleToTransaction() { } @Test - public void planMergePreservesExplicitlyEmptyReadAcrossConcurrentFirstAppend() { + public void planMergeKeepsExplicitEmptyReadFencedAcrossConcurrentFirstAppend() { InMemoryCatalog catalog = freshCatalog(); TableIdentifier id = TableIdentifier.of("db1", "tv2"); Table empty = catalog.createTable(id, SCHEMA, PartitionSpec.unpartitioned(), @@ -1751,8 +1751,9 @@ public void planMergePreservesExplicitlyEmptyReadAcrossConcurrentFirstAppend() { providerFor(ops.table, ctx).planWrite(new WriteSession(txn), new WriteHandle(emptyPinnedHandle).writeOperation(WriteOperation.MERGE)); - Assertions.assertNull(txn.getBaseSnapshotId(), - "an explicitly empty read must leave RowDelta validation unbounded across the first append"); + Assertions.assertEquals(Long.valueOf(-1L), txn.getBaseSnapshotId(), + "an explicit empty read is an OCC fence (base -1), not an absent pin: the pinned empty " + + "generation must survive the concurrent first append instead of drifting to S1"); } // ───────────────────────────── MERGE sink (TIcebergMergeSink) ─────────────────────────────