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 f075c8aabba [fix](binlog) Parse table stream properties before base
validation (#66683)
f075c8aabba is described below
commit f075c8aabba9af0a66f8d9f6bede48a482be4dcc
Author: Luwei <[email protected]>
AuthorDate: Mon Aug 17 11:16:00 2026 +0800
[fix](binlog) Parse table stream properties before base validation (#66683)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: CREATE STREAM validated the base table with the stream
object default MIN_DELTA type before parsing the requested stream
properties. This incorrectly rejected valid APPEND_ONLY and DETAIL
streams on UNIQUE or PRIMARY ROW-binlog tables that did not meet
MIN_DELTA historical-value requirements. Parse and apply the stream
properties before performing type-dependent base-table validation, while
preserving the validation for real MIN_DELTA streams.
### Release note
Fix CREATE STREAM so APPEND_ONLY and DETAIL streams are validated using
their requested type instead of the default MIN_DELTA type.
### Check List (For Author)
- Test:
- Unit Test: ./run-fe-ut.sh --run
org.apache.doris.catalog.CreateTableStreamTest,org.apache.doris.cloud.datasource.CloudInternalCatalogTableStreamTest
(passed: 9 tests)
- Regression test: ./run-regression-test.sh --run -d table_stream_p0 -s
test_create_append_only_stream_without_historical_value (passed: 1
suite)
- Behavior changed: Yes. Valid APPEND_ONLY and DETAIL streams are no
longer rejected by MIN_DELTA-only prerequisites.
- Does this need documentation: No
---
.../apache/doris/datasource/InternalCatalog.java | 4 +-
.../doris/catalog/CreateTableStreamTest.java | 46 ++++++++
.../CloudInternalCatalogTableStreamTest.java | 123 +++++++++++++++++++++
...append_only_stream_without_historical_value.out | 5 +
...end_only_stream_without_historical_value.groovy | 100 +++++++++++++++++
5 files changed, 276 insertions(+), 2 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index 59cdd7d9b9d..65a426da75f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -4096,13 +4096,13 @@ public class InternalCatalog implements
CatalogIf<Database> {
.withBaseTable(baseTable)
.build();
newStream.setComment(createStreamInfo.getComment());
- // check base table type is supported for stream
-
baseTable.checkAsTableStreamBaseTable(newStream.getStreamScanType());
try {
setTableStreamProperties(newStream, properties);
} catch (AnalysisException e) {
throw new DdlException(e.getMessage(), e);
}
+ // check base table type is supported for stream
+
baseTable.checkAsTableStreamBaseTable(newStream.getStreamScanType());
if (properties != null && !properties.isEmpty()) {
// before here, all properties should be checked
throw new DdlException("Unknown properties: " +
properties);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableStreamTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableStreamTest.java
index e775e3bc2a1..f0d665338b9 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableStreamTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableStreamTest.java
@@ -17,12 +17,14 @@
package org.apache.doris.catalog;
+import org.apache.doris.catalog.stream.BaseTableStream;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.ExceptionChecker;
import org.apache.doris.common.FeConstants;
import org.apache.doris.utframe.TestWithFeService;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -73,6 +75,50 @@ public class CreateTableStreamTest extends TestWithFeService
{
dropDatabase("test_stream");
}
+ @Test
+ public void testCreateStreamUsesRequestedTypeForBaseValidation() throws
Exception {
+ createDatabase("test_stream_type_validation");
+ String sql = "create table test_stream_type_validation.base_table\n" +
"(k1 int, k2 int)\n"
+ + "unique key(k1)\n"
+ + "distributed by hash(k1) buckets 1\n"
+ + "properties('replication_num' = '1',
'enable_unique_key_merge_on_write' = 'true', "
+ + "'binlog.enable' = 'true', 'binlog.format' = 'ROW', "
+ + "'binlog.need_historical_value' = 'false'); ";
+ createTable(sql);
+
+ ExceptionChecker.expectThrowsNoException(() ->
+ createTable("create stream
test_stream_type_validation.append_only_stream "
+ + "on table test_stream_type_validation.base_table\n"
+ + "properties('type' = 'append_only',
'show_initial_rows' = 'false'); "));
+ ExceptionChecker.expectThrowsNoException(() ->
+ createTable("create stream
test_stream_type_validation.detail_stream "
+ + "on table test_stream_type_validation.base_table\n"
+ + "properties('type' = 'detail', 'show_initial_rows' =
'false'); "));
+
+ Database db =
Env.getCurrentInternalCatalog().getDbOrDdlException("test_stream_type_validation");
+ BaseTableStream appendOnlyStream = (BaseTableStream)
db.getTableOrDdlException("append_only_stream");
+ BaseTableStream detailStream = (BaseTableStream)
db.getTableOrDdlException("detail_stream");
+ Assertions.assertEquals(BaseTableStream.StreamScanType.APPEND_ONLY,
+ appendOnlyStream.getStreamScanType());
+ Assertions.assertEquals(BaseTableStream.StreamScanType.DETAIL,
detailStream.getStreamScanType());
+
+ String minDeltaError = "MIN_DELTA table stream requires base mow table
to enable "
+ + "binlog.need_historical_value=true";
+ ExceptionChecker.expectThrowsWithMsg(DdlException.class, minDeltaError,
+ () -> createTable("create stream
test_stream_type_validation.default_stream "
+ + "on table test_stream_type_validation.base_table\n"
+ + "properties('show_initial_rows' = 'false'); "));
+ ExceptionChecker.expectThrowsWithMsg(DdlException.class, minDeltaError,
+ () -> createTable("create stream
test_stream_type_validation.min_delta_stream "
+ + "on table test_stream_type_validation.base_table\n"
+ + "properties('type' = 'min_delta',
'show_initial_rows' = 'false'); "));
+ ExceptionChecker.expectThrowsWithMsg(DdlException.class, "not
supported type: invalid_type",
+ () -> createTable("create stream
test_stream_type_validation.invalid_type_stream "
+ + "on table test_stream_type_validation.base_table\n"
+ + "properties('type' = 'invalid_type',
'show_initial_rows' = 'false'); "));
+ dropDatabase("test_stream_type_validation");
+ }
+
@Test
public void testCreateStreamAbnormalOLAP() throws Exception {
createDatabase("test_stream");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java
index acff2c9687f..fbf440ce271 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java
@@ -17,8 +17,14 @@
package org.apache.doris.cloud.datasource;
+import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.catalog.stream.BaseTableStream;
import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.catalog.stream.TableStreamBaseTableInfo;
import org.apache.doris.cloud.proto.Cloud;
@@ -26,6 +32,11 @@ import org.apache.doris.cloud.rpc.MetaServiceProxy;
import org.apache.doris.cloud.rpc.VersionHelper;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.Pair;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.nereids.trees.plans.commands.CreateStreamCommand;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateStreamInfo;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -35,12 +46,114 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
public class CloudInternalCatalogTableStreamTest {
+ @Test
+ public void testCreateUsesParsedTypeBeforeCloudPreparation() throws
Exception {
+ boolean previousEnableTableStream = Config.enable_table_stream;
+ String previousCloudUniqueId = Config.cloud_unique_id;
+ String previousMetaServiceEndpoint = Config.meta_service_endpoint;
+ Config.enable_table_stream = true;
+ Config.cloud_unique_id = "cloud_table_stream_type_validation_ut";
+ Config.meta_service_endpoint = "127.0.0.1:20121";
+ try {
+ CloudInternalCatalog catalog = Mockito.spy(new
CloudInternalCatalog());
+ Database streamDb = Mockito.mock(Database.class);
+ OlapTable baseTable = Mockito.mock(OlapTable.class);
+
Mockito.doReturn(streamDb).when(catalog).getDbNullable("test_stream");
+ Mockito.doReturn(streamDb).when(catalog).getDbNullable(10L);
+ Mockito.when(streamDb.getId()).thenReturn(10L);
+ Mockito.when(streamDb.getFullName()).thenReturn("test_stream");
+ Mockito.when(streamDb.getCatalog()).thenReturn(catalog);
+
Mockito.when(streamDb.getTable("append_only_stream")).thenReturn(Optional.empty());
+
Mockito.when(streamDb.getTable("min_delta_stream")).thenReturn(Optional.empty());
+
Mockito.when(streamDb.getTableOrDdlException("base_table")).thenReturn(baseTable);
+
Mockito.when(streamDb.getTable(20L)).thenReturn(Optional.of(baseTable));
+
Mockito.when(streamDb.createTableWithLock(Mockito.any(Table.class),
Mockito.eq(false),
+ Mockito.eq(false))).thenReturn(Pair.of(true, false));
+
+ Mockito.when(baseTable.getId()).thenReturn(20L);
+ Mockito.when(baseTable.getName()).thenReturn("base_table");
+ Mockito.when(baseTable.getDatabase()).thenReturn(streamDb);
+
Mockito.when(baseTable.getType()).thenReturn(TableIf.TableType.OLAP);
+
Mockito.when(baseTable.getBaseSchema()).thenReturn(List.<Column>of());
+ Mockito.when(baseTable.getPartitionIds()).thenReturn(List.of());
+ Mockito.when(baseTable.getBaseSchemaVersion()).thenReturn(7);
+
+ List<BaseTableStream.StreamScanType> checkedTypes = new
ArrayList<>();
+ Mockito.doAnswer(invocation -> {
+ BaseTableStream.StreamScanType type =
invocation.getArgument(0);
+ checkedTypes.add(type);
+ if (type == BaseTableStream.StreamScanType.MIN_DELTA) {
+ throw new DdlException("MIN_DELTA rejected before Cloud
preparation");
+ }
+ return null;
+ }).when(baseTable).checkAsTableStreamBaseTable(Mockito.any());
+
Mockito.doReturn(List.of()).when(catalog).captureTableStreamInitialOffsets(
+ Mockito.any(OlapTableStream.class),
Mockito.same(baseTable), Mockito.anyList());
+
+ Env env = Mockito.mock(Env.class);
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ Mockito.when(env.getNextId()).thenReturn(40L, 41L);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+
Mockito.doReturn(catalog).when(catalogMgr).getCatalog(InternalCatalog.INTERNAL_CATALOG_ID);
+
+ MetaServiceProxy proxy = Mockito.mock(MetaServiceProxy.class);
+ Cloud.IndexResponse indexResponse =
Cloud.IndexResponse.newBuilder()
+
.setStatus(Cloud.MetaServiceResponseStatus.newBuilder().setCode(Cloud.MetaServiceCode.OK))
+ .build();
+
Mockito.when(proxy.prepareIndex(Mockito.any())).thenReturn(indexResponse);
+
Mockito.when(proxy.commitIndex(Mockito.any())).thenReturn(indexResponse);
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class);
+ MockedStatic<MetaServiceProxy> mockedProxy =
Mockito.mockStatic(MetaServiceProxy.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
mockedProxy.when(MetaServiceProxy::getInstance).thenReturn(proxy);
+
+
catalog.createTableStream(createStreamCommand("append_only_stream",
"append_only"));
+ DdlException minDeltaException =
Assertions.assertThrows(DdlException.class,
+ () ->
catalog.createTableStream(createStreamCommand("min_delta_stream",
"min_delta")));
+ Assertions.assertTrue(minDeltaException.getMessage()
+ .contains("MIN_DELTA rejected before Cloud
preparation"));
+
+ Assertions.assertEquals(List.of(
+ BaseTableStream.StreamScanType.APPEND_ONLY,
+ BaseTableStream.StreamScanType.APPEND_ONLY,
+ BaseTableStream.StreamScanType.MIN_DELTA),
+ checkedTypes);
+ InOrder order = Mockito.inOrder(baseTable, proxy);
+ order.verify(baseTable).checkAsTableStreamBaseTable(
+ BaseTableStream.StreamScanType.APPEND_ONLY);
+ order.verify(proxy).prepareIndex(Mockito.any());
+ order.verify(baseTable).checkAsTableStreamBaseTable(
+ BaseTableStream.StreamScanType.APPEND_ONLY);
+ order.verify(proxy).commitIndex(Mockito.any());
+ order.verify(baseTable).checkAsTableStreamBaseTable(
+ BaseTableStream.StreamScanType.MIN_DELTA);
+ Mockito.verify(catalog).captureTableStreamInitialOffsets(
+ Mockito.any(OlapTableStream.class),
Mockito.same(baseTable), Mockito.anyList());
+ Mockito.verify(proxy).prepareIndex(Mockito.any());
+
+ ArgumentCaptor<Table> streamCaptor =
ArgumentCaptor.forClass(Table.class);
+
Mockito.verify(streamDb).createTableWithLock(streamCaptor.capture(),
Mockito.eq(false),
+ Mockito.eq(false));
+
Assertions.assertEquals(BaseTableStream.StreamScanType.APPEND_ONLY,
+ ((OlapTableStream)
streamCaptor.getValue()).getStreamScanType());
+ }
+ } finally {
+ Config.enable_table_stream = previousEnableTableStream;
+ Config.cloud_unique_id = previousCloudUniqueId;
+ Config.meta_service_endpoint = previousMetaServiceEndpoint;
+ }
+ }
+
@Test
public void testCreateBatchesOffsetsAndCommitsIndexLast() throws Exception
{
int previousBatchSize =
Config.cloud_table_stream_create_partition_batch_size;
@@ -268,6 +381,16 @@ public class CloudInternalCatalogTableStreamTest {
.build();
}
+ private static CreateStreamCommand createStreamCommand(String streamName,
String streamType) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("type", streamType);
+ properties.put("show_initial_rows", "false");
+ CreateStreamInfo createStreamInfo = new CreateStreamInfo(false, false,
+ new TableNameInfo(null, "test_stream", streamName),
+ new TableNameInfo(null, "test_stream", "base_table"),
properties, "");
+ return new CreateStreamCommand(createStreamInfo);
+ }
+
private static class CaptureCloudInternalCatalog extends
CloudInternalCatalog {
private List<Cloud.TableStreamOffsetPB> capture(OlapTableStream
stream, OlapTable baseTable,
List<Long> basePartitionIds) throws DdlException {
diff --git
a/regression-test/data/table_stream_p0/test_create_append_only_stream_without_historical_value.out
b/regression-test/data/table_stream_p0/test_create_append_only_stream_without_historical_value.out
new file mode 100644
index 00000000000..beb543d248d
--- /dev/null
+++
b/regression-test/data/table_stream_p0/test_create_append_only_stream_without_historical_value.out
@@ -0,0 +1,5 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !stream_types --
+test_append_only_without_historical_value_stream APPEND_ONLY
+test_detail_without_historical_value_stream DETAIL
+
diff --git
a/regression-test/suites/table_stream_p0/test_create_append_only_stream_without_historical_value.groovy
b/regression-test/suites/table_stream_p0/test_create_append_only_stream_without_historical_value.groovy
new file mode 100644
index 00000000000..d25d9d62ebd
--- /dev/null
+++
b/regression-test/suites/table_stream_p0/test_create_append_only_stream_without_historical_value.groovy
@@ -0,0 +1,100 @@
+// 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.
+
+suite("test_create_append_only_stream_without_historical_value",
"nonConcurrent") {
+ sql "DROP STREAM IF EXISTS
test_append_only_without_historical_value_stream"
+ sql "DROP STREAM IF EXISTS test_detail_without_historical_value_stream"
+ sql "DROP STREAM IF EXISTS test_default_without_historical_value_stream"
+ sql "DROP STREAM IF EXISTS test_min_delta_without_historical_value_stream"
+ sql "DROP STREAM IF EXISTS
test_invalid_type_without_historical_value_stream"
+ sql "DROP TABLE IF EXISTS test_stream_without_historical_value_base"
+
+ sql """
+ CREATE TABLE test_stream_without_historical_value_base (
+ id INT NOT NULL,
+ value INT
+ )
+ 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" = "false"
+ )
+ """
+
+ sql """
+ CREATE STREAM test_append_only_without_historical_value_stream
+ ON TABLE test_stream_without_historical_value_base
+ PROPERTIES (
+ "type" = "append_only",
+ "show_initial_rows" = "false"
+ )
+ """
+ sql """
+ CREATE STREAM test_detail_without_historical_value_stream
+ ON TABLE test_stream_without_historical_value_base
+ PROPERTIES (
+ "type" = "detail",
+ "show_initial_rows" = "false"
+ )
+ """
+
+ order_qt_stream_types """
+ SELECT STREAM_NAME, CONSUME_TYPE
+ FROM information_schema.table_streams
+ WHERE DB_NAME = DATABASE()
+ AND STREAM_NAME IN (
+ 'test_append_only_without_historical_value_stream',
+ 'test_detail_without_historical_value_stream'
+ )
+ ORDER BY STREAM_NAME
+ """
+
+ test {
+ sql """
+ CREATE STREAM test_default_without_historical_value_stream
+ ON TABLE test_stream_without_historical_value_base
+ PROPERTIES ("show_initial_rows" = "false")
+ """
+ exception "MIN_DELTA table stream requires base mow table to enable
binlog.need_historical_value=true"
+ }
+ test {
+ sql """
+ CREATE STREAM test_min_delta_without_historical_value_stream
+ ON TABLE test_stream_without_historical_value_base
+ PROPERTIES (
+ "type" = "min_delta",
+ "show_initial_rows" = "false"
+ )
+ """
+ exception "MIN_DELTA table stream requires base mow table to enable
binlog.need_historical_value=true"
+ }
+ test {
+ sql """
+ CREATE STREAM test_invalid_type_without_historical_value_stream
+ ON TABLE test_stream_without_historical_value_base
+ PROPERTIES (
+ "type" = "invalid_type",
+ "show_initial_rows" = "false"
+ )
+ """
+ exception "not supported type: invalid_type"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
