From 46dbb5c06134a3cb4829ff3ff7b501ff672a89ab Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:59:28 +0800 Subject: [PATCH 1/2] Enforce SQL read-only checks before analysis --- .../it/db/it/IoTDBSetSystemStatusTableIT.java | 191 ++++++++++++ .../db/queryengine/plan/Coordinator.java | 15 +- .../plan/ReadOnlyStatementValidator.java | 261 +++++++++++++++++ .../plan/relational/sql/ast/InsertRows.java | 45 +++ .../statement/crud/InsertBaseStatement.java | 22 +- .../statement/crud/InsertRowsStatement.java | 6 +- .../executor/RegionWriteExecutorTest.java | 39 +++ .../plan/ReadOnlyStatementValidatorTest.java | 277 ++++++++++++++++++ .../sql/AstMemoryEstimatorTest.java | 168 +++++++++++ 9 files changed, 1020 insertions(+), 4 deletions(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java index 4dac5e9feb4ec..98e77f39af5d5 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java @@ -25,6 +25,8 @@ import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; import org.apache.iotdb.itbase.env.BaseEnv; import org.apache.iotdb.itbase.exception.InconsistentDataException; +import org.apache.iotdb.itbase.runtime.ClusterTestConnection; +import org.apache.iotdb.rpc.TSStatusCode; import org.awaitility.Awaitility; import org.junit.AfterClass; @@ -36,6 +38,7 @@ import java.sql.Connection; import java.sql.ResultSet; +import java.sql.SQLException; import java.sql.Statement; import java.util.concurrent.TimeUnit; @@ -52,6 +55,194 @@ public static void tearDown() throws Exception { EnvFactory.getEnv().cleanClusterEnvironment(); } + @Test + public void testReadOnlySqlAndDiskRecovery() throws Exception { + try (Connection treeConnection = EnvFactory.getEnv().getConnection(); + Connection tableConnection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement tree = treeConnection.createStatement(); + Statement table = tableConnection.createStatement(); + // PREPARE and EXECUTE must use the same session, including with multi-node read checks. + Statement prepared = + (tableConnection instanceof ClusterTestConnection + ? ((ClusterTestConnection) tableConnection) + .writeConnection.getUnderlyingConnection() + : tableConnection) + .createStatement()) { + tree.execute("create database root.readonly_sql"); + tree.execute("create database root.readonly_drop"); + tree.execute("create timeseries root.readonly_sql.d.s with datatype=INT32,encoding=PLAIN"); + tree.execute("create timeseries root.readonly_drop.d.s with datatype=INT32,encoding=PLAIN"); + tree.execute("insert into root.readonly_sql.d(time,s) values(1,1)"); + tree.execute("insert into root.readonly_drop.d(time,s) values(1,1)"); + table.execute("create database readonly_sql"); + table.execute("create database readonly_drop"); + table.execute("create table readonly_drop.t(d STRING TAG,s INT32 FIELD)"); + table.execute("insert into readonly_drop.t(time,d,s) values(1,'d1',1)"); + table.execute("use readonly_sql"); + table.execute("create table t(d STRING TAG,a STRING ATTRIBUTE,s INT32 FIELD)"); + table.execute("create table prepared_t(d STRING TAG,a STRING ATTRIBUTE,s INT32 FIELD)"); + table.execute("insert into t(time,d,a,s) values(1,'d1','original',1)"); + tree.execute("flush"); + // Resolve both a read and a write before entering read-only mode. + prepared.execute("prepare readonly_read from select s from t"); + prepared.execute( + "prepare readonly_write from insert into prepared_t(time,d,a,s) " + + "values(1,'p1','prepared',11),(2,'p2','prepared',12)"); + prepared.execute("execute readonly_write"); + table.execute("update prepared_t set a='changed' where d='p1'"); + prepared.execute("execute readonly_write"); + prepared.execute( + "execute immediate 'insert into prepared_t(time,d,a,s) " + + "values(3,''p3'',''immediate'',13)'"); + try (ResultSet result = table.executeQuery("select d,a,s from prepared_t order by time")) { + for (int i = 1; i <= 3; i++) { + Assert.assertTrue(result.next()); + Assert.assertEquals("p" + i, result.getString(1)); + Assert.assertEquals(i == 3 ? "immediate" : "prepared", result.getString(2)); + Assert.assertEquals(10 + i, result.getInt(3)); + } + Assert.assertFalse(result.next()); + } + try { + table.execute("set system to readonly on cluster"); + awaitDataNodeStatus(table, "ReadOnly"); + + for (String sql : + new String[] { + "show databases", + "show timeseries root.readonly_sql.**", + "show devices root.readonly_sql.**", + "count timeseries root.readonly_sql.**", + "show pipes", + "show ttl on root.readonly_sql.**", + "list user" + }) { + assertQueryAvailable(tree, sql); + } + for (String sql : + new String[] { + "show tables", + "describe t", + "show create table t", + "show databases", + "show pipes", + "show devices from t", + "count devices from t", + "select * from information_schema.tables", + "execute immediate 'select s from t'", + "list user" + }) { + assertQueryAvailable(table, sql); + } + assertQueryAvailable(prepared, "execute readonly_read"); + assertQueryAvailable(prepared, "explain analyze execute readonly_read"); + try (ResultSet result = tree.executeQuery("select s from root.readonly_sql.d")) { + Assert.assertTrue(result.next()); + Assert.assertEquals(1, result.getInt(2)); + Assert.assertFalse(result.next()); + } + try (ResultSet result = table.executeQuery("select a,s from t where d='d1'")) { + Assert.assertTrue(result.next()); + Assert.assertEquals("original", result.getString(1)); + Assert.assertEquals(1, result.getInt(2)); + Assert.assertFalse(result.next()); + } + + for (String sql : + new String[] { + "insert into root.readonly_sql.d(time,s) values(2,2)", + "delete from root.readonly_sql.d.s where time<=1", + "select s into root.readonly_dest.d(s) from root.readonly_sql.d", + "create database root.readonly_new", + "delete timeseries root.readonly_sql.d.s", + "flush" + }) { + assertReadOnlyRejected(tree, sql); + } + for (String sql : + new String[] { + "insert into t(time,d,s) values(2,'d1',2)", + "insert into t select * from t", + "delete from t where time<=1", + "update t set a='changed' where d='d1'", + "create database readonly_new", + "create table newtable(d STRING TAG,s INT32 FIELD)", + "drop table t", + "alter database readonly_sql set properties ttl=1, schema_region_group_num=2", + "create role readonly_role", + "execute immediate 'delete from t where time<=1'", + "execute immediate 'insert into prepared_t(time,d,s) values(4,''p4'',14)'", + "flush" + }) { + assertReadOnlyRejected(table, sql); + } + assertReadOnlyRejected(prepared, "execute readonly_write"); + assertReadOnlyRejected(prepared, "explain analyze execute readonly_write"); + + tree.execute("set ttl to root.readonly_sql.** 1"); + tree.execute("delete database root.readonly_drop"); + table.execute("alter database readonly_sql set properties ttl=1"); + table.execute("alter table t set properties ttl=1"); + table.execute("drop database readonly_drop"); + try (ResultSet result = tree.executeQuery("show databases root.readonly_drop")) { + Assert.assertFalse(result.next()); + } + try (ResultSet result = table.executeQuery("show databases")) { + while (result.next()) { + Assert.assertNotEquals("readonly_drop", result.getString(1)); + } + } + // The recovery exceptions do not switch the system back to Running. + awaitDataNodeStatus(table, "ReadOnly"); + } finally { + table.execute("set system to running on cluster"); + awaitDataNodeStatus(table, "Running"); + } + // Normal writes resume after leaving read-only mode. + tree.execute("unset ttl to root.readonly_sql.**"); + table.execute("alter database readonly_sql set properties ttl=default"); + table.execute("alter table t set properties ttl=default"); + tree.execute("insert into root.readonly_sql.d(time,s) values(now(),2)"); + table.execute("insert into t(time,d,s) values(now(),'d1',2)"); + prepared.execute("deallocate prepare readonly_write"); + prepared.execute("deallocate prepare readonly_read"); + } + } + + private static void awaitDataNodeStatus(Statement statement, String expectedStatus) { + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .untilAsserted( + () -> { + try (ResultSet result = statement.executeQuery("show datanodes")) { + int count = 0; + while (result.next()) { + Assert.assertEquals(expectedStatus, result.getString("Status")); + count++; + } + Assert.assertEquals(EnvFactory.getEnv().getDataNodeWrapperList().size(), count); + } + }); + } + + private static void assertQueryAvailable(Statement statement, String sql) throws SQLException { + try (ResultSet result = statement.executeQuery(sql)) { + Assert.assertTrue(sql, result.getMetaData().getColumnCount() > 0); + while (result.next()) { + // Consume the full result to surface errors from distributed query execution. + } + } + } + + private static void assertReadOnlyRejected(Statement statement, String sql) { + try { + statement.execute(sql); + Assert.fail("Expected SYSTEM_READ_ONLY for " + sql); + } catch (SQLException e) { + Assert.assertEquals(sql, TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), e.getErrorCode()); + } + } + @Test public void setSystemStatus() { try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java index 79e6e38d4fe1f..5c9116841cdb2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java @@ -103,6 +103,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainOutputFormat; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExtendRegion; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Flush; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertRows; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.KillQuery; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadConfiguration; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadModel; @@ -430,6 +431,7 @@ private IQueryExecution createQueryExecutionForTreeModel( ISchemaFetcher schemaFetcher, long timeOut, long startTime) { + ReadOnlyStatementValidator.check(statement); queryContext.setTimeOut(timeOut); queryContext.setStartTime(startTime); if (statement instanceof ShowPipeMemoryStatement) { @@ -676,6 +678,7 @@ private IQueryExecution createQueryExecutionForTableModel( Metadata metadata, long timeOut, long startTime) { + ReadOnlyStatementValidator.check(statement); queryContext.setTimeOut(timeOut); queryContext.setStartTime(startTime); TableModelPlanner tableModelPlanner = @@ -778,6 +781,7 @@ private IQueryExecution createQueryExecutionForTableModel( || statement instanceof RemoveRegion || statement instanceof Prepare || statement instanceof Deallocate) { + ReadOnlyStatementValidator.check(statement, sqlParser, clientSession); return new ConfigExecution( queryContext, null, @@ -815,6 +819,10 @@ private IQueryExecution createQueryExecutionForTableModel( org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement resolvedSql = preparedInfo.getSql(); + if (resolvedSql instanceof InsertRows) { + resolvedSql = ((InsertRows) resolvedSql).copyForPreparedExecution(); + } + innerStatement = resolvedSql; parameterLookup = ParameterExtractor.bindParameters(resolvedSql, executeStatement.getParameters()); parameters = new ArrayList<>(executeStatement.getParameters()); @@ -838,6 +846,7 @@ private IQueryExecution createQueryExecutionForTableModel( org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement resolvedSql = sqlParser.createStatement(sql, clientSession.getZoneId(), clientSession); + innerStatement = resolvedSql; if (!literalParameters.isEmpty()) { parameterLookup = ParameterExtractor.bindParameters(resolvedSql, literalParameters); @@ -856,8 +865,10 @@ private IQueryExecution createQueryExecutionForTableModel( } } - if (statement instanceof WrappedInsertStatement) { - ((WrappedInsertStatement) statement).setContext(queryContext); + // Check the resolved SQL that will be analyzed, including EXECUTE and EXECUTE IMMEDIATE. + ReadOnlyStatementValidator.check(statementToUse, sqlParser, clientSession); + if (innerStatement instanceof WrappedInsertStatement) { + ((WrappedInsertStatement) innerStatement).setContext(queryContext); } // Create QueryExecution with TableModelPlanner diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java new file mode 100644 index 0000000000000..9bcb5027b9662 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java @@ -0,0 +1,261 @@ +/* + * 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.iotdb.db.queryengine.plan; + +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.exception.IoTDBRuntimeException; +import org.apache.iotdb.commons.i18n.CommonMessages; +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Query; +import org.apache.iotdb.db.protocol.session.IClientSession; +import org.apache.iotdb.db.queryengine.plan.analyze.QueryType; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterDB; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CountDB; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CountDevice; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CountStatement; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Deallocate; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Execute; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExecuteImmediate; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Explain; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainAnalyze; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FetchDevice; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.KillQuery; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Prepare; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Property; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RelationalAuthorStatement; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetProperties; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetSqlDialect; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowAIDevices; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowAINodes; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowAvailableUrls; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCluster; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowClusterId; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowConfigNodes; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCreateDatabase; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCreatePipe; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCreateTopic; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentDatabase; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentSqlDialect; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentTimestamp; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentUser; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDB; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDataNodes; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDevice; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowFunctions; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowIndex; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowLoadedModels; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowModels; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipePlugins; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipes; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowRegions; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowStatement; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowSubscriptions; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowTables; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowTopics; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowVariables; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowVersion; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Use; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.WrappedStatement; +import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; +import org.apache.iotdb.db.queryengine.plan.statement.IConfigStatement; +import org.apache.iotdb.db.queryengine.plan.statement.Statement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; +import org.apache.iotdb.db.queryengine.plan.statement.internal.DeviceSchemaFetchStatement; +import org.apache.iotdb.db.queryengine.plan.statement.internal.SeriesSchemaFetchStatement; +import org.apache.iotdb.db.queryengine.plan.statement.metadata.DeleteDatabaseStatement; +import org.apache.iotdb.db.queryengine.plan.statement.metadata.SetTTLStatement; +import org.apache.iotdb.db.queryengine.plan.statement.metadata.UnSetTTLStatement; +import org.apache.iotdb.db.queryengine.plan.statement.pipe.PipeEnrichedStatement; +import org.apache.iotdb.db.queryengine.plan.statement.sys.ExplainAnalyzeStatement; +import org.apache.iotdb.db.queryengine.plan.statement.sys.ExplainStatement; +import org.apache.iotdb.db.queryengine.plan.statement.sys.KillQueryStatement; +import org.apache.iotdb.db.queryengine.plan.statement.sys.SetSqlDialectStatement; +import org.apache.iotdb.db.queryengine.plan.statement.sys.SetSystemStatusStatement; +import org.apache.iotdb.rpc.TSStatusCode; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** Checks SQL before analysis, which can itself create schema or allocate partitions. */ +public final class ReadOnlyStatementValidator { + + private ReadOnlyStatementValidator() {} + + public static void check(final Statement statement) { + if (CommonDescriptor.getInstance().getConfig().isReadOnly() && !isAllowed(statement)) { + reject(); + } + } + + public static void check( + final org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement statement, + final SqlParser parser, + final IClientSession session) { + if (CommonDescriptor.getInstance().getConfig().isReadOnly() + && !isAllowed(statement, parser, session)) { + reject(); + } + } + + static boolean isAllowed(final Statement statement) { + if (statement instanceof PipeEnrichedStatement) { + return isAllowed(((PipeEnrichedStatement) statement).getInnerStatement()); + } + if (statement instanceof QueryStatement) { + return !((QueryStatement) statement).isSelectInto(); + } + if (statement instanceof ExplainStatement) { + return isAllowed(((ExplainStatement) statement).getQueryStatement()); + } + if (statement instanceof ExplainAnalyzeStatement) { + return isAllowed(((ExplainAnalyzeStatement) statement).getQueryStatement()); + } + // Deleting a whole database and setting TTL are explicit disk-recovery exceptions. + // They still use the normal metadata persistence and permission checks. + if (statement instanceof DeleteDatabaseStatement + || (statement instanceof SetTTLStatement && !(statement instanceof UnSetTTLStatement)) + || statement instanceof SetSystemStatusStatement + || statement instanceof SetSqlDialectStatement + || statement instanceof KillQueryStatement) { + return true; + } + if (statement instanceof IConfigStatement) { + return ((IConfigStatement) statement).getQueryType() == QueryType.READ; + } + // SELECT resolves schema through these internal reads on a schema-cache miss. + return statement instanceof SeriesSchemaFetchStatement + || statement instanceof DeviceSchemaFetchStatement + || statement + instanceof org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowStatement + || statement.isQuery(); + } + + static boolean isAllowed( + final org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement statement, + final SqlParser parser, + final IClientSession session) { + return isAllowed(statement, parser, session, new HashSet<>()); + } + + private static boolean isAllowed( + final org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement statement, + final SqlParser parser, + final IClientSession session, + final Set preparedStatements) { + if (statement instanceof WrappedStatement) { + return isAllowed(((WrappedStatement) statement).getInnerTreeStatement()); + } + if (statement instanceof Explain) { + return isAllowed(((Explain) statement).getStatement(), parser, session, preparedStatements); + } + if (statement instanceof ExplainAnalyze) { + return isAllowed( + ((ExplainAnalyze) statement).getStatement(), parser, session, preparedStatements); + } + if (statement instanceof Execute) { + final String name = ((Execute) statement).getStatementName().getValue(); + if (!preparedStatements.add(name)) { + return false; + } + final var prepared = session.getPreparedStatement(name); + // Let the normal execution path report an unknown prepared statement. + return prepared == null || isAllowed(prepared.getSql(), parser, session, preparedStatements); + } + if (statement instanceof ExecuteImmediate) { + return isAllowed( + parser.createStatement( + ((ExecuteImmediate) statement).getSqlString(), session.getZoneId(), session), + parser, + session, + preparedStatements); + } + if (statement instanceof RelationalAuthorStatement) { + return ((RelationalAuthorStatement) statement).getQueryType() == QueryType.READ; + } + if (statement instanceof AlterDB) { + return onlySetsTTL(((AlterDB) statement).getProperties()); + } + if (statement instanceof SetProperties) { + return ((SetProperties) statement).getType() == SetProperties.Type.TABLE + && onlySetsTTL(((SetProperties) statement).getProperties()); + } + // Other SHOW forms wrap tree statements above or are rewritten to information_schema reads. + // Session-local operations are safe and must remain usable while the server is read-only. + // Table reads retain the attribute-sync registration in RegionWriteExecutor for cache + // consistency. + return statement instanceof Query + || statement instanceof ShowStatement + || statement instanceof CountStatement + || statement instanceof ShowDB + || statement instanceof ShowCreateDatabase + || statement instanceof CountDB + || statement instanceof ShowAIDevices + || statement instanceof ShowAINodes + || statement instanceof ShowAvailableUrls + || statement instanceof ShowCluster + || statement instanceof ShowClusterId + || statement instanceof ShowConfigNodes + || statement instanceof ShowCurrentDatabase + || statement instanceof ShowCurrentSqlDialect + || statement instanceof ShowCurrentTimestamp + || statement instanceof ShowCurrentUser + || statement instanceof ShowDataNodes + || statement instanceof ShowFunctions + || statement instanceof ShowIndex + || statement instanceof ShowLoadedModels + || statement instanceof ShowModels + || statement instanceof ShowPipePlugins + || statement instanceof ShowRegions + || statement instanceof ShowVariables + || statement instanceof ShowVersion + || statement instanceof ShowTables + || statement instanceof DescribeTable + || statement instanceof ShowDevice + || statement instanceof CountDevice + || statement instanceof FetchDevice + || statement instanceof ShowPipes + || statement instanceof ShowCreatePipe + || statement instanceof ShowTopics + || statement instanceof ShowCreateTopic + || statement instanceof ShowSubscriptions + || statement instanceof DropDB + || statement instanceof Use + || statement instanceof KillQuery + || statement instanceof SetSqlDialect + || statement instanceof Prepare + || statement instanceof Deallocate; + } + + private static boolean onlySetsTTL(final List properties) { + return !properties.isEmpty() + && properties.stream() + .allMatch( + property -> + "ttl".equalsIgnoreCase(property.getName().getValue()) + && !property.isSetToDefault()); + } + + private static void reject() { + throw new IoTDBRuntimeException( + CommonMessages.SYSTEM_READ_ONLY, TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), true); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertRows.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertRows.java index 6b5804ba4ebb8..f40c8ca0350be 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertRows.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertRows.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.queryengine.plan.relational.sql.ast; import org.apache.iotdb.calc.exception.QueryProcessException; +import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.IAstVisitor; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; @@ -32,6 +33,7 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Constants; +import org.apache.tsfile.utils.RamUsageEstimator; import java.util.ArrayList; import java.util.Arrays; @@ -44,6 +46,9 @@ public class InsertRows extends WrappedInsertStatement { + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(InsertRows.class); + // Only InsertRows constructed by Pipe will be set to true private boolean allowCreateTable = false; @@ -61,6 +66,46 @@ public InsertRowsStatement getInnerTreeStatement() { return ((InsertRowsStatement) super.getInnerTreeStatement()); } + @Override + public long ramBytesUsed() { + return INSTANCE_SIZE + getInnerTreeStatement().ramBytesUsed(); + } + + /** Copy the parsed VALUES payload so analysis cannot mutate the prepared statement. */ + public InsertRows copyForPreparedExecution() { + InsertRowsStatement original = getInnerTreeStatement(); + List rows = new ArrayList<>(original.getInsertRowStatementList().size()); + for (InsertRowStatement row : original.getInsertRowStatementList()) { + InsertRowStatement copy = new InsertRowStatement(); + copy.setDevicePath(new PartialPath(row.getDevicePath().getNodes().clone())); + copy.setTime(row.getTime()); + copy.setMeasurements(copyArray(row.getMeasurements())); + copy.setMeasurementSchemas(copyArray(row.getMeasurementSchemas())); + copy.setDataTypes(copyArray(row.getDataTypes())); + copy.setColumnCategories(copyArray(row.getColumnCategories())); + copy.setValues(copyArray(row.getValues())); + copy.setNeedInferType(row.isNeedInferType()); + copy.setAligned(row.isAligned()); + copy.setWriteToTable(row.isWriteToTable()); + if (row.getFailedMeasurementInfoMap() != null) { + copy.setFailedMeasurementIndex2Info(new HashMap<>(row.getFailedMeasurementInfoMap())); + } + row.getDatabaseName().ifPresent(copy::setDatabaseName); + rows.add(copy); + } + InsertRowsStatement copy = new InsertRowsStatement(); + copy.setInsertRowStatementList(rows); + copy.setUniformTargets(original.isUniformTargets()); + copy.setWriteToTable(original.isWriteToTable()); + InsertRows result = new InsertRows(copy, null); + result.setAllowCreateTable(allowCreateTable); + return result; + } + + private static T[] copyArray(T[] array) { + return array == null ? null : array.clone(); + } + @Override public void updateAfterSchemaValidation(MPPQueryContext context) throws QueryProcessException { getInnerTreeStatement().updateAfterSchemaValidation(context); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java index e305a13fdbd27..385605e6cad96 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java @@ -53,6 +53,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.IdentityHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -579,7 +580,10 @@ public void removeAttributeColumns() { protected abstract void subRemoveAttributeColumns(List columnsToKeep); - public static class FailedMeasurementInfo { + public static class FailedMeasurementInfo implements Accountable { + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(FailedMeasurementInfo.class); + protected String measurement; protected TSDataType dataType; protected Object value; @@ -604,6 +608,21 @@ public TSDataType getDataType() { public Object getValue() { return value; } + + @Override + public long ramBytesUsed() { + long size = + INSTANCE_SIZE + + RamUsageEstimator.sizeOf(measurement) + + RamUsageEstimator.sizeOfObject(value); + // Exceptions can retain additional copies of the failed input in their messages. + Set visited = Collections.newSetFromMap(new IdentityHashMap<>()); + for (Throwable error = cause; error != null && visited.add(error); error = error.getCause()) { + size += + RamUsageEstimator.sizeOfObject(error) + RamUsageEstimator.sizeOf(error.getMessage()); + } + return size; + } } // endregion @@ -1015,6 +1034,7 @@ public long ramBytesUsed() { + RamUsageEstimator.sizeOf(measurements) + RamUsageEstimator.shallowSizeOf(dataTypes) + RamUsageEstimator.shallowSizeOf(columnCategories) + + RamUsageEstimator.sizeOfMap(failedMeasurementIndex2Info) // We assume that the integers are all cached by JVM + shallowSizeOfList(tagColumnIndices) + shallowSizeOfList(attrColumnIndices) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java index 728831a7d443b..e89d32231c536 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java @@ -233,8 +233,12 @@ protected long calculateBytesUsed() { return INSTANCE_SIZE + (Objects.nonNull(insertRowStatementList) ? UpdateDetailContainer.LIST_SIZE + + RamUsageEstimator.alignObjectSize( + RamUsageEstimator.NUM_BYTES_ARRAY_HEADER + + (long) RamUsageEstimator.NUM_BYTES_OBJECT_REF + * insertRowStatementList.size()) + insertRowStatementList.stream() - .mapToLong(InsertRowStatement::calculateBytesUsed) + .mapToLong(InsertRowStatement::ramBytesUsed) .reduce(0L, Long::sum) : 0); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java index 22521458c17b6..2d8f0fc4c5f08 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java @@ -19,7 +19,11 @@ package org.apache.iotdb.db.queryengine.execution.executor; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.consensus.SchemaRegionId; @@ -43,6 +47,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode; import org.apache.iotdb.db.queryengine.plan.statement.metadata.AlterTimeSeriesStatement.AlterType; import org.apache.iotdb.db.schemaengine.SchemaEngine; import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion; @@ -70,6 +75,40 @@ public class RegionWriteExecutorTest { + @Test + public void testReadOnlyAllowsAttributeSyncRegistration() throws ConsensusException { + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final NodeStatus originalStatus = config.getNodeStatus(); + final String originalReason = config.getStatusReason(); + final IConsensus consensus = Mockito.mock(IConsensus.class); + final DataNodeRegionManager regionManager = Mockito.mock(DataNodeRegionManager.class); + final SchemaRegionId regionId = new SchemaRegionId(1); + Mockito.when(regionManager.getRegionLock(regionId)).thenReturn(new ReentrantReadWriteLock()); + final RegionWriteExecutor executor = + new RegionWriteExecutor( + consensus, + consensus, + regionManager, + Mockito.mock(SchemaEngine.class), + Mockito.mock(ClusterTemplateManager.class), + Mockito.mock(TriggerFireVisitor.class)); + final TableNodeLocationAddNode registration = + new TableNodeLocationAddNode(new PlanNodeId("register"), new TDataNodeLocation()); + Mockito.when(consensus.write(regionId, registration)) + .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + try { + for (String reason : new String[] {NodeStatus.MANUAL, NodeStatus.DISK_FULL}) { + config.setNodeStatusWithReason(NodeStatus.ReadOnly, reason); + final RegionExecutionResult result = executor.execute(regionId, registration); + assertTrue(result.isAccepted()); + assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), result.getStatus().getCode()); + } + Mockito.verify(consensus, Mockito.times(2)).write(regionId, registration); + } finally { + config.setNodeStatusWithReason(originalStatus, originalReason); + } + } + @Test public void testAlterTimeSeriesRegionAvailability() throws Exception { // Alter validation fetches the measurement before consensus; a missing region must be diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java new file mode 100644 index 0000000000000..07a17f95b1c65 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java @@ -0,0 +1,277 @@ +/* + * 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.iotdb.db.queryengine.plan; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.exception.IoTDBRuntimeException; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.path.PathPatternTree; +import org.apache.iotdb.db.protocol.session.ClientSession; +import org.apache.iotdb.db.protocol.session.PreparedStatementInfo; +import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; +import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; +import org.apache.iotdb.db.queryengine.plan.statement.Statement; +import org.apache.iotdb.db.queryengine.plan.statement.internal.DeviceSchemaFetchStatement; +import org.apache.iotdb.db.queryengine.plan.statement.internal.InternalBatchActivateTemplateStatement; +import org.apache.iotdb.db.queryengine.plan.statement.internal.InternalCreateMultiTimeSeriesStatement; +import org.apache.iotdb.db.queryengine.plan.statement.internal.SeriesSchemaFetchStatement; +import org.apache.iotdb.db.queryengine.plan.statement.pipe.PipeEnrichedStatement; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.time.ZoneOffset; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +public class ReadOnlyStatementValidatorTest { + private final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + private final SqlParser parser = new SqlParser(); + private final ClientSession session = new ClientSession(null); + private NodeStatus originalStatus; + private String originalReason; + + @Before + public void setUp() { + originalStatus = config.getNodeStatus(); + originalReason = config.getStatusReason(); + config.setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.MANUAL); + session.setZoneId(ZoneOffset.UTC); + } + + @After + public void tearDown() { + config.setNodeStatusWithReason(originalStatus, originalReason); + } + + @Test + public void testTreeReads() { + for (String sql : + new String[] { + "show databases", + "show timeseries root.**", + "show devices root.**", + "count timeseries root.**", + "count devices root.**", + "show ttl on root.**", + "show pipes", + "show pipeplugins", + "show functions", + "show triggers", + "show schema templates", + "show cluster", + "show datanodes", + "show confignodes", + "show variables", + "show version", + "show current_user", + "show queries", + "show disk_usage from root.**", + "list user", + "list role", + "select s from root.db.d", + "explain select s from root.db.d", + "explain analyze select s from root.db.d" + }) { + ReadOnlyStatementValidator.check(tree(sql)); + } + } + + @Test + public void testInternalSchemaReadsAndWrites() throws Exception { + PathPatternTree patternTree = new PathPatternTree(); + patternTree.appendPathPattern(new PartialPath("root.db.**")); + patternTree.constructTree(); + ReadOnlyStatementValidator.check( + new SeriesSchemaFetchStatement( + patternTree, Collections.emptyMap(), false, false, true, false, false)); + ReadOnlyStatementValidator.check( + new DeviceSchemaFetchStatement(patternTree, patternTree, false)); + assertRejected( + () -> + ReadOnlyStatementValidator.check( + new InternalCreateMultiTimeSeriesStatement(Collections.emptyMap()))); + assertRejected( + () -> + ReadOnlyStatementValidator.check( + new InternalBatchActivateTemplateStatement(Collections.emptyMap()))); + } + + @Test + public void testTableReads() { + for (String sql : + new String[] { + "show databases", + "count databases", + "show tables from db", + "show tables details from db", + "describe db.t", + "describe db.t details", + "show create database db", + "show create table db.t", + "show pipes", + "show pipeplugins", + "show functions", + "show cluster", + "show datanodes", + "show confignodes", + "show variables", + "show version", + "show current_user", + "show current_database", + "show current_sql_dialect", + "show devices from db.t", + "count devices from db.t", + "list user", + "list role", + "select * from db.t", + "select * from information_schema.tables", + "explain select * from db.t", + "explain analyze select * from db.t" + }) { + checkTable(sql); + } + } + + @Test + public void testDiskRecoveryAndSessionOperations() { + ReadOnlyStatementValidator.check(tree("delete database root.db")); + ReadOnlyStatementValidator.check(tree("set ttl to root.db.** 1")); + ReadOnlyStatementValidator.check(tree("set system to running on local")); + ReadOnlyStatementValidator.check(tree("set sql_dialect=table")); + checkTable("drop database db"); + checkTable("alter database db set properties ttl=1"); + checkTable("alter table db.t set properties ttl=1"); + checkTable("set system to running on local"); + checkTable("use db"); + checkTable("kill query '20250206_093300_00001_1'"); + checkTable("set sql_dialect=tree"); + checkTable("prepare q from select * from db.t"); + checkTable("deallocate prepare q"); + } + + @Test + public void testTreeWritesAreRejected() { + for (String sql : + new String[] { + "insert into root.db.d(time,s) values(1,1)", + "delete from root.db.d.s where time<=1", + "select s into root.dest.d(s) from root.db.d", + "explain analyze select s into root.dest.d(s) from root.db.d", + "create database root.newdb", + "create timeseries root.db.d.s with datatype=INT32,encoding=PLAIN", + "delete timeseries root.db.d.s", + "create user readonlyuser 'IoTDB!Test2024'", + "flush", + "start repair data", + "unset ttl to root.db.**", + "set configuration 'query_timeout_threshold'='10000'" + }) { + assertRejected(sql, () -> ReadOnlyStatementValidator.check(tree(sql))); + } + assertRejected( + () -> + ReadOnlyStatementValidator.check( + new PipeEnrichedStatement(tree("create database root.newdb")))); + } + + @Test + public void testTableWritesAreRejected() { + for (String sql : + new String[] { + "insert into db.t select * from db.source", + "insert into db.t(time,s) select * from db.source", + "delete from db.t where time<=1", + "update db.t set a='new' where tag='d'", + "create database newdb", + "create table db.newtable(tag STRING TAG,s INT32 FIELD)", + "drop table db.t", + "alter table db.t add column s2 INT32 FIELD", + "alter database db set properties ttl=1, schema_region_group_num=2", + "alter table db.t set properties ttl=1, comment='changed'", + "alter table db.t set properties ttl=default", + "create role readonlyrole", + "flush", + "set configuration query_timeout_threshold='10000'", + "explain analyze execute immediate 'insert into db.t select * from db.source'" + }) { + assertRejected(sql, () -> checkTable(sql)); + } + } + + @Test + public void testPreparedExecutionChecksTheInnerStatement() { + session.addPreparedStatement( + "q", new PreparedStatementInfo("q", table("select * from db.t"), 0)); + session.addPreparedStatement( + "w", new PreparedStatementInfo("w", table("insert into db.t select * from db.source"), 0)); + checkTable("execute q"); + checkTable("explain analyze execute q"); + checkTable("execute immediate 'select * from db.t'"); + assertRejected(() -> checkTable("execute w")); + assertRejected(() -> checkTable("explain analyze execute w")); + assertRejected( + () -> checkTable("execute immediate 'insert into db.t select * from db.source'")); + session.addPreparedStatement( + "cycle", new PreparedStatementInfo("cycle", table("execute cycle"), 0)); + assertRejected(() -> checkTable("execute cycle")); + } + + @Test + public void testRunningModeKeepsExistingBehavior() { + config.setNodeStatus(NodeStatus.Running); + ReadOnlyStatementValidator.check(tree("create database root.db")); + checkTable("create database db"); + } + + private Statement tree(String sql) { + return StatementGenerator.createStatement(sql, ZoneOffset.UTC); + } + + private org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement table(String sql) { + return parser.createStatement(sql, ZoneOffset.UTC, session); + } + + private void checkTable(String sql) { + ReadOnlyStatementValidator.check(table(sql), parser, session); + } + + private void assertRejected(Runnable action) { + assertRejected("", action); + } + + private void assertRejected(String sql, Runnable action) { + try { + action.run(); + fail("Expected SYSTEM_READ_ONLY for " + sql); + } catch (IoTDBRuntimeException e) { + assertEquals( + sql + ": " + e.getMessage(), + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), + e.getErrorCode()); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/AstMemoryEstimatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/AstMemoryEstimatorTest.java index 0738b8cd27acc..c107593c84feb 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/AstMemoryEstimatorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/AstMemoryEstimatorTest.java @@ -19,17 +19,37 @@ package org.apache.iotdb.db.queryengine.plan.relational.sql; +import org.apache.iotdb.commons.memory.IMemoryBlock; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.protocol.session.ClientSession; import org.apache.iotdb.db.protocol.session.IClientSession; import org.apache.iotdb.db.protocol.session.InternalClientSession; +import org.apache.iotdb.db.protocol.session.PreparedStatementInfo; +import org.apache.iotdb.db.queryengine.plan.Coordinator; +import org.apache.iotdb.db.queryengine.plan.execution.config.session.PreparedStatementHelper; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertRows; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Prepare; import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.schemaengine.table.ITableCache; +import org.apache.tsfile.enums.TSDataType; import org.junit.Before; import org.junit.Test; import java.time.ZoneId; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; public class AstMemoryEstimatorTest { @@ -756,6 +776,154 @@ public void testPrepare() { assertTrue("Memory size should be positive", memorySize > 0); } + @Test + public void testPreparedInsertValuesMemoryAccounting() { + TsTable table = new TsTable("insert_memory"); + table.addColumnSchema(new TimeColumnSchema("time", TSDataType.TIMESTAMP)); + table.addColumnSchema(new TagColumnSchema("device", TSDataType.STRING)); + table.addColumnSchema(new FieldColumnSchema("value", TSDataType.STRING)); + ITableCache cache = DataNodeTableCache.getInstance(); + cache.preUpdateTable("testdb", table, null); + cache.commitUpdateTable("testdb", "insert_memory", null, null); + ClientSession session = new ClientSession(null); + IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + int originalDataNodeId = config.getDataNodeId(); + config.setDataNodeId(0); + try { + IMemoryBlock memory = Coordinator.getCoordinatorMemoryBlock(); + long initialUsage = memory.getUsedMemoryInBytes(); + Prepare prepare = + (Prepare) + parseSQL( + "PREPARE insert_values FROM INSERT INTO insert_memory(time,device,value) " + + "VALUES(1,'d1','small'),(2,'d2',null)"); + InsertRows insert = (InsertRows) prepare.getSql(); + long size = insert.ramBytesUsed(); + long rowSizes = + insert.getInnerTreeStatement().getInsertRowStatementList().stream() + .mapToLong(row -> row.ramBytesUsed()) + .sum(); + assertTrue(size > rowSizes); + assertTrue(prepare.ramBytesUsed() > size); + assertEquals(size, insert.ramBytesUsed()); + + Statement singleRow = + parseSQL("INSERT INTO insert_memory(time,device,value) VALUES(1,'d1','small')"); + Statement largeValue = + parseSQL( + "INSERT INTO insert_memory(time,device,value) VALUES(1,'d1','" + + "x".repeat(4096) + + "')"); + assertTrue(size > singleRow.ramBytesUsed()); + assertTrue(largeValue.ramBytesUsed() > singleRow.ramBytesUsed() + 4000); + + PreparedStatementInfo info = + PreparedStatementHelper.register(session, "insert_values", insert); + assertSame(insert, info.getSql()); + assertEquals(size, info.getMemorySizeInBytes()); + assertEquals(initialUsage + size, memory.getUsedMemoryInBytes()); + PreparedStatementHelper.unregister(session, "insert_values"); + assertNull(session.getPreparedStatement("insert_values")); + assertEquals(initialUsage, memory.getUsedMemoryInBytes()); + } finally { + if (session.getPreparedStatement("insert_values") != null) { + PreparedStatementHelper.unregister(session, "insert_values"); + } + cache.invalid("testdb"); + config.setDataNodeId(originalDataNodeId); + } + } + + @Test + public void testPreparedInsertValuesParseFailureMemoryAccounting() { + TsTable table = new TsTable("insert_failure_memory"); + table.addColumnSchema(new TimeColumnSchema("time", TSDataType.TIMESTAMP)); + table.addColumnSchema(new TagColumnSchema("device", TSDataType.STRING)); + table.addColumnSchema(new FieldColumnSchema("valid", TSDataType.INT32)); + table.addColumnSchema(new FieldColumnSchema("invalid", TSDataType.INT32)); + ITableCache cache = DataNodeTableCache.getInstance(); + cache.preUpdateTable("testdb", table, null); + cache.commitUpdateTable("testdb", "insert_failure_memory", null, null); + ClientSession session = new ClientSession(null); + IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + boolean originalPartialInsert = config.isEnablePartialInsert(); + int originalDataNodeId = config.getDataNodeId(); + config.setEnablePartialInsert(true); + config.setDataNodeId(0); + try { + InsertRows small = + (InsertRows) parseSQL("INSERT INTO insert_failure_memory VALUES(1,'d1',11,'bad')"); + InsertRows large = + (InsertRows) + parseSQL( + "INSERT INTO insert_failure_memory VALUES(1,'d1',11,'" + "x".repeat(4096) + "')"); + InsertRowStatement row = large.getInnerTreeStatement().getInsertRowStatementList().get(0); + assertEquals(1, row.getFailedMeasurementNumber()); + assertNull(row.getValues()[2]); + assertTrue(large.ramBytesUsed() > small.ramBytesUsed() + 4000); + + IMemoryBlock memory = Coordinator.getCoordinatorMemoryBlock(); + long initialUsage = memory.getUsedMemoryInBytes(); + PreparedStatementInfo info = + PreparedStatementHelper.register(session, "insert_failure", large); + assertEquals(large.ramBytesUsed(), info.getMemorySizeInBytes()); + assertEquals(initialUsage + large.ramBytesUsed(), memory.getUsedMemoryInBytes()); + PreparedStatementHelper.unregister(session, "insert_failure"); + assertEquals(initialUsage, memory.getUsedMemoryInBytes()); + } finally { + if (session.getPreparedStatement("insert_failure") != null) { + PreparedStatementHelper.unregister(session, "insert_failure"); + } + cache.invalid("testdb"); + config.setEnablePartialInsert(originalPartialInsert); + config.setDataNodeId(originalDataNodeId); + } + } + + @Test + public void testPreparedInsertValuesPreservesParseFailures() { + TsTable table = new TsTable("insert_partial"); + table.addColumnSchema(new TimeColumnSchema("time", TSDataType.TIMESTAMP)); + table.addColumnSchema(new TagColumnSchema("device", TSDataType.STRING)); + table.addColumnSchema(new FieldColumnSchema("valid", TSDataType.INT32)); + table.addColumnSchema(new FieldColumnSchema("invalid", TSDataType.INT32)); + ITableCache cache = DataNodeTableCache.getInstance(); + cache.preUpdateTable("testdb", table, null); + cache.commitUpdateTable("testdb", "insert_partial", null, null); + IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + boolean originalPartialInsert = config.isEnablePartialInsert(); + config.setEnablePartialInsert(true); + try { + Prepare prepare = + (Prepare) + parseSQL( + "PREPARE insert_partial FROM INSERT INTO insert_partial " + + "VALUES(1,'d1',11,'invalid integer')"); + InsertRows original = (InsertRows) prepare.getSql(); + InsertRowStatement originalRow = + original.getInnerTreeStatement().getInsertRowStatementList().get(0); + assertEquals(1, originalRow.getFailedMeasurementNumber()); + for (int i = 0; i < 2; i++) { + InsertRows execution = original.copyForPreparedExecution(); + InsertRowStatement executionRow = + execution.getInnerTreeStatement().getInsertRowStatementList().get(0); + execution.getInnerTreeStatement().semanticCheck(); + assertEquals(originalRow.getFailedMeasurements(), executionRow.getFailedMeasurements()); + assertEquals(originalRow.getFailedMessages(), executionRow.getFailedMessages()); + assertEquals(11, executionRow.getValues()[1]); + assertNotSame( + originalRow.getFailedMeasurementInfoMap(), executionRow.getFailedMeasurementInfoMap()); + executionRow.removeAllFailedMeasurementMarks(); + assertEquals(1, originalRow.getFailedMeasurementNumber()); + assertNull(originalRow.getMeasurements()[2]); + assertNull(originalRow.getValues()[2]); + } + } finally { + cache.invalid("testdb"); + config.setEnablePartialInsert(originalPartialInsert); + } + } + @Test public void testExecute() { Statement statement = parseSQL("EXECUTE myquery"); From 395a9866a80a3a6395dbe60fa7cb69d9e406b03e Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:53:18 +0800 Subject: [PATCH 2/2] Narrow PR to prepared VALUES fixes --- .../it/db/it/IoTDBSetSystemStatusTableIT.java | 170 +---------- .../db/queryengine/plan/Coordinator.java | 5 - .../plan/ReadOnlyStatementValidator.java | 261 ----------------- .../executor/RegionWriteExecutorTest.java | 39 --- .../plan/ReadOnlyStatementValidatorTest.java | 277 ------------------ 5 files changed, 7 insertions(+), 745 deletions(-) delete mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java delete mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java index 98e77f39af5d5..685610a663592 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java @@ -26,7 +26,6 @@ import org.apache.iotdb.itbase.env.BaseEnv; import org.apache.iotdb.itbase.exception.InconsistentDataException; import org.apache.iotdb.itbase.runtime.ClusterTestConnection; -import org.apache.iotdb.rpc.TSStatusCode; import org.awaitility.Awaitility; import org.junit.AfterClass; @@ -38,7 +37,6 @@ import java.sql.Connection; import java.sql.ResultSet; -import java.sql.SQLException; import java.sql.Statement; import java.util.concurrent.TimeUnit; @@ -56,41 +54,24 @@ public static void tearDown() throws Exception { } @Test - public void testReadOnlySqlAndDiskRecovery() throws Exception { - try (Connection treeConnection = EnvFactory.getEnv().getConnection(); - Connection tableConnection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - Statement tree = treeConnection.createStatement(); + public void testPreparedInsertValuesExecution() throws Exception { + try (Connection tableConnection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); Statement table = tableConnection.createStatement(); - // PREPARE and EXECUTE must use the same session, including with multi-node read checks. Statement prepared = (tableConnection instanceof ClusterTestConnection ? ((ClusterTestConnection) tableConnection) .writeConnection.getUnderlyingConnection() : tableConnection) .createStatement()) { - tree.execute("create database root.readonly_sql"); - tree.execute("create database root.readonly_drop"); - tree.execute("create timeseries root.readonly_sql.d.s with datatype=INT32,encoding=PLAIN"); - tree.execute("create timeseries root.readonly_drop.d.s with datatype=INT32,encoding=PLAIN"); - tree.execute("insert into root.readonly_sql.d(time,s) values(1,1)"); - tree.execute("insert into root.readonly_drop.d(time,s) values(1,1)"); - table.execute("create database readonly_sql"); - table.execute("create database readonly_drop"); - table.execute("create table readonly_drop.t(d STRING TAG,s INT32 FIELD)"); - table.execute("insert into readonly_drop.t(time,d,s) values(1,'d1',1)"); - table.execute("use readonly_sql"); - table.execute("create table t(d STRING TAG,a STRING ATTRIBUTE,s INT32 FIELD)"); + table.execute("create database prepared_insert_execution"); + table.execute("use prepared_insert_execution"); table.execute("create table prepared_t(d STRING TAG,a STRING ATTRIBUTE,s INT32 FIELD)"); - table.execute("insert into t(time,d,a,s) values(1,'d1','original',1)"); - tree.execute("flush"); - // Resolve both a read and a write before entering read-only mode. - prepared.execute("prepare readonly_read from select s from t"); prepared.execute( - "prepare readonly_write from insert into prepared_t(time,d,a,s) " + "prepare prepared_insert from insert into prepared_t(time,d,a,s) " + "values(1,'p1','prepared',11),(2,'p2','prepared',12)"); - prepared.execute("execute readonly_write"); + prepared.execute("execute prepared_insert"); table.execute("update prepared_t set a='changed' where d='p1'"); - prepared.execute("execute readonly_write"); + prepared.execute("execute prepared_insert"); prepared.execute( "execute immediate 'insert into prepared_t(time,d,a,s) " + "values(3,''p3'',''immediate'',13)'"); @@ -103,143 +84,6 @@ public void testReadOnlySqlAndDiskRecovery() throws Exception { } Assert.assertFalse(result.next()); } - try { - table.execute("set system to readonly on cluster"); - awaitDataNodeStatus(table, "ReadOnly"); - - for (String sql : - new String[] { - "show databases", - "show timeseries root.readonly_sql.**", - "show devices root.readonly_sql.**", - "count timeseries root.readonly_sql.**", - "show pipes", - "show ttl on root.readonly_sql.**", - "list user" - }) { - assertQueryAvailable(tree, sql); - } - for (String sql : - new String[] { - "show tables", - "describe t", - "show create table t", - "show databases", - "show pipes", - "show devices from t", - "count devices from t", - "select * from information_schema.tables", - "execute immediate 'select s from t'", - "list user" - }) { - assertQueryAvailable(table, sql); - } - assertQueryAvailable(prepared, "execute readonly_read"); - assertQueryAvailable(prepared, "explain analyze execute readonly_read"); - try (ResultSet result = tree.executeQuery("select s from root.readonly_sql.d")) { - Assert.assertTrue(result.next()); - Assert.assertEquals(1, result.getInt(2)); - Assert.assertFalse(result.next()); - } - try (ResultSet result = table.executeQuery("select a,s from t where d='d1'")) { - Assert.assertTrue(result.next()); - Assert.assertEquals("original", result.getString(1)); - Assert.assertEquals(1, result.getInt(2)); - Assert.assertFalse(result.next()); - } - - for (String sql : - new String[] { - "insert into root.readonly_sql.d(time,s) values(2,2)", - "delete from root.readonly_sql.d.s where time<=1", - "select s into root.readonly_dest.d(s) from root.readonly_sql.d", - "create database root.readonly_new", - "delete timeseries root.readonly_sql.d.s", - "flush" - }) { - assertReadOnlyRejected(tree, sql); - } - for (String sql : - new String[] { - "insert into t(time,d,s) values(2,'d1',2)", - "insert into t select * from t", - "delete from t where time<=1", - "update t set a='changed' where d='d1'", - "create database readonly_new", - "create table newtable(d STRING TAG,s INT32 FIELD)", - "drop table t", - "alter database readonly_sql set properties ttl=1, schema_region_group_num=2", - "create role readonly_role", - "execute immediate 'delete from t where time<=1'", - "execute immediate 'insert into prepared_t(time,d,s) values(4,''p4'',14)'", - "flush" - }) { - assertReadOnlyRejected(table, sql); - } - assertReadOnlyRejected(prepared, "execute readonly_write"); - assertReadOnlyRejected(prepared, "explain analyze execute readonly_write"); - - tree.execute("set ttl to root.readonly_sql.** 1"); - tree.execute("delete database root.readonly_drop"); - table.execute("alter database readonly_sql set properties ttl=1"); - table.execute("alter table t set properties ttl=1"); - table.execute("drop database readonly_drop"); - try (ResultSet result = tree.executeQuery("show databases root.readonly_drop")) { - Assert.assertFalse(result.next()); - } - try (ResultSet result = table.executeQuery("show databases")) { - while (result.next()) { - Assert.assertNotEquals("readonly_drop", result.getString(1)); - } - } - // The recovery exceptions do not switch the system back to Running. - awaitDataNodeStatus(table, "ReadOnly"); - } finally { - table.execute("set system to running on cluster"); - awaitDataNodeStatus(table, "Running"); - } - // Normal writes resume after leaving read-only mode. - tree.execute("unset ttl to root.readonly_sql.**"); - table.execute("alter database readonly_sql set properties ttl=default"); - table.execute("alter table t set properties ttl=default"); - tree.execute("insert into root.readonly_sql.d(time,s) values(now(),2)"); - table.execute("insert into t(time,d,s) values(now(),'d1',2)"); - prepared.execute("deallocate prepare readonly_write"); - prepared.execute("deallocate prepare readonly_read"); - } - } - - private static void awaitDataNodeStatus(Statement statement, String expectedStatus) { - Awaitility.await() - .atMost(30, TimeUnit.SECONDS) - .untilAsserted( - () -> { - try (ResultSet result = statement.executeQuery("show datanodes")) { - int count = 0; - while (result.next()) { - Assert.assertEquals(expectedStatus, result.getString("Status")); - count++; - } - Assert.assertEquals(EnvFactory.getEnv().getDataNodeWrapperList().size(), count); - } - }); - } - - private static void assertQueryAvailable(Statement statement, String sql) throws SQLException { - try (ResultSet result = statement.executeQuery(sql)) { - Assert.assertTrue(sql, result.getMetaData().getColumnCount() > 0); - while (result.next()) { - // Consume the full result to surface errors from distributed query execution. - } - } - } - - private static void assertReadOnlyRejected(Statement statement, String sql) { - try { - statement.execute(sql); - Assert.fail("Expected SYSTEM_READ_ONLY for " + sql); - } catch (SQLException e) { - Assert.assertEquals(sql, TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), e.getErrorCode()); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java index 5c9116841cdb2..eb3d0e6869871 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java @@ -431,7 +431,6 @@ private IQueryExecution createQueryExecutionForTreeModel( ISchemaFetcher schemaFetcher, long timeOut, long startTime) { - ReadOnlyStatementValidator.check(statement); queryContext.setTimeOut(timeOut); queryContext.setStartTime(startTime); if (statement instanceof ShowPipeMemoryStatement) { @@ -678,7 +677,6 @@ private IQueryExecution createQueryExecutionForTableModel( Metadata metadata, long timeOut, long startTime) { - ReadOnlyStatementValidator.check(statement); queryContext.setTimeOut(timeOut); queryContext.setStartTime(startTime); TableModelPlanner tableModelPlanner = @@ -781,7 +779,6 @@ private IQueryExecution createQueryExecutionForTableModel( || statement instanceof RemoveRegion || statement instanceof Prepare || statement instanceof Deallocate) { - ReadOnlyStatementValidator.check(statement, sqlParser, clientSession); return new ConfigExecution( queryContext, null, @@ -865,8 +862,6 @@ private IQueryExecution createQueryExecutionForTableModel( } } - // Check the resolved SQL that will be analyzed, including EXECUTE and EXECUTE IMMEDIATE. - ReadOnlyStatementValidator.check(statementToUse, sqlParser, clientSession); if (innerStatement instanceof WrappedInsertStatement) { ((WrappedInsertStatement) innerStatement).setContext(queryContext); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java deleted file mode 100644 index 9bcb5027b9662..0000000000000 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidator.java +++ /dev/null @@ -1,261 +0,0 @@ -/* - * 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.iotdb.db.queryengine.plan; - -import org.apache.iotdb.commons.conf.CommonDescriptor; -import org.apache.iotdb.commons.exception.IoTDBRuntimeException; -import org.apache.iotdb.commons.i18n.CommonMessages; -import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Query; -import org.apache.iotdb.db.protocol.session.IClientSession; -import org.apache.iotdb.db.queryengine.plan.analyze.QueryType; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterDB; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CountDB; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CountDevice; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CountStatement; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Deallocate; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Execute; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExecuteImmediate; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Explain; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainAnalyze; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FetchDevice; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.KillQuery; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Prepare; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Property; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RelationalAuthorStatement; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetProperties; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetSqlDialect; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowAIDevices; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowAINodes; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowAvailableUrls; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCluster; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowClusterId; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowConfigNodes; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCreateDatabase; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCreatePipe; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCreateTopic; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentDatabase; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentSqlDialect; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentTimestamp; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentUser; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDB; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDataNodes; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDevice; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowFunctions; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowIndex; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowLoadedModels; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowModels; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipePlugins; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipes; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowRegions; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowStatement; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowSubscriptions; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowTables; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowTopics; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowVariables; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowVersion; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Use; -import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.WrappedStatement; -import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; -import org.apache.iotdb.db.queryengine.plan.statement.IConfigStatement; -import org.apache.iotdb.db.queryengine.plan.statement.Statement; -import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; -import org.apache.iotdb.db.queryengine.plan.statement.internal.DeviceSchemaFetchStatement; -import org.apache.iotdb.db.queryengine.plan.statement.internal.SeriesSchemaFetchStatement; -import org.apache.iotdb.db.queryengine.plan.statement.metadata.DeleteDatabaseStatement; -import org.apache.iotdb.db.queryengine.plan.statement.metadata.SetTTLStatement; -import org.apache.iotdb.db.queryengine.plan.statement.metadata.UnSetTTLStatement; -import org.apache.iotdb.db.queryengine.plan.statement.pipe.PipeEnrichedStatement; -import org.apache.iotdb.db.queryengine.plan.statement.sys.ExplainAnalyzeStatement; -import org.apache.iotdb.db.queryengine.plan.statement.sys.ExplainStatement; -import org.apache.iotdb.db.queryengine.plan.statement.sys.KillQueryStatement; -import org.apache.iotdb.db.queryengine.plan.statement.sys.SetSqlDialectStatement; -import org.apache.iotdb.db.queryengine.plan.statement.sys.SetSystemStatusStatement; -import org.apache.iotdb.rpc.TSStatusCode; - -import java.util.HashSet; -import java.util.List; -import java.util.Set; - -/** Checks SQL before analysis, which can itself create schema or allocate partitions. */ -public final class ReadOnlyStatementValidator { - - private ReadOnlyStatementValidator() {} - - public static void check(final Statement statement) { - if (CommonDescriptor.getInstance().getConfig().isReadOnly() && !isAllowed(statement)) { - reject(); - } - } - - public static void check( - final org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement statement, - final SqlParser parser, - final IClientSession session) { - if (CommonDescriptor.getInstance().getConfig().isReadOnly() - && !isAllowed(statement, parser, session)) { - reject(); - } - } - - static boolean isAllowed(final Statement statement) { - if (statement instanceof PipeEnrichedStatement) { - return isAllowed(((PipeEnrichedStatement) statement).getInnerStatement()); - } - if (statement instanceof QueryStatement) { - return !((QueryStatement) statement).isSelectInto(); - } - if (statement instanceof ExplainStatement) { - return isAllowed(((ExplainStatement) statement).getQueryStatement()); - } - if (statement instanceof ExplainAnalyzeStatement) { - return isAllowed(((ExplainAnalyzeStatement) statement).getQueryStatement()); - } - // Deleting a whole database and setting TTL are explicit disk-recovery exceptions. - // They still use the normal metadata persistence and permission checks. - if (statement instanceof DeleteDatabaseStatement - || (statement instanceof SetTTLStatement && !(statement instanceof UnSetTTLStatement)) - || statement instanceof SetSystemStatusStatement - || statement instanceof SetSqlDialectStatement - || statement instanceof KillQueryStatement) { - return true; - } - if (statement instanceof IConfigStatement) { - return ((IConfigStatement) statement).getQueryType() == QueryType.READ; - } - // SELECT resolves schema through these internal reads on a schema-cache miss. - return statement instanceof SeriesSchemaFetchStatement - || statement instanceof DeviceSchemaFetchStatement - || statement - instanceof org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowStatement - || statement.isQuery(); - } - - static boolean isAllowed( - final org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement statement, - final SqlParser parser, - final IClientSession session) { - return isAllowed(statement, parser, session, new HashSet<>()); - } - - private static boolean isAllowed( - final org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement statement, - final SqlParser parser, - final IClientSession session, - final Set preparedStatements) { - if (statement instanceof WrappedStatement) { - return isAllowed(((WrappedStatement) statement).getInnerTreeStatement()); - } - if (statement instanceof Explain) { - return isAllowed(((Explain) statement).getStatement(), parser, session, preparedStatements); - } - if (statement instanceof ExplainAnalyze) { - return isAllowed( - ((ExplainAnalyze) statement).getStatement(), parser, session, preparedStatements); - } - if (statement instanceof Execute) { - final String name = ((Execute) statement).getStatementName().getValue(); - if (!preparedStatements.add(name)) { - return false; - } - final var prepared = session.getPreparedStatement(name); - // Let the normal execution path report an unknown prepared statement. - return prepared == null || isAllowed(prepared.getSql(), parser, session, preparedStatements); - } - if (statement instanceof ExecuteImmediate) { - return isAllowed( - parser.createStatement( - ((ExecuteImmediate) statement).getSqlString(), session.getZoneId(), session), - parser, - session, - preparedStatements); - } - if (statement instanceof RelationalAuthorStatement) { - return ((RelationalAuthorStatement) statement).getQueryType() == QueryType.READ; - } - if (statement instanceof AlterDB) { - return onlySetsTTL(((AlterDB) statement).getProperties()); - } - if (statement instanceof SetProperties) { - return ((SetProperties) statement).getType() == SetProperties.Type.TABLE - && onlySetsTTL(((SetProperties) statement).getProperties()); - } - // Other SHOW forms wrap tree statements above or are rewritten to information_schema reads. - // Session-local operations are safe and must remain usable while the server is read-only. - // Table reads retain the attribute-sync registration in RegionWriteExecutor for cache - // consistency. - return statement instanceof Query - || statement instanceof ShowStatement - || statement instanceof CountStatement - || statement instanceof ShowDB - || statement instanceof ShowCreateDatabase - || statement instanceof CountDB - || statement instanceof ShowAIDevices - || statement instanceof ShowAINodes - || statement instanceof ShowAvailableUrls - || statement instanceof ShowCluster - || statement instanceof ShowClusterId - || statement instanceof ShowConfigNodes - || statement instanceof ShowCurrentDatabase - || statement instanceof ShowCurrentSqlDialect - || statement instanceof ShowCurrentTimestamp - || statement instanceof ShowCurrentUser - || statement instanceof ShowDataNodes - || statement instanceof ShowFunctions - || statement instanceof ShowIndex - || statement instanceof ShowLoadedModels - || statement instanceof ShowModels - || statement instanceof ShowPipePlugins - || statement instanceof ShowRegions - || statement instanceof ShowVariables - || statement instanceof ShowVersion - || statement instanceof ShowTables - || statement instanceof DescribeTable - || statement instanceof ShowDevice - || statement instanceof CountDevice - || statement instanceof FetchDevice - || statement instanceof ShowPipes - || statement instanceof ShowCreatePipe - || statement instanceof ShowTopics - || statement instanceof ShowCreateTopic - || statement instanceof ShowSubscriptions - || statement instanceof DropDB - || statement instanceof Use - || statement instanceof KillQuery - || statement instanceof SetSqlDialect - || statement instanceof Prepare - || statement instanceof Deallocate; - } - - private static boolean onlySetsTTL(final List properties) { - return !properties.isEmpty() - && properties.stream() - .allMatch( - property -> - "ttl".equalsIgnoreCase(property.getName().getValue()) - && !property.isSetToDefault()); - } - - private static void reject() { - throw new IoTDBRuntimeException( - CommonMessages.SYSTEM_READ_ONLY, TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), true); - } -} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java index 2d8f0fc4c5f08..22521458c17b6 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java @@ -19,11 +19,7 @@ package org.apache.iotdb.db.queryengine.execution.executor; -import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TSStatus; -import org.apache.iotdb.commons.cluster.NodeStatus; -import org.apache.iotdb.commons.conf.CommonConfig; -import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.consensus.SchemaRegionId; @@ -47,7 +43,6 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode; -import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode; import org.apache.iotdb.db.queryengine.plan.statement.metadata.AlterTimeSeriesStatement.AlterType; import org.apache.iotdb.db.schemaengine.SchemaEngine; import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion; @@ -75,40 +70,6 @@ public class RegionWriteExecutorTest { - @Test - public void testReadOnlyAllowsAttributeSyncRegistration() throws ConsensusException { - final CommonConfig config = CommonDescriptor.getInstance().getConfig(); - final NodeStatus originalStatus = config.getNodeStatus(); - final String originalReason = config.getStatusReason(); - final IConsensus consensus = Mockito.mock(IConsensus.class); - final DataNodeRegionManager regionManager = Mockito.mock(DataNodeRegionManager.class); - final SchemaRegionId regionId = new SchemaRegionId(1); - Mockito.when(regionManager.getRegionLock(regionId)).thenReturn(new ReentrantReadWriteLock()); - final RegionWriteExecutor executor = - new RegionWriteExecutor( - consensus, - consensus, - regionManager, - Mockito.mock(SchemaEngine.class), - Mockito.mock(ClusterTemplateManager.class), - Mockito.mock(TriggerFireVisitor.class)); - final TableNodeLocationAddNode registration = - new TableNodeLocationAddNode(new PlanNodeId("register"), new TDataNodeLocation()); - Mockito.when(consensus.write(regionId, registration)) - .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); - try { - for (String reason : new String[] {NodeStatus.MANUAL, NodeStatus.DISK_FULL}) { - config.setNodeStatusWithReason(NodeStatus.ReadOnly, reason); - final RegionExecutionResult result = executor.execute(regionId, registration); - assertTrue(result.isAccepted()); - assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), result.getStatus().getCode()); - } - Mockito.verify(consensus, Mockito.times(2)).write(regionId, registration); - } finally { - config.setNodeStatusWithReason(originalStatus, originalReason); - } - } - @Test public void testAlterTimeSeriesRegionAvailability() throws Exception { // Alter validation fetches the measurement before consensus; a missing region must be diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java deleted file mode 100644 index 07a17f95b1c65..0000000000000 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/ReadOnlyStatementValidatorTest.java +++ /dev/null @@ -1,277 +0,0 @@ -/* - * 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.iotdb.db.queryengine.plan; - -import org.apache.iotdb.commons.cluster.NodeStatus; -import org.apache.iotdb.commons.conf.CommonConfig; -import org.apache.iotdb.commons.conf.CommonDescriptor; -import org.apache.iotdb.commons.exception.IoTDBRuntimeException; -import org.apache.iotdb.commons.path.PartialPath; -import org.apache.iotdb.commons.path.PathPatternTree; -import org.apache.iotdb.db.protocol.session.ClientSession; -import org.apache.iotdb.db.protocol.session.PreparedStatementInfo; -import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; -import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; -import org.apache.iotdb.db.queryengine.plan.statement.Statement; -import org.apache.iotdb.db.queryengine.plan.statement.internal.DeviceSchemaFetchStatement; -import org.apache.iotdb.db.queryengine.plan.statement.internal.InternalBatchActivateTemplateStatement; -import org.apache.iotdb.db.queryengine.plan.statement.internal.InternalCreateMultiTimeSeriesStatement; -import org.apache.iotdb.db.queryengine.plan.statement.internal.SeriesSchemaFetchStatement; -import org.apache.iotdb.db.queryengine.plan.statement.pipe.PipeEnrichedStatement; -import org.apache.iotdb.rpc.TSStatusCode; - -import org.junit.After; -import org.junit.Before; -import org.junit.Test; - -import java.time.ZoneOffset; -import java.util.Collections; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.fail; - -public class ReadOnlyStatementValidatorTest { - private final CommonConfig config = CommonDescriptor.getInstance().getConfig(); - private final SqlParser parser = new SqlParser(); - private final ClientSession session = new ClientSession(null); - private NodeStatus originalStatus; - private String originalReason; - - @Before - public void setUp() { - originalStatus = config.getNodeStatus(); - originalReason = config.getStatusReason(); - config.setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.MANUAL); - session.setZoneId(ZoneOffset.UTC); - } - - @After - public void tearDown() { - config.setNodeStatusWithReason(originalStatus, originalReason); - } - - @Test - public void testTreeReads() { - for (String sql : - new String[] { - "show databases", - "show timeseries root.**", - "show devices root.**", - "count timeseries root.**", - "count devices root.**", - "show ttl on root.**", - "show pipes", - "show pipeplugins", - "show functions", - "show triggers", - "show schema templates", - "show cluster", - "show datanodes", - "show confignodes", - "show variables", - "show version", - "show current_user", - "show queries", - "show disk_usage from root.**", - "list user", - "list role", - "select s from root.db.d", - "explain select s from root.db.d", - "explain analyze select s from root.db.d" - }) { - ReadOnlyStatementValidator.check(tree(sql)); - } - } - - @Test - public void testInternalSchemaReadsAndWrites() throws Exception { - PathPatternTree patternTree = new PathPatternTree(); - patternTree.appendPathPattern(new PartialPath("root.db.**")); - patternTree.constructTree(); - ReadOnlyStatementValidator.check( - new SeriesSchemaFetchStatement( - patternTree, Collections.emptyMap(), false, false, true, false, false)); - ReadOnlyStatementValidator.check( - new DeviceSchemaFetchStatement(patternTree, patternTree, false)); - assertRejected( - () -> - ReadOnlyStatementValidator.check( - new InternalCreateMultiTimeSeriesStatement(Collections.emptyMap()))); - assertRejected( - () -> - ReadOnlyStatementValidator.check( - new InternalBatchActivateTemplateStatement(Collections.emptyMap()))); - } - - @Test - public void testTableReads() { - for (String sql : - new String[] { - "show databases", - "count databases", - "show tables from db", - "show tables details from db", - "describe db.t", - "describe db.t details", - "show create database db", - "show create table db.t", - "show pipes", - "show pipeplugins", - "show functions", - "show cluster", - "show datanodes", - "show confignodes", - "show variables", - "show version", - "show current_user", - "show current_database", - "show current_sql_dialect", - "show devices from db.t", - "count devices from db.t", - "list user", - "list role", - "select * from db.t", - "select * from information_schema.tables", - "explain select * from db.t", - "explain analyze select * from db.t" - }) { - checkTable(sql); - } - } - - @Test - public void testDiskRecoveryAndSessionOperations() { - ReadOnlyStatementValidator.check(tree("delete database root.db")); - ReadOnlyStatementValidator.check(tree("set ttl to root.db.** 1")); - ReadOnlyStatementValidator.check(tree("set system to running on local")); - ReadOnlyStatementValidator.check(tree("set sql_dialect=table")); - checkTable("drop database db"); - checkTable("alter database db set properties ttl=1"); - checkTable("alter table db.t set properties ttl=1"); - checkTable("set system to running on local"); - checkTable("use db"); - checkTable("kill query '20250206_093300_00001_1'"); - checkTable("set sql_dialect=tree"); - checkTable("prepare q from select * from db.t"); - checkTable("deallocate prepare q"); - } - - @Test - public void testTreeWritesAreRejected() { - for (String sql : - new String[] { - "insert into root.db.d(time,s) values(1,1)", - "delete from root.db.d.s where time<=1", - "select s into root.dest.d(s) from root.db.d", - "explain analyze select s into root.dest.d(s) from root.db.d", - "create database root.newdb", - "create timeseries root.db.d.s with datatype=INT32,encoding=PLAIN", - "delete timeseries root.db.d.s", - "create user readonlyuser 'IoTDB!Test2024'", - "flush", - "start repair data", - "unset ttl to root.db.**", - "set configuration 'query_timeout_threshold'='10000'" - }) { - assertRejected(sql, () -> ReadOnlyStatementValidator.check(tree(sql))); - } - assertRejected( - () -> - ReadOnlyStatementValidator.check( - new PipeEnrichedStatement(tree("create database root.newdb")))); - } - - @Test - public void testTableWritesAreRejected() { - for (String sql : - new String[] { - "insert into db.t select * from db.source", - "insert into db.t(time,s) select * from db.source", - "delete from db.t where time<=1", - "update db.t set a='new' where tag='d'", - "create database newdb", - "create table db.newtable(tag STRING TAG,s INT32 FIELD)", - "drop table db.t", - "alter table db.t add column s2 INT32 FIELD", - "alter database db set properties ttl=1, schema_region_group_num=2", - "alter table db.t set properties ttl=1, comment='changed'", - "alter table db.t set properties ttl=default", - "create role readonlyrole", - "flush", - "set configuration query_timeout_threshold='10000'", - "explain analyze execute immediate 'insert into db.t select * from db.source'" - }) { - assertRejected(sql, () -> checkTable(sql)); - } - } - - @Test - public void testPreparedExecutionChecksTheInnerStatement() { - session.addPreparedStatement( - "q", new PreparedStatementInfo("q", table("select * from db.t"), 0)); - session.addPreparedStatement( - "w", new PreparedStatementInfo("w", table("insert into db.t select * from db.source"), 0)); - checkTable("execute q"); - checkTable("explain analyze execute q"); - checkTable("execute immediate 'select * from db.t'"); - assertRejected(() -> checkTable("execute w")); - assertRejected(() -> checkTable("explain analyze execute w")); - assertRejected( - () -> checkTable("execute immediate 'insert into db.t select * from db.source'")); - session.addPreparedStatement( - "cycle", new PreparedStatementInfo("cycle", table("execute cycle"), 0)); - assertRejected(() -> checkTable("execute cycle")); - } - - @Test - public void testRunningModeKeepsExistingBehavior() { - config.setNodeStatus(NodeStatus.Running); - ReadOnlyStatementValidator.check(tree("create database root.db")); - checkTable("create database db"); - } - - private Statement tree(String sql) { - return StatementGenerator.createStatement(sql, ZoneOffset.UTC); - } - - private org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement table(String sql) { - return parser.createStatement(sql, ZoneOffset.UTC, session); - } - - private void checkTable(String sql) { - ReadOnlyStatementValidator.check(table(sql), parser, session); - } - - private void assertRejected(Runnable action) { - assertRejected("", action); - } - - private void assertRejected(String sql, Runnable action) { - try { - action.run(); - fail("Expected SYSTEM_READ_ONLY for " + sql); - } catch (IoTDBRuntimeException e) { - assertEquals( - sql + ": " + e.getMessage(), - TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), - e.getErrorCode()); - } - } -}