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 da8feed859d [fix](binlog) Report streams with missing base tables as
stale (#68087)
da8feed859d is described below
commit da8feed859d1d9e282a86742d8219cd3a31e694a
Author: Luwei <[email protected]>
AuthorDate: Fri Sep 18 10:29:48 2026 +0800
[fix](binlog) Report streams with missing base tables as stale (#68087)
### What problem does this PR solve?
Issue Number: None
Related PR: #67173
Problem Summary: Dropping a base table leaves its table stream object in
metadata, but the table stream metadata function previously reported the
retained stream as enabled and non-stale even though its original base
table ID could no longer be resolved. This change derives the displayed
health from the same base-table lookup used for the identity columns,
while preserving the stream configured state so recovering the original
table restores its health.
### Release note
Table streams whose base table is unavailable are now reported as
disabled and stale with an explicit reason.
### Check List (For Author)
- Test: Unit Test
- DropTableStreamTest
- TableStreamManagerCleanupTest
- TableStreamBaseTableInfoTest
- Behavior changed: Yes. Missing-base streams are reported as disabled
and stale.
- Does this need documentation: No
---
.../doris/catalog/stream/TableStreamManager.java | 12 ++--
.../apache/doris/catalog/DropTableStreamTest.java | 69 ++++++++++++++++++++++
2 files changed, 77 insertions(+), 4 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
index f768f5afeb9..35275748397 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java
@@ -63,6 +63,7 @@ import java.util.concurrent.locks.LockSupport;
public class TableStreamManager extends MasterDaemon implements Writable,
GsonPostProcessable {
private static final Logger LOG =
LogManager.getLogger(TableStreamManager.class);
+ private static final String BASE_TABLE_NOT_FOUND_STALE_REASON = "Base
table does not exist";
@SerializedName(value = "dbStreamMap")
private Map<Long, Set<Long>> dbStreamMap;
protected MonitoredReentrantReadWriteLock rwLock;
@@ -356,7 +357,8 @@ public class TableStreamManager extends MasterDaemon
implements Writable, GsonPo
// STREAM_COMMENT
trow.addToColumnValue(new
TCell().setStringVal(stream.getComment()));
TableIf baseTable = stream.getBaseTableNullable();
- if (baseTable == null) {
+ boolean baseTableExists = baseTable != null;
+ if (!baseTableExists) {
// BASE_TABLE_NAME
trow.addToColumnValue(new
TCell().setStringVal("N/A"));
// BASE_TABLE_DB
@@ -377,11 +379,13 @@ public class TableStreamManager extends MasterDaemon
implements Writable, GsonPo
trow.addToColumnValue(new
TCell().setStringVal(baseTable.getType().name()));
}
// ENABLED
- trow.addToColumnValue(new
TCell().setBoolVal(!stream.isDisabled()));
+ trow.addToColumnValue(
+ new TCell().setBoolVal(baseTableExists &&
!stream.isDisabled()));
// IS_STALE
- trow.addToColumnValue(new
TCell().setBoolVal(stream.isStale()));
+ trow.addToColumnValue(new
TCell().setBoolVal(!baseTableExists || stream.isStale()));
// STALE_REASON
- trow.addToColumnValue(new
TCell().setStringVal(stream.getStaleReason()));
+ trow.addToColumnValue(new
TCell().setStringVal(baseTableExists
+ ? stream.getStaleReason() :
BASE_TABLE_NOT_FOUND_STALE_REASON));
dataBatch.add(trow);
} finally {
stream.readUnlock();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
index 83a43edfb00..5e223251ecc 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/DropTableStreamTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.catalog;
+import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.ExceptionChecker;
@@ -25,11 +26,15 @@ import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.plans.commands.DropStreamCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.thrift.TRow;
import org.apache.doris.utframe.TestWithFeService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.List;
+
public class DropTableStreamTest extends TestWithFeService {
@Override
@@ -67,6 +72,33 @@ public class DropTableStreamTest extends TestWithFeService {
}
}
+ private void createBaseTableAndStream(String tableName, String streamName)
throws Exception {
+ createTable("create table test_stream." + tableName + " (k1 int, k2
int) "
+ + "unique key(k1) distributed by hash(k1) buckets 1 "
+ + "properties('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW', "
+ + "'binlog.need_historical_value' = 'true')");
+ createTable("create stream test_stream." + streamName + " on table
test_stream." + tableName
+ + " properties('show_initial_rows' = 'true')");
+ }
+
+ private TRow getStreamMetadataRow(String streamName) {
+ List<TRow> rows = new ArrayList<>();
+
Env.getCurrentEnv().getTableStreamManager().fillTableStreamValuesMetadataResult(rows);
+ return rows.stream()
+ .filter(row ->
streamName.equals(row.getColumnValue().get(1).getStringVal()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ }
+
+ private void assertStreamMetadataState(String streamName, String
baseTableType,
+ boolean enabled, boolean stale, String staleReason) {
+ TRow row = getStreamMetadataRow(streamName);
+ Assertions.assertEquals(baseTableType,
row.getColumnValue().get(9).getStringVal());
+ Assertions.assertEquals(enabled,
row.getColumnValue().get(10).isBoolVal());
+ Assertions.assertEquals(stale,
row.getColumnValue().get(11).isBoolVal());
+ Assertions.assertEquals(staleReason,
row.getColumnValue().get(12).getStringVal());
+ }
+
@Test
public void testNormalDropStream() throws Exception {
// test drop
@@ -104,6 +136,43 @@ public class DropTableStreamTest extends TestWithFeService
{
}
}
+ @Test
+ public void testStreamMetadataFollowsRecoverableBaseTable() throws
Exception {
+ createBaseTableAndStream("tbl_recover", "s_recover");
+ Database db =
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
+ long baseTableId = db.getTableOrMetaException("tbl_recover").getId();
+
+ assertStreamMetadataState("s_recover", "OLAP", true, false, "N/A");
+
+ dropTableWithSql("drop table test_stream.tbl_recover");
+ assertStreamMetadataState("s_recover", "N/A", false, true, "Base table
does not exist");
+
+ recoverTable("recover table test_stream.tbl_recover");
+ Assertions.assertEquals(baseTableId,
db.getTableOrMetaException("tbl_recover").getId());
+ assertStreamMetadataState("s_recover", "OLAP", true, false, "N/A");
+ }
+
+ @Test
+ public void testStreamMetadataRejectsSameNameBaseTable() throws Exception {
+ createBaseTableAndStream("tbl_force", "s_force");
+ Database db =
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
+ long oldBaseTableId = db.getTableOrMetaException("tbl_force").getId();
+ OlapTableStream stream = (OlapTableStream)
db.getTableOrMetaException("s_force");
+
+ dropTableWithSql("drop table test_stream.tbl_force force");
+ Assertions.assertNull(stream.getBaseTableNullable());
+ assertStreamMetadataState("s_force", "N/A", false, true, "Base table
does not exist");
+
+ createTable("create table test_stream.tbl_force (k1 int, k2 int) "
+ + "unique key(k1) distributed by hash(k1) buckets 1 "
+ + "properties('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW', "
+ + "'binlog.need_historical_value' = 'true')");
+
+ Assertions.assertNotEquals(oldBaseTableId,
db.getTableOrMetaException("tbl_force").getId());
+ Assertions.assertNull(stream.getBaseTableNullable());
+ assertStreamMetadataState("s_force", "N/A", false, true, "Base table
does not exist");
+ }
+
@Override
protected void runAfterAll() throws Exception {
dropDatabase("test_stream");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
