This is an automated email from the ASF dual-hosted git repository.
morrySnow 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 18e939bcf3a [fix](table stream) remove base table cache & use dynamic
schema (#67173)
18e939bcf3a is described below
commit 18e939bcf3abb9eb98b0403e10efedd5bb93eed6
Author: TsukiokaKogane <[email protected]>
AuthorDate: Fri Aug 28 11:04:52 2026 +0800
[fix](table stream) remove base table cache & use dynamic schema (#67173)
### What problem does this PR solve?
Issue Number: close #67095 #67109
Related PR: #65418
Problem Summary:
Table Streams currently retain both a cached `TableIf` reference and a
copied schema from the base table. These snapshots can become stale in
two scenarios. First, if the original base table is dropped and another
table is created with the same name, relation collection may lock the
replacement table by name while stream binding continues to use the
cached object of the dropped table, causing the planner lock target and
scan target to refer to different table identities. Second, when the
base table schema changes through supported ADD/DROP COLUMN operations,
especially light schema changes, the stream continues exposing its
creation-time schema through DESCRIBE, SHOW COLUMNS, information_schema,
and query column resolution. The stale schema can also remain after an
FE metadata reload.
This change removes the cached base-table object and resolves internal
base tables from the persisted catalog, database, and table IDs on each
access. It also removes the copied stream schema and derives the current
stream schema dynamically from the visible columns of the base table,
followed by the hidden stream sequence and change-type columns. As a
result, a dropped and recreated table with the same name is not treated
as the original base table, while supported base-table schema changes
are reflected by stream metadata and query planning. The added
regression case covers schema visibility before and after ADD/DROP
COLUMN and verifies stream query output after ADD COLUMN.
---
.../main/java/org/apache/doris/catalog/Table.java | 4 +-
.../doris/catalog/stream/BaseTableStream.java | 51 ++++++--
.../doris/catalog/stream/OlapTableStream.java | 50 ++++++--
.../catalog/stream/TableStreamBuildFactory.java | 25 +---
.../test_olap_table_stream_schema_sync.out | 44 +++++++
.../test_olap_table_stream_schema_sync.groovy | 129 +++++++++++++++++++++
6 files changed, 259 insertions(+), 44 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java
index 4e59eb43cf6..1fd82c3d306 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java
@@ -161,8 +161,8 @@ public abstract class Table extends MetaObject implements
Writable, TableIf, Gso
nameToColumn.put(col.getDefineName(), col);
}
} else {
- // Only view in with-clause have null base
- Preconditions.checkArgument(type == TableType.VIEW, "Table has no
columns");
+ // Only view & table stream in with-clause have null base
+ Preconditions.checkArgument(type == TableType.VIEW || type ==
TableType.STREAM, "Table has no columns");
}
this.rwLock = new MonitoredReentrantReadWriteLock(true);
this.createTime = Instant.now().getEpochSecond();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/BaseTableStream.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/BaseTableStream.java
index 951df48b20b..0111c243fea 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/BaseTableStream.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/BaseTableStream.java
@@ -93,30 +93,61 @@ public abstract class BaseTableStream extends Table {
@SerializedName("sr")
private String staleReason = "N/A";
- protected volatile TableIf baseTable;
-
// for persist
public BaseTableStream() {
super(TableType.STREAM);
}
- public BaseTableStream(long id, String streamName, List<Column>
fullSchema, TableIf baseTable) {
- super(id, streamName, TableType.STREAM, fullSchema);
+ public BaseTableStream(long id, String streamName, TableIf baseTable) {
+ super(id, streamName, TableType.STREAM, null);
this.baseTableInfo = new TableStreamBaseTableInfo(baseTable);
- this.baseTable = baseTable;
this.disabled = false;
this.stale = false;
}
- public BaseTableStream(String streamName, List<Column> fullSchema, TableIf
baseTable) {
- this(-1, streamName, fullSchema, baseTable);
+ public BaseTableStream(String streamName, TableIf baseTable) {
+ this(-1, streamName, baseTable);
}
public TableIf getBaseTableNullable() {
- if (baseTable == null) {
- baseTable = baseTableInfo.getTableNullable();
+ return baseTableInfo.getTableNullable();
+ }
+
+ // Dynamically generate the stream schema from the base table so that base
table schema
+ // changes are reflected automatically. The returned list is immutable,
allowing schema
+ // reading interfaces below to return it directly without extra copies.
+ protected abstract List<Column> generateDynamicSchema();
+
+ @Override
+ public List<Column> getFullSchema() {
+ return generateDynamicSchema();
+ }
+
+ @Override
+ public List<Column> getBaseSchema(boolean full) {
+ List<Column> schema = generateDynamicSchema();
+ if (full) {
+ return schema;
+ }
+ return
schema.stream().filter(Column::isVisible).collect(ImmutableList.toImmutableList());
+ }
+
+ @Override
+ public List<Column> getColumns() {
+ return generateDynamicSchema();
+ }
+
+ @Override
+ public Column getColumn(String colName) {
+ if (colName == null) {
+ return null;
+ }
+ for (Column column : generateDynamicSchema()) {
+ if (column.getName().equalsIgnoreCase(colName)) {
+ return column;
+ }
}
- return baseTable;
+ return null;
}
public void setProperties(Map<String, String> properties) throws
org.apache.doris.common.AnalysisException {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
index 62937e2a141..20f6bf915e8 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.Type;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
@@ -30,6 +31,7 @@ import org.apache.doris.thrift.TCell;
import org.apache.doris.thrift.TRow;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
import com.google.gson.annotations.SerializedName;
import java.io.DataInput;
@@ -58,17 +60,16 @@ public class OlapTableStream extends BaseTableStream {
super();
}
- public OlapTableStream(long id, String streamName, List<Column>
fullSchema, TableIf baseTable) {
- super(id, streamName, fullSchema, baseTable);
+ public OlapTableStream(long id, String streamName, TableIf baseTable) {
+ super(id, streamName, baseTable);
Preconditions.checkArgument(baseTable instanceof OlapTable);
this.partitionOffset = new HashMap<>();
this.partitionConsumptionTime = new HashMap<>();
this.historicalPartitionTSO = new HashMap<>();
- this.baseTable = baseTable;
}
- public OlapTableStream(String streamName, List<Column> fullSchema, TableIf
baseTable) {
- this(-1, streamName, fullSchema, baseTable);
+ public OlapTableStream(String streamName, TableIf baseTable) {
+ this(-1, streamName, baseTable);
}
@Override
@@ -85,6 +86,35 @@ public class OlapTableStream extends BaseTableStream {
return (OlapTable) baseTable;
}
+ @Override
+ protected List<Column> generateDynamicSchema() {
+ OlapTable baseTable = getBaseTableNullable();
+ if (baseTable == null) {
+ return ImmutableList.of();
+ }
+ ImmutableList.Builder<Column> builder = ImmutableList.builder();
+ // inherit base table's visible columns
+ for (Column column : baseTable.getBaseSchema()) {
+ if (column.isVisible()) {
+ builder.add(column);
+ }
+ }
+ // extra stream columns
+ Column sequenceColumn = new Column(Column.STREAM_SEQ_COL, Type.BIGINT);
+ sequenceColumn.setIsVisible(false);
+ builder.add(sequenceColumn);
+ Column changeTypeColumn = new Column(Column.STREAM_CHANGE_TYPE_COL,
Type.VARCHAR);
+ changeTypeColumn.setIsVisible(false);
+ builder.add(changeTypeColumn);
+ // Only expose stream LSN when the base table stores row LSN, e.g. dup
table with binlog.
+ if (baseTable.hasRowLsnColumn()) {
+ Column lsnColumn = new Column(Column.STREAM_LSN_COL, Type.BIGINT);
+ lsnColumn.setIsVisible(false);
+ builder.add(lsnColumn);
+ }
+ return builder.build();
+ }
+
// used for init, should inside base table read lock
@Override
public void setProperties(Map<String, String> properties) throws
AnalysisException {
@@ -99,17 +129,21 @@ public class OlapTableStream extends BaseTableStream {
private void initializeLocalOffsets() {
// set offset according to baseTable
+ OlapTable baseTable = getBaseTableNullable();
+ if (baseTable == null) {
+ return;
+ }
if (!showInitialRows) {
// set partition offset
- ((OlapTable) baseTable).getPartitions()
+ baseTable.getPartitions()
.forEach(p -> partitionOffset.put(p.getId(), p.getTso()));
} else {
- ((OlapTable) baseTable).getPartitions()
+ baseTable.getPartitions()
.stream()
.filter(p -> p.getVisibleVersion() >
Partition.PARTITION_INIT_VERSION)
.forEach(p -> {
historicalPartitionTSO.put(p.getId(),
p.getTso());
- }
+ }
);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
index 619b1e60a9f..21a72793846 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
@@ -18,17 +18,11 @@
package org.apache.doris.catalog.stream;
-import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.TableIf;
-import org.apache.doris.catalog.Type;
import org.apache.doris.common.DdlException;
import com.google.common.base.Preconditions;
-import java.util.List;
-import java.util.stream.Collectors;
-
public class TableStreamBuildFactory {
public static class BuildParams {
String tableStreamName;
@@ -55,26 +49,9 @@ public class TableStreamBuildFactory {
Preconditions.checkNotNull(params, "The factory isn't initialized.");
Preconditions.checkNotNull(params.tableStreamName, "Stream name isn't
initialized.");
Preconditions.checkNotNull(params.baseTable, "Stream base table isn't
initialized.");
- List<Column> schema = new
java.util.ArrayList<>(params.baseTable.getBaseSchema());
- // filter irrelevant invisible columns
- schema =
schema.stream().filter(Column::isVisible).collect(Collectors.toList());
- // extra columns
- Column sequenceColumn = new Column(Column.STREAM_SEQ_COL, Type.BIGINT);
- sequenceColumn.setIsVisible(false);
- schema.add(sequenceColumn);
- // Only expose stream LSN when the base table stores row LSN, e.g. dup
table with binlog.
- if (params.baseTable instanceof OlapTable
- && ((OlapTable) params.baseTable).hasRowLsnColumn()) {
- Column lsnColumn = new Column(Column.STREAM_LSN_COL, Type.BIGINT);
- lsnColumn.setIsVisible(false);
- schema.add(lsnColumn);
- }
- Column changeTypeColumn = new Column(Column.STREAM_CHANGE_TYPE_COL,
Type.VARCHAR);
- changeTypeColumn.setIsVisible(false);
- schema.add(changeTypeColumn);
switch (params.baseTable.getType()) {
case OLAP:
- return new OlapTableStream(params.tableStreamName, schema,
params.baseTable);
+ return new OlapTableStream(params.tableStreamName,
params.baseTable);
default:
throw new DdlException("unsupported stream base table type: "
+ params.baseTable.getType());
}
diff --git
a/regression-test/data/table_stream_p0/test_olap_table_stream_schema_sync.out
b/regression-test/data/table_stream_p0/test_olap_table_stream_schema_sync.out
new file mode 100644
index 00000000000..f7c0ae888ff
--- /dev/null
+++
b/regression-test/data/table_stream_p0/test_olap_table_stream_schema_sync.out
@@ -0,0 +1,44 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !desc_before --
+id bigint Yes true \N
+v1 int Yes false \N NONE
+
+-- !show_columns_before --
+id bigint YES YES \N
+v1 int YES NO \N NONE
+
+-- !info_schema_before --
+id bigint
+v1 int
+
+-- !desc_after_add --
+id bigint Yes true \N
+v1 int Yes false \N NONE
+v2 varchar(32) Yes false x NONE
+
+-- !show_columns_after_add --
+id bigint YES YES \N
+v1 int YES NO \N NONE
+v2 varchar(32) YES NO x NONE
+
+-- !info_schema_after_add --
+id bigint
+v1 int
+v2 varchar
+
+-- !select_after_add --
+1 10 x UPDATE_BEFORE
+1 11 a UPDATE_AFTER
+
+-- !desc_after_drop --
+id bigint Yes true \N
+v1 int Yes false \N NONE
+
+-- !show_columns_after_drop --
+id bigint YES YES \N
+v1 int YES NO \N NONE
+
+-- !info_schema_after_drop --
+id bigint
+v1 int
+
diff --git
a/regression-test/suites/table_stream_p0/test_olap_table_stream_schema_sync.groovy
b/regression-test/suites/table_stream_p0/test_olap_table_stream_schema_sync.groovy
new file mode 100644
index 00000000000..123639d31be
--- /dev/null
+++
b/regression-test/suites/table_stream_p0/test_olap_table_stream_schema_sync.groovy
@@ -0,0 +1,129 @@
+// 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.
+
+// The stream schema is generated dynamically from the base table, so a base
table
+// schema change (ADD/DROP COLUMN) must be reflected by the stream
automatically.
+suite("test_olap_table_stream_schema_sync", "nonConcurrent") {
+ if (isCloudMode()) {
+ return
+ }
+ sql "DROP DATABASE IF EXISTS test_olap_table_stream_schema_sync_db"
+ sql "CREATE DATABASE test_olap_table_stream_schema_sync_db"
+ sql "USE test_olap_table_stream_schema_sync_db"
+
+ def baseTable = "schema_sync_base"
+ def streamName = "schema_sync_stream"
+
+ def delta_time = 1000
+ def useTime = 0
+ def wait_for_latest_op_on_table_finish = { tableName, opTimeout ->
+ for (int t = delta_time; t <= opTimeout; t += delta_time) {
+ def alter_res = sql """SHOW ALTER TABLE COLUMN WHERE TableName =
"${tableName}" ORDER BY CreateTime DESC LIMIT 1;"""
+ alter_res = alter_res.toString()
+ if (alter_res.contains("FINISHED")) {
+ sleep(3000) // wait change table state to normal
+ logger.info(tableName + " latest alter job finished, detail: "
+ alter_res)
+ break
+ }
+ useTime = t
+ sleep(delta_time)
+ }
+ assertTrue(useTime <= opTimeout, "wait_for_latest_op_on_table_finish
timeout")
+ }
+
+ try {
+ sql "DROP STREAM IF EXISTS ${streamName}"
+ sql "DROP TABLE IF EXISTS ${baseTable}"
+
+ sql """
+ CREATE TABLE ${baseTable} (
+ id BIGINT,
+ v1 INT
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql "INSERT INTO ${baseTable} VALUES (1, 10)"
+ sql """
+ CREATE STREAM ${streamName}
+ ON TABLE ${baseTable}
+ PROPERTIES (
+ "type" = "min_delta",
+ "show_initial_rows" = "false"
+ )
+ """
+ sql "sync"
+
+ // Initial schema: base visible columns (id, v1) + stream hidden
columns.
+ qt_desc_before "DESC ${streamName}"
+ qt_show_columns_before "SHOW COLUMNS FROM ${streamName}"
+ qt_info_schema_before """
+ SELECT COLUMN_NAME, DATA_TYPE
+ FROM information_schema.columns
+ WHERE TABLE_SCHEMA = 'test_olap_table_stream_schema_sync_db'
+ AND TABLE_NAME = '${streamName}'
+ ORDER BY COLUMN_NAME
+ """
+
+ // ADD COLUMN on base table, the stream should expose the new column
automatically.
+ sql "ALTER TABLE ${baseTable} ADD COLUMN v2 VARCHAR(32) DEFAULT 'x'"
+ wait_for_latest_op_on_table_finish(baseTable, 60000)
+ sql "sync"
+ qt_desc_after_add "DESC ${streamName}"
+ qt_show_columns_after_add "SHOW COLUMNS FROM ${streamName}"
+ qt_info_schema_after_add """
+ SELECT COLUMN_NAME, DATA_TYPE
+ FROM information_schema.columns
+ WHERE TABLE_SCHEMA = 'test_olap_table_stream_schema_sync_db'
+ AND TABLE_NAME = '${streamName}'
+ ORDER BY COLUMN_NAME
+ """
+
+ sql "INSERT INTO ${baseTable} VALUES (1, 11, 'a')"
+ sql "sync"
+ sleep(1200)
+ // SELECT * must return all current visible columns of the base table
(id, v1, v2).
+ order_qt_select_after_add """
+ SELECT id, v1, v2, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM ${streamName}
+ ORDER BY id, v1, __DORIS_STREAM_CHANGE_TYPE_COL__
+ """
+
+ // DROP COLUMN on base table, the stream should drop the column
automatically.
+ sql "ALTER TABLE ${baseTable} DROP COLUMN v2"
+ wait_for_latest_op_on_table_finish(baseTable, 60000)
+ sql "sync"
+ qt_desc_after_drop "DESC ${streamName}"
+ qt_show_columns_after_drop "SHOW COLUMNS FROM ${streamName}"
+ qt_info_schema_after_drop """
+ SELECT COLUMN_NAME, DATA_TYPE
+ FROM information_schema.columns
+ WHERE TABLE_SCHEMA = 'test_olap_table_stream_schema_sync_db'
+ AND TABLE_NAME = '${streamName}'
+ ORDER BY COLUMN_NAME
+ """
+ } finally {
+ sql "DROP DATABASE IF EXISTS test_olap_table_stream_schema_sync_db"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
