[feat](stream) Support ALTER STREAM ... SET/MODIFY COMMENT

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -35,6 +35,7 @@
 import org.apache.doris.catalog.Tablet;
 import org.apache.doris.catalog.View;
 import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.catalog.stream.BaseTableStream;
 import org.apache.doris.cloud.alter.CloudSchemaChangeHandler;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
@@ -470,6 +471,27 @@ private void processModifyTableComment(Database db, OlapTable tbl, AlterOp alter
         }
     }
 
+    /**
+     * Modify the comment of a table stream, e.g. ALTER STREAM s1 SET COMMENT 'new comment'.
+     * The comment of a stream is kept in the Table metadata only, so it shares the same
+     * edit log entry and the same replay path with ALTER TABLE ... MODIFY COMMENT.
+     */
+    public void processAlterStreamComment(long dbId, BaseTableStream stream, String comment) throws DdlException {
+        if (!Config.enable_table_stream) {
+            throw new DdlException("Table Stream is experimental."
+                    + " Please set enable_table_stream=true to enable it.");
+        }
+        stream.writeLockOrDdlException();
+        try {
+            stream.setComment(comment);
+            // log
+            ModifyCommentOperationLog op = ModifyCommentOperationLog.forTable(dbId, stream.getId(), comment);
+            Env.getCurrentEnv().getEditLog().logModifyComment(op);
+        } finally {
+            stream.writeUnlock();
+        }
+    }
+
     private void processModifyColumnComment(Database db, OlapTable tbl, List<AlterOp> alterOps)
             throws DdlException {
         tbl.writeLockOrDdlException();

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -666,6 +666,7 @@
 import org.apache.doris.nereids.trees.plans.commands.AlterSqlBlockRuleCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterStoragePolicyCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterStorageVaultCommand;
+import org.apache.doris.nereids.trees.plans.commands.AlterStreamCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterSystemCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterSystemRenameComputeGroupCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterTableCommand;
@@ -4011,8 +4012,8 @@ public LogicalPlan visitCreateStream(CreateStreamContext ctx) {
         if (ctx.REPLACE() != null && ctx.EXISTS() != null) {
             throw new AnalysisException("[OR REPLACE] and [IF NOT EXISTS] cannot used at the same time");
         }
-        String comment = ctx.STRING_LITERAL() == null ? "" : LogicalPlanBuilderAssistant.escapeBackSlash(
-                ctx.STRING_LITERAL().getText().substring(1, ctx.STRING_LITERAL().getText().length() - 1));
+        String comment = ctx.STRING_LITERAL() == null ? ""
+                : SqlLiteralUtils.parseStringLiteral(ctx.STRING_LITERAL().getText());
         Map<String, String> properties = ctx.properties != null
                 // NOTICE: we should not generate immutable map here, because it will be modified when analyzing.
                 ? Maps.newHashMap(visitPropertyClause(ctx.properties))
@@ -7268,6 +7269,13 @@ public LogicalPlan visitAlterCatalogComment(AlterCatalogCommentContext ctx) {
         return new AlterCatalogCommentCommand(catalogName, comment);
     }
 
+    @Override
+    public LogicalPlan visitAlterStreamComment(DorisParser.AlterStreamCommentContext ctx) {
+        TableNameInfo streamName = new TableNameInfo(visitMultipartIdentifier(ctx.name));
+        String comment = SqlLiteralUtils.parseStringLiteral(ctx.comment.getText());
+        return new AlterStreamCommand(streamName, AlterStreamCommand.AlterType.SET_COMMENT, comment);
+    }
+
     @Override
     public LogicalPlan visitAlterDatabaseRename(AlterDatabaseRenameContext ctx) {
         String dbName = Optional.ofNullable(ctx.name)

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java

@@ -487,6 +487,7 @@ public enum PlanType {
     DROP_INDEX_NORMALIZER_COMMAND,
     SHOW_INDEX_NORMALIZER_COMMAND,
     CREATE_STREAM_COMMAND,
+    ALTER_STREAM_COMMAND,
     DROP_STREAM_COMMAND,
     SHOW_CREATE_STREAM_COMMAND,
     SHOW_STREAMS,

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterStreamCommand.java

@@ -0,0 +1,112 @@
+// 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.
+
+package org.apache.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.catalog.stream.BaseTableStream;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.InternalDatabaseUtil;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.StmtExecutor;
+
+import com.google.common.base.Strings;
+
+/**
+ * AlterStreamCommand, e.g. ALTER STREAM s1 SET COMMENT 'new comment'
+ */
+public class AlterStreamCommand extends AlterCommand {
+    /**
+     * The kind of alteration applied to the stream.
+     */
+    public enum AlterType {
+        SET_COMMENT
+    }
+
+    private final TableNameInfo streamName;
+    private final AlterType alterType;
+    private final String comment;
+
+    public AlterStreamCommand(TableNameInfo streamName, AlterType alterType, String comment) {
+        super(PlanType.ALTER_STREAM_COMMAND);
+        this.streamName = streamName;
+        this.alterType = alterType;
+        this.comment = comment;
+    }
+
+    @Override
+    public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate(ctx);
+
+        DatabaseIf db = Env.getCurrentEnv().getCatalogMgr()
+                .getCatalogOrDdlException(streamName.getCtl())
+                .getDbOrDdlException(streamName.getDb());
+        TableIf table = db.getTableOrDdlException(streamName.getTbl());
+        if (!(table instanceof BaseTableStream)) {
+            ErrorReport.reportDdlException(ErrorCode.ERR_WRONG_OBJECT, streamName.getDb(), streamName.getTbl(),
+                    "STREAM", "Use 'ALTER TABLE " + streamName.getTbl() + "'");
+        }
+
+        switch (alterType) {
+            case SET_COMMENT:
+                Env.getCurrentEnv().getAlterInstance()
+                        .processAlterStreamComment(db.getId(), (BaseTableStream) table, comment);
+                break;
+            default:
+                throw new UserException("Unsupported alter stream operation: " + alterType);
+        }
+    }
+
+    private void validate(ConnectContext ctx) throws UserException {
+        if (Strings.isNullOrEmpty(streamName.getDb())) {
+            streamName.setDb(ctx.getDatabase());
+        }
+        streamName.analyze(ctx.getNameSpaceContext());
+        InternalDatabaseUtil.checkDatabase(streamName.getDb(), ctx);
+        if (!Env.getCurrentEnv().getAccessManager()
+                .checkTblPriv(ctx, streamName.getCtl(), streamName.getDb(), streamName.getTbl(),
+                        PrivPredicate.ALTER)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "ALTER STREAM",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(), streamName.getDb() + ": " + streamName.getTbl());
+        }
+    }
+
+    @Override
+    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+        return visitor.visitAlterStreamCommand(this, context);
+    }
+
+    public TableNameInfo getStreamName() {
+        return streamName;
+    }
+
+    public AlterType getAlterType() {
+        return alterType;
+    }
+
+    public String getComment() {
+        return comment;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/CommandVisitor.java

@@ -53,6 +53,7 @@
 import org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterSqlBlockRuleCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterStoragePolicyCommand;
+import org.apache.doris.nereids.trees.plans.commands.AlterStreamCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterTableCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterTableStatsCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterUserCommand;
@@ -989,6 +990,10 @@ default R visitDropStreamCommand(DropStreamCommand command, C context) {
         return visitCommand(command, context);
     }
 
+    default R visitAlterStreamCommand(AlterStreamCommand command, C context) {
+        return visitCommand(command, context);
+    }
+
     default R visitDropRoleCommand(DropRoleCommand dropRoleCommand, C context) {
         return visitCommand(dropRoleCommand, context);
     }

fe/fe-core/src/test/java/org/apache/doris/catalog/AlterTableStreamCommentTest.java

@@ -0,0 +1,120 @@
+// 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.
+
+package org.apache.doris.catalog;
+
+import org.apache.doris.catalog.stream.BaseTableStream;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ExceptionChecker;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.qe.SqlModeHelper;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class AlterTableStreamCommentTest extends TestWithFeService {
+
+    @Override
+    protected int backendNum() {
+        return 3;
+    }
+
+    @Override
+    protected void runBeforeAll() throws Exception {
+        FeConstants.runningUnitTest = true;
+        Config.allow_replica_on_same_host = true;
+        Config.enable_table_stream = true;
+    }
+
+    @Test
+    public void testAlterStreamComment() throws Exception {
+        createDatabase("test_alter_stream_comment");
+        createTable("create table test_alter_stream_comment.base_tbl (k1 int, v1 int)\n"
+                + "duplicate key(k1)\n"
+                + "distributed by hash(k1) buckets 1\n"
+                + "properties('replication_num' = '1', 'binlog.enable' = 'true', 'binlog.format' = 'ROW');");
+        createTable("create stream test_alter_stream_comment.s1 on table test_alter_stream_comment.base_tbl\n"
+                + "comment 'initial comment'\n"
+                + "properties('type' = 'append_only');");
+
+        Database db = Env.getCurrentInternalCatalog().getDbOrDdlException("test_alter_stream_comment");
+        BaseTableStream stream = (BaseTableStream) db.getTableOrDdlException("s1");
+        Assertions.assertEquals("initial comment", stream.getComment());
+
+        executeSql("alter stream test_alter_stream_comment.s1 set comment 'updated comment'");
+        Assertions.assertEquals("updated comment", stream.getComment());
+
+        // MODIFY COMMENT is accepted as well, to stay consistent with ALTER TABLE
+        executeSql("alter stream test_alter_stream_comment.s1 modify comment 'modified comment'");
+        Assertions.assertEquals("modified comment", stream.getComment());
+
+        // an empty comment clears the comment
+        executeSql("alter stream test_alter_stream_comment.s1 set comment ''");
+        Assertions.assertEquals("", stream.getComment());
+
+        // altering a normal table through ALTER STREAM is rejected
+        ExceptionChecker.expectThrowsWithMsg(IllegalStateException.class, "is not STREAM",
+                () -> executeSql("alter stream test_alter_stream_comment.base_tbl set comment 'not a stream'"));
+
+        // altering an unknown stream is rejected
+        ExceptionChecker.expectThrowsWithMsg(IllegalStateException.class, "Unknown table",
+                () -> executeSql("alter stream test_alter_stream_comment.not_exist set comment 'no such stream'"));
+
+        dropDatabase("test_alter_stream_comment");
+    }
+
+    @Test
+    public void testAlterStreamCommentStringLiteral() throws Exception {
+        createDatabase("test_alter_stream_comment_literal");
+        createTable("create table test_alter_stream_comment_literal.base_tbl (k1 int, v1 int)\n"
+                + "duplicate key(k1)\n"
+                + "distributed by hash(k1) buckets 1\n"
+                + "properties('replication_num' = '1', 'binlog.enable' = 'true', 'binlog.format' = 'ROW');");
+        // CREATE STREAM and ALTER STREAM must decode the string literal in the same way:
+        // a doubled quote is one quote and, unless NO_BACKSLASH_ESCAPES is set, backslash escapes are decoded
+        createTable("create stream test_alter_stream_comment_literal.s1"
+                + " on table test_alter_stream_comment_literal.base_tbl\n"
+                + "comment 'a''b\\nc'\n"
+                + "properties('type' = 'append_only');");
+
+        Database db = Env.getCurrentInternalCatalog().getDbOrDdlException("test_alter_stream_comment_literal");
+        BaseTableStream stream = (BaseTableStream) db.getTableOrDdlException("s1");
+        Assertions.assertEquals("a'b\nc", stream.getComment());
+
+        executeSql("alter stream test_alter_stream_comment_literal.s1 set comment 'x''y\\tz'");
+        Assertions.assertEquals("x'y\tz", stream.getComment());
+
+        // a double quoted literal doubles the double quote instead
+        executeSql("alter stream test_alter_stream_comment_literal.s1 set comment \"p\"\"q\"");
+        Assertions.assertEquals("p\"q", stream.getComment());
+
+        // under NO_BACKSLASH_ESCAPES a backslash is an ordinary character. Both the lexer and
+        // SqlLiteralUtils read the sql mode from the thread local ConnectContext, so bind it first.
+        connectContext.setThreadLocalInfo();
+        long originalSqlMode = connectContext.getSessionVariable().getSqlMode();
+        try {
+            connectContext.getSessionVariable().setSqlMode(SqlModeHelper.MODE_NO_BACKSLASH_ESCAPES);
+            executeSql("alter stream test_alter_stream_comment_literal.s1 set comment 'a\\nb'");
+            Assertions.assertEquals("a\\nb", stream.getComment());
+        } finally {
+            connectContext.getSessionVariable().setSqlMode(originalSqlMode);
+        }
+
+        dropDatabase("test_alter_stream_comment_literal");
+    }
+}

fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -468,6 +468,8 @@ alterStatement
         properties=propertyClause?                                                          #alterWorkloadPolicy
     | ALTER SQL_BLOCK_RULE name=identifier properties=propertyClause?                       #alterSqlBlockRule
     | ALTER CATALOG name=identifier MODIFY COMMENT comment=STRING_LITERAL                   #alterCatalogComment
+    | ALTER STREAM name=multipartIdentifier
+        (SET | MODIFY) COMMENT comment=STRING_LITERAL                                       #alterStreamComment
     | ALTER DATABASE name=identifier RENAME newName=identifier                              #alterDatabaseRename
     | ALTER STORAGE POLICY name=identifierOrText
         properties=propertyClause                                                           #alterStoragePolicy

regression-test/data/table_stream_p0/test_table_stream_alter_comment.out

@@ -0,0 +1,19 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !comment_after_create --
+initial comment
+
+-- !comment_after_set --
+updated comment
+
+-- !comment_after_modify --
+modified comment
+
+-- !comment_doubled_quote --
+a'b	3
+
+-- !comment_after_clear --
+
+
+-- !comment_final --
+final comment
+