This is an automated email from the ASF dual-hosted git repository.
luwei16 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f311f80e404 [fix](binlog) Fix partition pruning for incremental
queries with tablet IDs (#69494)
f311f80e404 is described below
commit f311f80e4042c7c69f1c94a8236e6780224a8831
Author: Luwei <[email protected]>
AuthorDate: Wed Oct 28 16:06:46 2026 +0800
[fix](binlog) Fix partition pruning for incremental queries with tablet IDs
(#69494)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Incremental queries on RANGE or LIST partitioned tables
can return empty results when TABLET(...) specifies valid tablets from a
separate changelog index. Partition pruning enumerates the physical base
index instead of the changelog wrapper's index, removes the matching
partitions, and produces an empty relation. Resolve the index through
the table's existing getPartitionIndex API, consistent with scan-range
generation and its base-tablet compatibility fallback.
### Release note
Fix empty results from incremental queries on partitioned changelog
tables when changelog tablets are explicitly selected.
### Check List (For Author)
- Test: Unit Test
- FE_UT_PARALLEL=1 bash run-fe-ut.sh --run
'PruneOlapScanPartitionTest,OlapTableWrapperTest,PruneEmptyPartitionTest'
(21 passed)
- The RANGE/LIST queries without WHERE reproduced LogicalEmptyRelation
before the fix.
- Cover explicit partitions and filters, multiple tablets, invalid
tablets, ordinary scans, base-tablet fallback, and generated scan
tablets.
- FE Core Checkstyle and git diff --check
- Behavior changed: Yes (valid changelog tablet selections retain their
partitions)
- Does this need documentation: No
---
.../rules/rewrite/PruneOlapScanPartition.java | 2 +-
.../rules/rewrite/PruneOlapScanPartitionTest.java | 109 +++++++++++++++++++++
2 files changed, 110 insertions(+), 1 deletion(-)
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 ebc994a83e3..4ac8b61e14b 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
@@ -237,7 +237,7 @@ public class PruneOlapScanPartition implements
RewriteRuleFactory {
if (!selectedPartitions.isEmpty() &&
!selectedPartitions.contains(partition.getId())) {
continue;
}
- MaterializedIndex baseIndex = partition.getBaseIndex();
+ MaterializedIndex baseIndex = table.getPartitionIndex(partition,
table.getBaseIndexId());
for (Tablet tablet : baseIndex.getTablets()) {
if (manuallySpecifiedTabletIds.contains(tablet.getId())) {
selectPartitionIds.add(partition.getId());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartitionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartitionTest.java
index 9d9c62af572..633c37b4df3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartitionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PruneOlapScanPartitionTest.java
@@ -21,19 +21,34 @@ import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.nereids.util.MemoPatternMatchSupported;
import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.planner.OlapScanNode;
+import org.apache.doris.tso.MasterTsoProvider;
+import org.apache.doris.tso.TSOTimestamp;
import org.apache.doris.utframe.TestWithFeService;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
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;
+
+import java.util.List;
+import java.util.Locale;
class PruneOlapScanPartitionTest extends TestWithFeService implements
MemoPatternMatchSupported {
+ private static final long CHANGELOG_REFERENCE_TSO =
TSOTimestamp.composePhysicalTimestamp(1700000000001L);
+
@Override
protected void runBeforeAll() throws Exception {
createDatabase("test");
@@ -202,6 +217,77 @@ class PruneOlapScanPartitionTest extends TestWithFeService
implements MemoPatter
}
+ @ParameterizedTest
+ @CsvSource({"RANGE, false", "LIST, false", "RANGE, true", "LIST, true"})
+ void testChangelogTabletPartitionPruning(String partitionType, boolean
reuseBaseTablets) throws Exception {
+ String tableName = "test_changelog_tablet_" +
partitionType.toLowerCase(Locale.ROOT) + "_" + reuseBaseTablets;
+ String partitions = partitionType.equals("RANGE")
+ ? "partition by range(part) (partition p1 values less than
('10'), "
+ + "partition p2 values less than ('20')) "
+ : "partition by list(part) (partition p1 values in ('1'),
partition p2 values in ('11')) ";
+ boolean originalEnableFeatureChangelog =
Config.enable_feature_changelog;
+ Config.enable_feature_changelog = true;
+ try {
+ createTable("create table " + tableName + " (id int, part int not
null, v int) "
+ + "duplicate key(id, part) " + partitions
+ + "distributed by hash(id) buckets 1 "
+ + "properties ('replication_num'='1',
'changelog.enable'='true')");
+ } finally {
+ Config.enable_feature_changelog = originalEnableFeatureChangelog;
+ }
+
+ Database db =
Env.getCurrentInternalCatalog().getDbOrMetaException("test");
+ OlapTable table = (OlapTable) db.getTableOrMetaException(tableName);
+ Partition p1 = table.getPartition("p1");
+ Partition p2 = table.getPartition("p2");
+ long changelogIndexId = table.getChangelogIndexMeta().getIndexId();
+ long baseTabletId = p1.getBaseIndex().getTablets().get(0).getId();
+ Assertions.assertNotEquals(baseTabletId,
p1.getIndex(changelogIndexId).getTablets().get(0).getId());
+ if (reuseBaseTablets) {
+ // Older metadata has the changelog schema but no separate
partition index.
+ table.writeLock();
+ try {
+ p1.deleteRollupIndex(changelogIndexId);
+ p2.deleteRollupIndex(changelogIndexId);
+ } finally {
+ table.writeUnlock();
+ }
+ }
+ long physicalIndexId = reuseBaseTablets ? table.getBaseIndexId() :
changelogIndexId;
+ long tablet1 =
p1.getIndex(physicalIndexId).getTablets().get(0).getId();
+ long tablet2 =
p2.getIndex(physicalIndexId).getTablets().get(0).getId();
+ String incr = "select * from " + tableName +
"@incr('incrementType'='DETAIL')";
+ String oneTablet = " tablet(" + tablet1 + ")";
+ String twoTablets = " tablet(" + tablet1 + ", " + tablet2 + ")";
+
+ // No WHERE clause is needed to trigger tablet-based partition pruning.
+ checkSelectedPartitions(incr + oneTablet, changelogIndexId,
ImmutableList.of(p1.getId()));
+ checkSelectedPartitions(incr + twoTablets, changelogIndexId,
ImmutableList.of(p1.getId(), p2.getId()));
+ checkSelectedPartitions(incr + twoTablets + " where part = 1",
changelogIndexId,
+ ImmutableList.of(p1.getId()));
+ checkSelectedPartitions(incr + " partition(p1)" + twoTablets,
changelogIndexId,
+ ImmutableList.of(p1.getId()));
+ checkSelectedPartitions(incr + oneTablet + " where part = 11",
changelogIndexId, ImmutableList.of());
+ checkSelectedPartitions(incr + " partition(p2)" + oneTablet,
changelogIndexId, ImmutableList.of());
+ checkSelectedPartitions(incr + " tablet(" + Long.MAX_VALUE + ")",
changelogIndexId, ImmutableList.of());
+ // APPEND_ONLY adds an internal filter even without a user WHERE
clause.
+ checkSelectedPartitions(incr.replace("DETAIL", "APPEND_ONLY") +
oneTablet, changelogIndexId,
+ ImmutableList.of(p1.getId()));
+ checkSelectedPartitions("select * from " + tableName + " tablet(" +
baseTabletId + ")",
+ table.getBaseIndexId(), ImmutableList.of(p1.getId()));
+ if (!reuseBaseTablets) {
+ checkSelectedPartitions(incr + " tablet(" + baseTabletId + ")",
changelogIndexId, ImmutableList.of());
+ }
+
+ // EXPLAIN builds scan ranges without waiting for a live changelog
visibility fence.
+ try (MockedStatic<MasterTsoProvider> mockedTso =
Mockito.mockStatic(MasterTsoProvider.class)) {
+ mockedTso.when(() ->
MasterTsoProvider.getCurrentTso(connectContext)).thenReturn(CHANGELOG_REFERENCE_TSO);
+ OlapScanNode scanNode = (OlapScanNode) getSQLPlanner("explain " +
incr + oneTablet).getScanNodes().get(0);
+ Assertions.assertEquals(changelogIndexId,
scanNode.getSelectedIndexId());
+ Assertions.assertEquals(ImmutableList.of(tablet1),
scanNode.getScanTabletIds());
+ }
+ }
+
@Test
void testOlapScanPartitionPruneWithMultiColumnCase() throws Exception {
createTable("create table
testOlapScanPartitionPruneWithMultiColumnCase("
@@ -466,6 +552,29 @@ class PruneOlapScanPartitionTest extends TestWithFeService
implements MemoPatter
}
+ private PlanChecker parseWithChangelogReference(String sql) {
+ PlanChecker checker = PlanChecker.from(connectContext).parse(sql);
+ // Direct analyzer tests bypass the planner's statement reference
initialization.
+
checker.getCascadesContext().getStatementContext().getOrRegisterChangelogReferenceTso(
+ () -> CHANGELOG_REFERENCE_TSO);
+ return checker;
+ }
+
+ private void checkSelectedPartitions(String sql, long indexId, List<Long>
partitionIds) {
+ PlanChecker checker =
parseWithChangelogReference(sql).analyze().rewrite();
+ if (partitionIds.isEmpty()) {
+ checker.matches(logicalEmptyRelation());
+ } else {
+ checker.matches(logicalOlapScan().when(scan -> {
+ Assertions.assertEquals(indexId, scan.getSelectedIndexId());
+ Assertions.assertEquals(ImmutableSet.copyOf(partitionIds),
+ ImmutableSet.copyOf(scan.getSelectedPartitionIds()));
+ Assertions.assertEquals(scan.getManuallySpecifiedTabletIds(),
scan.getSelectedTabletIds());
+ return true;
+ }));
+ }
+ }
+
private void test(String table, String filter, int expectScanPartitionNum)
{
test(table, filter, "", expectScanPartitionNum);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
