Skip to content

[fix](stream) Mark streams stale when base tables are dropped #66287

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
seawinde wants to merge 8 commits into
base: master
Choose a base branch
from
Open

[fix](stream) Mark streams stale when base tables are dropped #66287

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
28af68e
[fix](stream) Mark streams stale when base tables are dropped
seawinde Jul 30, 2026
6282603
[fix](stream) Keep base table metadata consistent
seawinde Aug 3, 2026
cc88b06
[fix](stream) Resolve recycled base table qualifiers by ID
seawinde Aug 4, 2026
7ea0064
[fix](stream) Lock stream base table by stable ID
seawinde Aug 5, 2026
a7850ad
[fix](stream) Keep stream lock dependencies out of relation cache
seawinde Aug 10, 2026
9c8100c
[chore](stream) Document stream base lock handling
seawinde Aug 11, 2026
25d0c27
[fix](stream) Track stream base dependencies by identity
seawinde Aug 11, 2026
6f451cc
[fix](stream) Close stream recovery compatibility gaps
seawinde Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,16 @@ public boolean isRecycleTable(long dbId, long tableId) {
return isRecycleDatabase(dbId) || idToTable.containsKey(tableId);
}

public Table getRecycledTableNullable(long dbId, long tableId) {
readLock();
try {
RecycleTableInfo tableInfo = idToTable.get(tableId);
return tableInfo != null && tableInfo.getDbId() == dbId ? tableInfo.getTable() : null;
} finally {
readUnlock();
}
}

public boolean isRecyclePartition(long dbId, long tableId, long partitionId) {
return isRecycleTable(dbId, tableId) || idToPartition.containsKey(partitionId);
}
Expand Down
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,10 @@ public void unmarkDropped() {
isDropped = false;
}

public boolean isDropped() {
return isDropped;
}

public void readLock() {
this.rwLock.readLock().lock();
}
Expand Down
7 changes: 1 addition & 6 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4590,12 +4590,7 @@ public static void getDdlStmt(Command command, String dbName, TableIf table, Lis

sb.append("CREATE STREAM ");
sb.append('`').append(table.getName()).append('`').append('\n');
TableIf baseTable = stream.getBaseTableNullable();
if (baseTable != null) {
sb.append("ON TABLE ").append(baseTable.getNameWithFullQualifiers());
} else {
sb.append("ON TABLE ").append("UNKNOWN");
}
sb.append("ON TABLE ").append(String.join(".", stream.getBaseTableFullQualifiers()));
// (COMMENT STRING_LITERAL)?
addTableComment(table, sb);
// properties=propertyClause?
Expand Down
69 changes: 62 additions & 7 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/stream/BaseTableStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
package org.apache.doris.catalog.stream;

import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.UserException;
Expand All @@ -37,6 +39,8 @@
import java.util.Map;

public abstract class BaseTableStream extends Table {
private static final String BASE_TABLE_NOT_FOUND_STALE_REASON = "Base table does not exist";

public enum StreamScanType {
APPEND_ONLY,
MIN_DELTA,
Expand Down Expand Up @@ -113,10 +117,33 @@ public BaseTableStream(String streamName, List<Column> fullSchema, TableIf baseT
}

public TableIf getBaseTableNullable() {
if (baseTable == null) {
baseTable = baseTableInfo.getTableNullable();
TableIf cachedBaseTable = baseTable;
if (isBaseTableAvailable(cachedBaseTable)) {
return cachedBaseTable;
}
if (cachedBaseTable != null) {
baseTable = null;
}
TableIf resolvedBaseTable = baseTableInfo.getTableNullable();
if (!isBaseTableAvailable(resolvedBaseTable)) {
return null;
}
return baseTable;
baseTable = resolvedBaseTable;
return resolvedBaseTable;
}

private boolean isBaseTableAvailable(TableIf candidate) {
if (candidate == null || candidate instanceof Table && ((Table) candidate).isDropped) {
return false;
}
if (!baseTableInfo.isInternalTable()) {
return true;
}
// Table recovery publishes into database maps before clearing the table flag, while database recovery clears
// member-table flags before publishing the database. Require both catalog mappings to reject either window.
Database database = Env.getCurrentInternalCatalog().getDbNullable(baseTableInfo.getDbId());
return database != null && !database.isDropped()
&& database.getTable(baseTableInfo.getTableId()).orElse(null) == candidate;
}

public void setProperties(Map<String, String> properties) throws org.apache.doris.common.AnalysisException {
Expand All @@ -139,23 +166,35 @@ public StreamScanType getStreamScanType() {
}

public boolean isDisabled() {
return disabled;
return isDisabled(getBaseTableNullable());
}

boolean isDisabled(TableIf availableBaseTable) {
return disabled || availableBaseTable == null;
}

public void setDisabled(boolean disabled) {
this.disabled = disabled;
}

public boolean isStale() {
return stale;
return isStale(getBaseTableNullable());
}

boolean isStale(TableIf availableBaseTable) {
return stale || availableBaseTable == null;
}

public void setStale(boolean stale) {
this.stale = stale;
}

public String getStaleReason() {
return staleReason;
return getStaleReason(getBaseTableNullable());
}

String getStaleReason(TableIf availableBaseTable) {
return availableBaseTable == null ? BASE_TABLE_NOT_FOUND_STALE_REASON : staleReason;
}

public void setStaleReason(String staleReason) {
Expand Down Expand Up @@ -198,7 +237,23 @@ public TableIf getBaseTableOrNereidsAnalysisException() throws AnalysisException
}

public List<String> getBaseTableFullQualifiers() {
return baseTableInfo.getFullQualifiers();
return getBaseTableFullQualifiers(getBaseTableNullable());

@github-actions github-actions Bot Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Collect the stable-ID base object, not its display name

Create s on A as orig, rename the same-ID A to old, recoverably drop A, then create B named old. This new recycle-backed display helper returns old, so CollectRelation caches B and StatementContext.lock() locks B; before this diff it returned persisted orig, so collection failed instead. A concurrent RECOVER TABLE old AS new can publish/unmark A while recovery still owns A's write lock (it takes database/recycle/A locks, not B or the stream), and BindRelation.makeOlapTableStreamScan() then resolves A by the persisted ID. The wrapper now reads A even though A is absent from plannerResources; a concurrent drop can even make getStreamUpdate() re-resolve null and dereference .getPartition(...).

Please have collection resolve/cache the exact stable-ID base object (failing while that ID is unavailable), then reuse or fence that same snapshot during binding. Cover orig -> old -> drop -> B(old) with a latch-controlled collect/lock/recover-as/bind test and an end-to-end replacement query.

@seawinde seawinde Aug 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in a0238b249e6. collectFromTableStream() now resolves the base table through getBaseTableNullable(), which uses the persisted stable (dbId, tableId), and fails collection when that exact base object is unavailable instead of resolving a same-name replacement through the display qualifiers. The resolved base object is then included in planner locking before binding. The separate qualifier-key collision reported in r3746321533 is distinct and still needs a follow-up fix.

}

List<String> getBaseTableFullQualifiers(TableIf availableBaseTable) {
TableIf displayBaseTable = availableBaseTable;
if (displayBaseTable == null && baseTableInfo.isInternalTable()) {
displayBaseTable = Env.getCurrentRecycleBin().getRecycledTableNullable(
baseTableInfo.getDbId(), baseTableInfo.getTableId());
}
if (baseTableInfo.isInternalTable()) {
return ImmutableList.of(
baseTableInfo.getCtlName(),
Env.getCurrentInternalCatalog().getDb(baseTableInfo.getDbId())
.map(db -> db.getFullName()).orElse(baseTableInfo.getDbName()),
displayBaseTable == null ? baseTableInfo.getTableName() : displayBaseTable.getName());
}
return displayBaseTable == null ? baseTableInfo.getFullQualifiers() : displayBaseTable.getFullQualifiers();

@github-actions github-actions Bot Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Resolve recycled qualifiers without the detached table's database pointer

This is distinct from the existing same-ID table-rename thread: after DROP TABLE db.base, ALTER DATABASE db RENAME db2 updates qualifiedDbName only for active Database.idToTable entries. The recycled base therefore still resolves getDatabase() through db, which no longer exists. This line then calls the default TableIf.getFullQualifiers() and dereferences getDatabase().getCatalog(), so one stale stream aborts information_schema.table_streams, SHOW CREATE STREAM, and stream collection with an NPE instead of reporting disabled/stale. The stale name also survives image/replay; after recycle erasure, the descriptor fallback still reports db.

Derive the internal catalog/database qualifiers from stable IDs/current live-or-recycled database identity, and preserve only the latest table name independently of the recycled object's lifetime; cover drop-base -> rename-database, reload, and recycle erasure.

@seawinde seawinde Aug 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed the detached-table failure in d2b13a42cf1. Internal base-table qualifiers are now assembled without calling getFullQualifiers() on the recycled table: the database name is resolved by the stable dbId, and only the table name is read from the live/recycled table, with the persisted creation-time names as the final fallback. This keeps table_streams and SHOW CREATE STREAM valid after DROP TABLE followed by ALTER DATABASE ... RENAME, including Gson reload and recycle-entry erasure. Added testBaseTableQualifiersFollowDatabaseRenameAfterDrop; ./run-fe-ut.sh --run org.apache.doris.catalog.DropTableStreamTest passes all 6 tests.

}

public TableStreamBaseTableInfo getBaseTableInfo() {
Expand Down
2 changes: 1 addition & 1 deletion fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBaseTableInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ public TableIf getTableNullable() {
}
}
}
LOG.warn("invalid base table: {}", this);
LOG.debug("invalid base table: {}", this);
return null;
}

Expand Down
36 changes: 13 additions & 23 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -350,32 +350,22 @@ public void fillTableStreamValuesMetadataResult(List<TRow> dataBatch) {
// STREAM_COMMENT
trow.addToColumnValue(new TCell().setStringVal(stream.getComment()));
TableIf baseTable = stream.getBaseTableNullable();
if (baseTable == null) {
// BASE_TABLE_NAME
trow.addToColumnValue(new TCell().setStringVal("N/A"));
// BASE_TABLE_DB
trow.addToColumnValue(new TCell().setStringVal("N/A"));
// BASE_TABLE_CTL
trow.addToColumnValue(new TCell().setStringVal("N/A"));
// BASE_TABLE_TYPE
trow.addToColumnValue(new TCell().setStringVal("N/A"));
} else {
List<String> baseTableQualifiers = baseTable.getFullQualifiers();
// BASE_TABLE_NAME
trow.addToColumnValue(new TCell().setStringVal(baseTableQualifiers.get(2)));
// BASE_TABLE_DB
trow.addToColumnValue(new TCell().setStringVal(baseTableQualifiers.get(1)));
// BASE_TABLE_CTL
trow.addToColumnValue(new TCell().setStringVal(baseTableQualifiers.get(0)));
// BASE_TABLE_TYPE
trow.addToColumnValue(new TCell().setStringVal(baseTable.getType().name()));
}
List<String> baseTableQualifiers = stream.getBaseTableFullQualifiers(baseTable);
// BASE_TABLE_NAME

@github-actions github-actions Bot Jul 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Use live qualifiers while the base table is available

TableStreamBaseTableInfo keeps the creation-time names but resolves internal tables by ID. After a supported same-ID rename (ALTER TABLE old RENAME new or RECOVER TABLE old AS new), this line still emits old even though the stream resolves successfully and is reported enabled; SHOW CREATE STREAM emits new, and callers filtering table_streams by the current base name miss the stream. Resolve baseTable first and use baseTable.getFullQualifiers() when it is non-null, falling back to the stored qualifiers only for the unavailable-table case.

@seawinde seawinde Aug 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 6baebd9e78a. table_streams now resolves the base table first and uses baseTable.getFullQualifiers() while it is available. Stored qualifiers are only the final fallback when neither the live table nor a same-ID recycle-bin entry is available. The unit test covers ALTER TABLE ... RENAME and RECOVER TABLE ... AS.

trow.addToColumnValue(new TCell().setStringVal(baseTableQualifiers.get(2)));
// BASE_TABLE_DB
trow.addToColumnValue(new TCell().setStringVal(baseTableQualifiers.get(1)));
// BASE_TABLE_CTL
trow.addToColumnValue(new TCell().setStringVal(baseTableQualifiers.get(0)));
// BASE_TABLE_TYPE
trow.addToColumnValue(new TCell().setStringVal(

@github-actions github-actions Bot Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Derive the row from one availability snapshot

baseTable is reused only for BASE_TABLE_TYPE; isDisabled(), isStale(), and getStaleReason() each resolve it again. Because base-table drop/recovery uses different locks from this stream read lock, a recovery after the type lookup can emit N/A with enabled/non-stale state, while a drop can emit OLAP with disabled/stale state. Resolve availability once per row and derive all four fields from that snapshot while still combining the persisted flags; a latch-controlled DDL test would cover the interleaving.

@seawinde seawinde Aug 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 6baebd9e78a. fillTableStreamValuesMetadataResult() resolves baseTable once per TRow and passes that snapshot to the qualifier, disabled, stale, and stale-reason calculations. The overloads still combine the snapshot with the persisted disabled/stale flags, so one row can no longer mix pre-drop/post-drop or pre-recovery/post-recovery availability.

baseTable == null ? "N/A" : baseTable.getType().name()));
// ENABLED
trow.addToColumnValue(new TCell().setBoolVal(!stream.isDisabled()));
trow.addToColumnValue(new TCell().setBoolVal(!stream.isDisabled(baseTable)));
// IS_STALE
trow.addToColumnValue(new TCell().setBoolVal(stream.isStale()));
trow.addToColumnValue(new TCell().setBoolVal(stream.isStale(baseTable)));
// STALE_REASON
trow.addToColumnValue(new TCell().setStringVal(stream.getStaleReason()));
trow.addToColumnValue(new TCell().setStringVal(stream.getStaleReason(baseTable)));
dataBatch.add(trow);
} finally {
stream.readUnlock();
Expand Down
7 changes: 6 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,10 @@ public static Pair<Set<TableIf>, Set<TableIf>> getBaseTableFromQuery(String quer
try {
NereidsPlanner planner = new NereidsPlanner(ctx.getStatementContext());
planner.planWithLock(logicalPlan, PhysicalProperties.ANY, ExplainLevel.ANALYZED_PLAN);
return Pair.of(Sets.newHashSet(ctx.getStatementContext().getTables().values()),
Set<TableIf> baseTables = Sets.newHashSet(ctx.getStatementContext().getTables().values());
// Implicit dependencies are all-level tables, not relations written at the first query level.
baseTables.addAll(ctx.getStatementContext().getImplicitTableDependencies());

@github-actions github-actions Bot Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Migrate persisted stream dependencies on upgrade

These additions repair newly analyzed creation and refresh, but an MTMV image can already contain stream S without stable base T: old dependency collection could name-resolve replacement B while the stream wrapper still scanned its cached T, then persist a successful refresh as {S, B} with no T snapshot. On load, MTMVRelation.compatible() only normalizes stored entries and MTMV.compatible() re-registers them unchanged. T therefore has no commit-event edge or recorded snapshot, while freshness treats S as synchronous; the MTMV can remain rewrite-eligible with stale rows until a later refresh succeeds. Please expand persisted stream dependencies before registration (or conservatively invalidate them) and add an old-image upgrade test.

return Pair.of(baseTables,
Sets.newHashSet(ctx.getStatementContext().getOneLevelTables().values()));
} finally {
ctx.setStatementContext(original);
Expand Down Expand Up @@ -448,6 +451,8 @@ public static MTMVAnalyzeQueryInfo analyzeQuery(ConnectContext ctx, Map<String,
}

Set<TableIf> baseTables = Sets.newHashSet(statementContext.getTables().values());
// Implicit dependencies are all-level tables, not relations written at the first query level.
baseTables.addAll(statementContext.getImplicitTableDependencies());
Set<TableIf> oneLevelTables = Sets.newHashSet(statementContext.getOneLevelTables().values());
for (TableIf table : baseTables) {
if (table.isTemporary()) {
Expand Down
33 changes: 33 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@

package org.apache.doris.mtmv;

import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.stream.BaseTableStream;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.datasource.CatalogMgr;
import org.apache.doris.persist.gson.GsonPostProcessable;

Expand Down Expand Up @@ -104,6 +107,12 @@ public void compatible(CatalogMgr catalogMgr) throws Exception {
compatible(catalogMgr, baseTables);
compatible(catalogMgr, baseViews);
compatible(catalogMgr, baseTablesOneLevel);
addStreamBaseTables(baseTables);
if (CollectionUtils.isEmpty(baseTablesOneLevelAndFromView)) {
// Preserve the existing fallback for older images in a separate set before adding implicit stream bases.
baseTablesOneLevelAndFromView = new HashSet<>(getBaseTablesOneLevel());
}
addStreamBaseTables(baseTablesOneLevelAndFromView);
}

private void compatible(CatalogMgr catalogMgr, Set<BaseTableInfo> infos) throws Exception {
Expand All @@ -114,4 +123,28 @@ private void compatible(CatalogMgr catalogMgr, Set<BaseTableInfo> infos) throws
baseTableInfo.compatible(catalogMgr);
}
}

private void addStreamBaseTables(Set<BaseTableInfo> infos) throws Exception {
if (CollectionUtils.isEmpty(infos)) {
return;
}
// Older images may contain only the stream relation; add its stable base so freshness and invalidation survive
// an upgrade without inventing a historical snapshot for the newly discovered dependency.
for (BaseTableInfo info : new HashSet<>(infos)) {
TableIf table;
try {
table = MTMVUtil.getTable(info);
} catch (AnalysisException e) {

@github-actions github-actions Bot Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Do not complete migration while the stream is missing

For an old image where M records stream S but not its base T, recoverably drop S before loading the image. MTMVUtil.getTable(info) throws here, but this continue lets compatibility succeed and re-register only the old S relation. After same-name RECOVER TABLE S, compatibility is not rerun; freshness treats the recovered stream as synchronous, while M still has neither a snapshot nor an invalidation edge for T, so writes to T can leave stale rows rewrite-eligible. This is a residual branch in the newly added migration, distinct from the prior available-stream case. Please fail compatibility or persist a pending migration when the relation cannot be resolved, or rerun expansion before a recovered stream can make the MTMV a candidate, and add an old-image drop/recover-stream test.

continue;
}
if (table instanceof BaseTableStream) {
TableIf baseTable = ((BaseTableStream) table).getBaseTableNullable();
if (baseTable == null) {
throw new AnalysisException(
"Failed to resolve stream base table during MTMV compatibility: " + info);
}
infos.add(new BaseTableInfo(baseTable));
}
}
}
}
30 changes: 23 additions & 7 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,9 @@ public enum TableFrom {

// tables in this query directly
private final Map<List<String>, TableIf> tables = Maps.newHashMap();
// Underlying tables resolved while collecting explicit relations. They are not independently named by SQL, so
// keep them out of the qualifier-keyed relation maps and reuse the same snapshot through planning.
private final Set<TableIf> implicitTableDependencies = Sets.newIdentityHashSet();
// onelevel tables in this query directly,
// if
// create v1 as select * from t1
Expand Down Expand Up @@ -442,6 +445,14 @@ public Map<List<String>, TableIf> getTables() {
return tables;
}

public void addImplicitTableDependency(TableIf table) {
implicitTableDependencies.add(table);
}

public Set<TableIf> getImplicitTableDependencies() {
return implicitTableDependencies;
}

public Map<List<String>, TableIf> getOneLevelTables() {
return oneLevelTables;
}
Expand Down Expand Up @@ -959,16 +970,20 @@ public Map<RelationId, Statistics> getRelationIdToStatisticsMap() {
*/
public synchronized void lock() {
if (!needLockTables
|| (tables.isEmpty() && mtmvRelatedTables.isEmpty() && insertTargetTables.isEmpty())
|| (tables.isEmpty() && mtmvRelatedTables.isEmpty() && insertTargetTables.isEmpty()
&& implicitTableDependencies.isEmpty())
|| !plannerResources.isEmpty()) {
return;
}
// The same object can be both an explicit relation and an implicit dependency; lock it only once.
Set<TableIf> tablesToLock = Sets.newIdentityHashSet();
tablesToLock.addAll(tables.values());
tablesToLock.addAll(mtmvRelatedTables.values());
tablesToLock.addAll(insertTargetTables.values());
tablesToLock.addAll(implicitTableDependencies);
PriorityQueue<TableIf> tableIfs = new PriorityQueue<>(
tables.size() + mtmvRelatedTables.size() + insertTargetTables.size(),
Comparator.comparing(TableIf::getId));
tableIfs.addAll(tables.values());
tableIfs.addAll(mtmvRelatedTables.values());
tableIfs.addAll(insertTargetTables.values());
tablesToLock.size(), Comparator.comparing(TableIf::getId));
tableIfs.addAll(tablesToLock);
while (!tableIfs.isEmpty()) {
TableIf tableIf = tableIfs.poll();
if (!tableIf.needReadLockWhenPlan()) {
Expand Down Expand Up @@ -1276,7 +1291,8 @@ public int getExternalTablePreloadCandidateCount() {
public boolean hasAnyPlanReadLockTable() {
return containsPlanReadLockTable(tables.values())
|| containsPlanReadLockTable(mtmvRelatedTables.values())
|| containsPlanReadLockTable(insertTargetTables.values());
|| containsPlanReadLockTable(insertTargetTables.values())
|| containsPlanReadLockTable(implicitTableDependencies);
}

public Optional<ExternalMetadataPreloadResult> getExternalMetadataPreloadResult() {
Expand Down
16 changes: 9 additions & 7 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CollectRelation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,9 +235,8 @@ private void collectFromUnboundRelation(CascadesContext cascadesContext,
if (table instanceof View) {
parseAndCollectFromView(tableQualifier, (View) table, cascadesContext);
}
// we need to collect stream table's base table as well
if (table instanceof BaseTableStream) {
collectFromTableStream((BaseTableStream) table, cascadesContext, tableFrom, unboundRelation);
collectFromTableStream((BaseTableStream) table, cascadesContext.getStatementContext());
}
}

Expand Down Expand Up @@ -313,10 +312,13 @@ protected void parseAndCollectFromView(List<String> tableQualifier, View view, C
parentContext.addPlanProcesses(viewContext.getPlanProcesses());
}

private void collectFromTableStream(BaseTableStream tableStream, CascadesContext cascadesContext,
TableFrom tableFrom, Optional<UnboundRelation> unboundRelation) {
StatementContext statementContext = cascadesContext.getConnectContext().getStatementContext();
List<String> tableQualifier = tableStream.getBaseTableFullQualifiers();
statementContext.getAndCacheTable(tableQualifier, tableFrom, unboundRelation);
private void collectFromTableStream(BaseTableStream tableStream, StatementContext statementContext) {
// Capture the stable-ID result once so preload, locking, and dependency tracking use the same table object.
TableIf baseTable = tableStream.getBaseTableNullable();
if (baseTable == null) {
throw new AnalysisException("Table ["
+ tableStream.getBaseTableFullQualifiers().get(2) + "] does not exist");
}
statementContext.addImplicitTableDependency(baseTable);
}
}
Loading
Loading