From e362051dd15008f168fcb615942d7654df01a6ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=A5=E6=A0=96?= Date: Thu, 16 Jul 2026 14:03:36 +0800 Subject: [PATCH] [FLINK-40146][connect/starrocks] Support AlterTableCommentEvent in StarRocks sink --- .../pipeline-connectors/starrocks.md | 2 +- .../pipeline-connectors/starrocks.md | 2 +- .../sink/StarRocksEnrichedCatalog.java | 108 +++++++++++++++++- .../sink/StarRocksMetadataApplier.java | 30 +++-- .../starrocks/sink/MockStarRocksCatalog.java | 44 +++++++ .../sink/StarRocksMetadataApplierITCase.java | 24 ++++ .../sink/StarRocksMetadataApplierTest.java | 88 ++++++++++++++ .../sink/utils/StarRocksSinkTestBase.java | 18 +++ 8 files changed, 303 insertions(+), 13 deletions(-) diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/starrocks.md b/docs/content.zh/docs/connectors/pipeline-connectors/starrocks.md index 39e4b5d488c..b788a1a6970 100644 --- a/docs/content.zh/docs/connectors/pipeline-connectors/starrocks.md +++ b/docs/content.zh/docs/connectors/pipeline-connectors/starrocks.md @@ -247,7 +247,7 @@ pipeline: 自动设置分桶数量。对于 StarRocks 2.5 之前的版本必须设置，否则无法自动创建表。 * 对于表结构变更同步 - * 支持创建/删除/清空表，增加/删除/重命名列，修改列类型 + * 支持创建/删除/清空表，增加/删除/重命名列，修改列类型；StarRocks 3.1 及之后版本支持修改表注释 * 新增列只能添加到最后一列 * 如果使用 StarRocks 3.2 及之后版本，并且通过连接器来自动建表, 可以通过配置 `table.create.properties.fast_schema_evolution` 为 `true` 来加速 StarRocks 执行变更。 diff --git a/docs/content/docs/connectors/pipeline-connectors/starrocks.md b/docs/content/docs/connectors/pipeline-connectors/starrocks.md index b36303be33a..76ced5d75d4 100644 --- a/docs/content/docs/connectors/pipeline-connectors/starrocks.md +++ b/docs/content/docs/connectors/pipeline-connectors/starrocks.md @@ -255,7 +255,7 @@ pipeline: otherwise you must set the option. * For schema change synchronization - * supports create/drop/truncate table, add/drop/rename columns and alter column types + * supports create/drop/truncate table, add/drop/rename columns, alter column types, and alter table comments for StarRocks 3.1 or later * the new column will always be added to the last position * if your StarRocks version is 3.2 or later, and using the connector to create table automatically, you can set `table.create.properties.fast_schema_evolution` to `true` to speed up the schema change. diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java index 70f1f50f7eb..94257f65ab3 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java @@ -28,16 +28,32 @@ import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; import java.util.Optional; +import java.util.regex.Matcher; +import java.util.regex.Pattern; /** An enriched {@code StarRocksCatalog} with more schema evolution abilities. */ public class StarRocksEnrichedCatalog extends StarRocksCatalog { + private static final Logger LOG = LoggerFactory.getLogger(StarRocksEnrichedCatalog.class); + private static final Pattern VERSION_PATTERN = Pattern.compile("(\\d+)\\.(\\d+)(?:\\.(\\d+))?"); + + private final String jdbcUrl; + private final String username; + private final String password; + private Boolean supportsAlterTableComment; + public StarRocksEnrichedCatalog(String jdbcUrl, String username, String password) { super(jdbcUrl, username, password); + this.jdbcUrl = jdbcUrl; + this.username = username; + this.password = password; } - private static final Logger LOG = LoggerFactory.getLogger(StarRocksEnrichedCatalog.class); - public void truncateTable(String databaseName, String tableName) throws StarRocksCatalogException { checkTableArgument(databaseName, tableName); @@ -137,6 +153,41 @@ public void alterColumnType(String databaseName, String tableName, StarRocksColu } } + public boolean supportsAlterTableComment() throws StarRocksCatalogException { + if (supportsAlterTableComment == null) { + supportsAlterTableComment = + isAlterTableCommentSupportedVersion(fetchStarRocksVersion()); + } + return supportsAlterTableComment; + } + + public void alterTableComment(String databaseName, String tableName, String comment) + throws StarRocksCatalogException { + checkTableArgument(databaseName, tableName); + Preconditions.checkArgument(comment != null, "Table comment cannot be null."); + String alterSql = buildAlterTableCommentSql(databaseName, tableName, comment); + try { + long startTimeMillis = System.currentTimeMillis(); + executeUpdateStatement(alterSql); + LOG.info( + "Success to alter table {}.{} comment, duration: {}ms, sql: {}", + databaseName, + tableName, + System.currentTimeMillis() - startTimeMillis, + alterSql); + } catch (Exception e) { + LOG.error( + "Failed to alter table {}.{} comment, sql: {}", + databaseName, + tableName, + alterSql, + e); + throw new StarRocksCatalogException( + String.format("Failed to alter table %s.%s comment", databaseName, tableName), + e); + } + } + private String buildTruncateTableSql(String databaseName, String tableName) { return String.format("TRUNCATE TABLE `%s`.`%s`;", databaseName, tableName); } @@ -158,6 +209,13 @@ private String buildAlterColumnTypeSql( "ALTER TABLE `%s`.`%s` MODIFY COLUMN %s", databaseName, tableName, columnStmt); } + private String buildAlterTableCommentSql( + String databaseName, String tableName, String comment) { + return String.format( + "ALTER TABLE `%s`.`%s` COMMENT = \"%s\";", + databaseName, tableName, escapeSqlStringLiteral(comment)); + } + private void executeUpdateStatement(String sql) throws StarRocksCatalogException { try { Method m = @@ -171,6 +229,52 @@ private void executeUpdateStatement(String sql) throws StarRocksCatalogException } } + private String fetchStarRocksVersion() throws StarRocksCatalogException { + try { + return executeScalar("SELECT current_version()"); + } catch (Exception e) { + LOG.debug("Failed to get StarRocks version by current_version().", e); + } + + try { + // version() returns the MySQL-compatible server version, not the StarRocks version. + return executeScalar("SELECT @@version_comment"); + } catch (Exception e) { + throw new StarRocksCatalogException("Failed to get StarRocks version.", e); + } + } + + private String executeScalar(String sql) throws SQLException { + try (Connection connection = DriverManager.getConnection(jdbcUrl, username, password); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(sql)) { + if (resultSet.next()) { + return resultSet.getString(1); + } + } + throw new SQLException("No result returned for SQL: " + sql); + } + + static boolean isAlterTableCommentSupportedVersion(String version) { + if (StringUtils.isNullOrWhitespaceOnly(version)) { + return false; + } + Matcher matcher = VERSION_PATTERN.matcher(version); + if (!matcher.find()) { + return false; + } + int major = Integer.parseInt(matcher.group(1)); + int minor = Integer.parseInt(matcher.group(2)); + return major > 3 || (major == 3 && minor >= 1); + } + + static String escapeSqlStringLiteral(String str) { + return str.replace("\\", "\\\\") + .replace("\"", "\\\"") + .replace("\n", "\\n") + .replace("\r", "\\r"); + } + private void checkTableArgument(String databaseName, String tableName) { Preconditions.checkArgument( !StringUtils.isNullOrWhitespaceOnly(databaseName), diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java index 7f020844106..d89108b452b 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java @@ -19,6 +19,7 @@ import org.apache.flink.cdc.common.event.AddColumnEvent; import org.apache.flink.cdc.common.event.AlterColumnTypeEvent; +import org.apache.flink.cdc.common.event.AlterTableCommentEvent; import org.apache.flink.cdc.common.event.CreateTableEvent; import org.apache.flink.cdc.common.event.DropColumnEvent; import org.apache.flink.cdc.common.event.DropTableEvent; @@ -29,6 +30,7 @@ import org.apache.flink.cdc.common.event.TruncateTableEvent; import org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor; import org.apache.flink.cdc.common.exceptions.SchemaEvolveException; +import org.apache.flink.cdc.common.exceptions.UnsupportedSchemaChangeEventException; import org.apache.flink.cdc.common.schema.Column; import org.apache.flink.cdc.common.sink.MetadataApplier; import org.apache.flink.cdc.common.types.DataType; @@ -93,7 +95,8 @@ public Set getSupportedSchemaEvolutionTypes() { SchemaChangeEventType.RENAME_COLUMN, SchemaChangeEventType.ALTER_COLUMN_TYPE, SchemaChangeEventType.DROP_TABLE, - SchemaChangeEventType.TRUNCATE_TABLE); + SchemaChangeEventType.TRUNCATE_TABLE, + SchemaChangeEventType.ALTER_TABLE_COMMENT); } @Override @@ -113,14 +116,7 @@ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) this::applyDropTable, this::applyRenameColumn, this::applyTruncateTable, - alterTableCommentEvent -> { - // TODO Currently, table comments cannot be modified. - // See - // https://docs.starrocks.io/docs/sql-reference/sql-statements/table_bucket_part_index/ALTER_TABLE/#alter-table-comment-from-v31 - LOG.warn( - "AlterTableCommentEvent is not supported by StarRocks connector yet. Event: {}", - alterTableCommentEvent); - }); + this::applyAlterTableComment); } private void applyCreateTable(CreateTableEvent createTableEvent) throws SchemaEvolveException { @@ -334,6 +330,22 @@ private void applyAlterColumnType(AlterColumnTypeEvent event) throws SchemaEvolv } } + private void applyAlterTableComment(AlterTableCommentEvent event) throws SchemaEvolveException { + try { + if (!catalog.supportsAlterTableComment()) { + throw new UnsupportedSchemaChangeEventException( + event, "Alter table comment requires StarRocks 3.1 or later."); + } + TableId tableId = event.tableId(); + catalog.alterTableComment( + tableId.getSchemaName(), tableId.getTableName(), event.getComment()); + } catch (UnsupportedSchemaChangeEventException e) { + throw e; + } catch (Exception e) { + throw new SchemaEvolveException(event, "fail to apply alter table comment event", e); + } + } + private void applyTruncateTable(TruncateTableEvent truncateTableEvent) { try { catalog.truncateTable( diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/MockStarRocksCatalog.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/MockStarRocksCatalog.java index c12be85a735..8edd61a16b1 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/MockStarRocksCatalog.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/MockStarRocksCatalog.java @@ -34,9 +34,12 @@ public class MockStarRocksCatalog extends StarRocksEnrichedCatalog { /** database name -> table name -> table. */ private final Map> tables; + private boolean supportsAlterTableComment; + public MockStarRocksCatalog() { super("jdbc:mysql://127.0.0.1:9030", "root", ""); this.tables = new HashMap<>(); + this.supportsAlterTableComment = true; } @Override @@ -96,6 +99,33 @@ public void createTable(StarRocksTable table, boolean ignoreIfExists) } } + public void setSupportsAlterTableComment(boolean supportsAlterTableComment) { + this.supportsAlterTableComment = supportsAlterTableComment; + } + + @Override + public boolean supportsAlterTableComment() { + return supportsAlterTableComment; + } + + @Override + public void alterTableComment(String databaseName, String tableName, String comment) + throws StarRocksCatalogException { + Map dbTables = tables.get(databaseName); + if (dbTables == null) { + throw new StarRocksCatalogException( + String.format("database %s does not exist", databaseName)); + } + + StarRocksTable oldTable = dbTables.get(tableName); + if (oldTable == null) { + throw new StarRocksCatalogException( + String.format("table %s.%s does not exist", databaseName, tableName)); + } + + dbTables.put(tableName, copyTableWithComment(oldTable, comment)); + } + @Override public void alterAddColumns( String databaseName, @@ -195,4 +225,18 @@ public void alterDropColumns( .build(); dbTables.put(tableName, newTable); } + + private StarRocksTable copyTableWithComment(StarRocksTable oldTable, String comment) { + return new StarRocksTable.Builder() + .setDatabaseName(oldTable.getDatabaseName()) + .setTableName(oldTable.getTableName()) + .setTableType(oldTable.getTableType()) + .setColumns(oldTable.getColumns()) + .setTableKeys(oldTable.getTableKeys().orElse(null)) + .setDistributionKeys(oldTable.getDistributionKeys().orElse(null)) + .setNumBuckets(oldTable.getNumBuckets().orElse(null)) + .setComment(comment) + .setTableProperties(oldTable.getProperties()) + .build(); + } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java index d9164d60cbd..06550866735 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java @@ -22,6 +22,7 @@ import org.apache.flink.cdc.common.data.binary.BinaryStringData; import org.apache.flink.cdc.common.event.AddColumnEvent; import org.apache.flink.cdc.common.event.AlterColumnTypeEvent; +import org.apache.flink.cdc.common.event.AlterTableCommentEvent; import org.apache.flink.cdc.common.event.CreateTableEvent; import org.apache.flink.cdc.common.event.DataChangeEvent; import org.apache.flink.cdc.common.event.DropColumnEvent; @@ -380,6 +381,29 @@ void testStarRocksAlterColumnType() throws Exception { assertEqualsInOrder(expected, actual); } + @Test + void testStarRocksAlterTableComment() throws Exception { + TableId tableId = + TableId.tableId( + StarRocksContainer.STARROCKS_DATABASE_NAME, + StarRocksContainer.STARROCKS_TABLE_NAME); + + Schema schema = + Schema.newBuilder() + .column(new PhysicalColumn("id", DataTypes.INT().notNull(), null)) + .column(new PhysicalColumn("name", DataTypes.STRING(), null)) + .primaryKey("id") + .comment("old table comment") + .build(); + + runJobWithEvents( + Arrays.asList( + new CreateTableEvent(tableId, schema), + new AlterTableCommentEvent(tableId, "new table comment"))); + + Assertions.assertThat(inspectTableComment(tableId)).isEqualTo("new table comment"); + } + @Test void testStarRocksNarrowingAlterColumnType() throws Exception { Assertions.assertThatThrownBy( diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierTest.java index 2da34d6cf4f..b19f5a41207 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierTest.java @@ -19,9 +19,12 @@ import org.apache.flink.cdc.common.configuration.Configuration; import org.apache.flink.cdc.common.event.AddColumnEvent; +import org.apache.flink.cdc.common.event.AlterTableCommentEvent; import org.apache.flink.cdc.common.event.CreateTableEvent; import org.apache.flink.cdc.common.event.DropColumnEvent; +import org.apache.flink.cdc.common.event.SchemaChangeEventType; import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.exceptions.UnsupportedSchemaChangeEventException; import org.apache.flink.cdc.common.schema.Column; import org.apache.flink.cdc.common.schema.Schema; import org.apache.flink.cdc.common.types.BooleanType; @@ -69,6 +72,12 @@ public void setup() { new StarRocksMetadataApplier(catalog, tableCreateConfig, schemaChangeConfig); } + @Test + void testSupportedSchemaEvolutionTypes() { + Assertions.assertThat(metadataApplier.getSupportedSchemaEvolutionTypes()) + .contains(SchemaChangeEventType.ALTER_TABLE_COMMENT); + } + @Test void testCreateTable() throws Exception { TableId tableId = TableId.parse("test.tbl1"); @@ -122,6 +131,85 @@ void testCreateTable() throws Exception { Assertions.assertThat(actualTable).isEqualTo(expectTable); } + @Test + void testAlterTableComment() throws Exception { + TableId tableId = TableId.parse("test.tbl_comment"); + Schema schema = + Schema.newBuilder() + .physicalColumn("col1", new IntType()) + .primaryKey("col1") + .comment("old table comment") + .build(); + metadataApplier.applySchemaChange(new CreateTableEvent(tableId, schema)); + + metadataApplier.applySchemaChange(new AlterTableCommentEvent(tableId, "new table comment")); + + StarRocksTable actualTable = + catalog.getTable(tableId.getSchemaName(), tableId.getTableName()).orElse(null); + Assertions.assertThat(actualTable).isNotNull(); + Assertions.assertThat(actualTable.getComment()).hasValue("new table comment"); + } + + @Test + void testAlterTableCommentWithUnsupportedVersion() throws Exception { + TableId tableId = TableId.parse("test.tbl_comment_unsupported_version"); + Schema schema = + Schema.newBuilder() + .physicalColumn("col1", new IntType()) + .primaryKey("col1") + .build(); + metadataApplier.applySchemaChange(new CreateTableEvent(tableId, schema)); + + catalog.setSupportsAlterTableComment(false); + + Assertions.assertThatThrownBy( + () -> + metadataApplier.applySchemaChange( + new AlterTableCommentEvent(tableId, "new table comment"))) + .isExactlyInstanceOf(UnsupportedSchemaChangeEventException.class) + .extracting(e -> ((UnsupportedSchemaChangeEventException) e).getExceptionMessage()) + .asString() + .contains("StarRocks 3.1 or later"); + } + + @Test + void testAlterTableCommentSupportedVersion() { + Assertions.assertThat(StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion("3.1.0")) + .isTrue(); + Assertions.assertThat( + StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion("3.5.10")) + .isTrue(); + Assertions.assertThat( + StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion( + "StarRocks-3.5.10")) + .isTrue(); + Assertions.assertThat( + StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion( + "StarRocks version 3.1.0")) + .isTrue(); + Assertions.assertThat(StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion("4.0.0")) + .isTrue(); + Assertions.assertThat(StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion("3.0.9")) + .isFalse(); + Assertions.assertThat( + StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion( + "StarRocks version 2.1.2")) + .isFalse(); + Assertions.assertThat(StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion("2.5.0")) + .isFalse(); + Assertions.assertThat( + StarRocksEnrichedCatalog.isAlterTableCommentSupportedVersion("unknown")) + .isFalse(); + } + + @Test + void testEscapeSqlStringLiteral() { + Assertions.assertThat( + StarRocksEnrichedCatalog.escapeSqlStringLiteral( + "comment with \"quote\", backslash \\, newline\nand carriage\r")) + .isEqualTo("comment with \\\"quote\\\", backslash \\\\, newline\\nand carriage\\r"); + } + @Test void testAddColumn() throws Exception { TableId tableId = TableId.parse("test.tbl2"); diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksSinkTestBase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksSinkTestBase.java index 471690d09fc..8d68a26d500 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksSinkTestBase.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksSinkTestBase.java @@ -40,6 +40,7 @@ import org.testcontainers.lifecycle.Startables; import java.sql.Connection; +import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.time.Duration; @@ -207,6 +208,23 @@ public List inspectTableSchema(TableId tableId) throws SQLException { return results; } + public String inspectTableComment(TableId tableId) throws SQLException { + String sql = + "SELECT TABLE_COMMENT FROM information_schema.TABLES " + + "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?"; + try (Connection connection = STARROCKS_CONTAINER.createConnection(""); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, tableId.getSchemaName()); + statement.setString(2, tableId.getTableName()); + try (ResultSet resultSet = statement.executeQuery()) { + if (resultSet.next()) { + return resultSet.getString(1); + } + } + } + return null; + } + public List fetchTableContent(TableId tableId, int columnCount) throws SQLException { List results = new ArrayList<>(); ResultSet rs =