feat: support automatic schema creation in Session

integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBSessionRelationalIT.java

@@ -456,6 +456,45 @@ public void partialInsertSQLTest() throws IoTDBConnectionException, StatementExe
     }
   }
 
+  @Test
+  public void disableAutoCreateSchemaForTableSessionTest()
+      throws IoTDBConnectionException, StatementExecutionException {
+    final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+    try (ITableSession session =
+        new TableSessionBuilder()
+            .nodeUrls(Collections.singletonList(dataNode.getIpAndPortString()))
+            .database("db1")
+            .enableAutoCreateSchema(false)
+            .build()) {
+      final List<IMeasurementSchema> schemaList = new ArrayList<>();
+      schemaList.add(new MeasurementSchema("tag1", TSDataType.STRING));
+      schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+      schemaList.add(new MeasurementSchema("s2", TSDataType.INT64));
+      final Tablet tablet =
+          new Tablet(
+              "session_auto_create",
+              IMeasurementSchema.getMeasurementNameList(schemaList),
+              IMeasurementSchema.getDataTypeList(schemaList),
+              Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD),
+              1);
+      tablet.addTimestamp(0, 1);
+      tablet.addValue("tag1", 0, "d1");
+      tablet.addValue("s1", 0, 1L);
+      tablet.addValue("s2", 0, 2L);
+
+      assertThrows(StatementExecutionException.class, () -> session.insert(tablet));
+
+      session.executeNonQueryStatement(
+          "CREATE TABLE session_auto_create (tag1 string tag, s1 int64 field)");
+      assertThrows(StatementExecutionException.class, () -> session.insert(tablet));
+
+      try (SessionDataSet dataSet =
+          session.executeQueryStatement("SELECT * FROM session_auto_create")) {
+        assertEquals(Arrays.asList("time", "tag1", "s1"), dataSet.getColumnNames());
+      }
+    }
+  }
+
   @Test
   public void partialInsertRelationalTabletTest()
       throws IoTDBConnectionException, StatementExecutionException {

iotdb-client/session/src/main/java/org/apache/iotdb/session/AbstractSessionBuilder.java

@@ -80,6 +80,8 @@ public abstract class AbstractSessionBuilder {
 
   public String database;
 
+  public boolean enableAutoCreateSchema = true;
+
   public Boolean isIoTDBRpcCompressionEnabled = true;
 
   public Boolean isThriftRpcCompactionEnabled = false;

iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java

@@ -204,6 +204,8 @@ public class Session implements ISession {
   // may be null
   protected volatile String database;
 
+  protected volatile boolean enableAutoCreateSchema = true;
+
   private static final String VALUES_SIZE_SHOULD_BE_EQUAL =
       "times, measurementsList and valuesList's size should be equal";
 
@@ -484,6 +486,7 @@ public Session(AbstractSessionBuilder builder) {
     this.sqlDialect = builder.sqlDialect;
     this.queryTimeoutInMs = builder.timeOut;
     this.database = builder.database;
+    this.enableAutoCreateSchema = builder.enableAutoCreateSchema;
   }
 
   @Override
@@ -4523,6 +4526,11 @@ public Builder database(String database) {
       return this;
     }
 
+    public Builder enableAutoCreateSchema(boolean enableAutoCreateSchema) {
+      this.enableAutoCreateSchema = enableAutoCreateSchema;
+      return this;
+    }
+
     public Builder useSSL(boolean useSSL) {
       this.useSSL = useSSL;
       return this;

iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java

@@ -99,6 +99,7 @@ public class SessionConnection {
 
   private static final Logger logger = LoggerFactory.getLogger(SessionConnection.class);
   private static final String USE_ENCRYPTED_PASSWORD_KEY = "use_encrypted_password";
+  private static final String ENABLE_AUTO_CREATE_SCHEMA_KEY = "enable_auto_create_schema";
   public static final String MSG_RECONNECTION_FAIL =
       "Fail to reconnect to server. Please check server status.";
   protected Session session;
@@ -244,6 +245,9 @@ private void init(
     if (session.useEncryptedPassword) {
       openReq.putToConfiguration(USE_ENCRYPTED_PASSWORD_KEY, Boolean.TRUE.toString());
     }
+    if (!session.enableAutoCreateSchema) {
+      openReq.putToConfiguration(ENABLE_AUTO_CREATE_SCHEMA_KEY, Boolean.FALSE.toString());
+    }
     if (database != null) {
       openReq.putToConfiguration("db", database);
     }

iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java

@@ -179,6 +179,18 @@ public TableSessionBuilder enableAutoFetch(boolean enableAutoFetch) {
     return this;
   }
 
+  /**
+   * Enables or disables schema auto-creation for this session's write requests.
+   *
+   * @param enableAutoCreateSchema whether to allow auto creating tables and columns.
+   * @return the current {@link TableSessionBuilder} instance.
+   * @defaultValue true
+   */
+  public TableSessionBuilder enableAutoCreateSchema(boolean enableAutoCreateSchema) {
+    this.enableAutoCreateSchema = enableAutoCreateSchema;
+    return this;
+  }
+
   /**
    * Sets the maximum number of retries for connection attempts.
    *

iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java

@@ -188,6 +188,8 @@ public class SessionPool implements ISessionPool {
   // may be null
   protected String database;
 
+  protected boolean enableAutoCreateSchema = true;
+
   private static final String INSERT_RECORD_FAIL = SessionMessages.INSERT_RECORD_FAILED;
 
   private static final String INSERT_RECORD_ERROR_MSG =
@@ -553,6 +555,7 @@ public SessionPool(AbstractSessionPoolBuilder builder) {
     this.sqlDialect = builder.sqlDialect;
     this.database = builder.database;
     this.queryTimeoutInMs = builder.timeOut;
+    this.enableAutoCreateSchema = builder.enableAutoCreateSchema;
 
     if (enableAutoFetch) {
       initThreadPool();
@@ -612,6 +615,7 @@ private Session constructNewSession() {
               .retryIntervalInMs(retryIntervalInMs)
               .sqlDialect(sqlDialect)
               .database(database)
+              .enableAutoCreateSchema(enableAutoCreateSchema)
               .timeOut(queryTimeoutInMs)
               .enableIoTDBRpcCompression(enableIoTDBRpcCompression)
               .enableThriftRpcCompression(enableThriftCompression)
@@ -640,6 +644,7 @@ private Session constructNewSession() {
               .retryIntervalInMs(retryIntervalInMs)
               .sqlDialect(sqlDialect)
               .database(database)
+              .enableAutoCreateSchema(enableAutoCreateSchema)
               .timeOut(queryTimeoutInMs)
               .enableIoTDBRpcCompression(enableIoTDBRpcCompression)
               .enableThriftRpcCompression(enableThriftCompression)

iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java

@@ -317,6 +317,18 @@ public TableSessionPoolBuilder sslProtocol(String sslProtocol) {
     return this;
   }
 
+  /**
+   * Enables or disables schema auto-creation for this pool's write requests.
+   *
+   * @param enableAutoCreateSchema whether to allow auto creating tables and columns.
+   * @return the current {@link TableSessionPoolBuilder} instance.
+   * @defaultValue true
+   */
+  public TableSessionPoolBuilder enableAutoCreateSchema(boolean enableAutoCreateSchema) {
+    this.enableAutoCreateSchema = enableAutoCreateSchema;
+    return this;
+  }
+
   /**
    * Builds and returns a configured {@link ITableSessionPool} instance.
    *

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/IClientSession.java

@@ -59,6 +59,8 @@ public abstract class IClientSession {
 
   @Nullable private String databaseName;
 
+  private boolean enableAutoCreateSchema = true;
+
   private long lastActiveTime = CommonDateTimeUtils.currentTime();
 
   public abstract String getClientAddress();
@@ -201,6 +203,14 @@ public void setDatabaseName(@Nullable String databaseName) {
     }
   }
 
+  public boolean isEnableAutoCreateSchema() {
+    return enableAutoCreateSchema;
+  }
+
+  public void setEnableAutoCreateSchema(boolean enableAutoCreateSchema) {
+    this.enableAutoCreateSchema = enableAutoCreateSchema;
+  }
+
   /**
    * Add a prepared statement to this session.
    *

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java

@@ -510,7 +510,8 @@ public SessionInfo getSessionInfo(IClientSession session) {
         session.getZoneId(),
         session.getClientVersion(),
         session.getDatabaseName(),
-        session.getSqlDialect());
+        session.getSqlDialect(),
+        session.isEnableAutoCreateSchema());
   }
 
   // Sometimes we need to switch from table model to tree model,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java

@@ -270,6 +270,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
 
   public static final String ERROR_CODE = DataNodeMiscMessages.ERROR_CODE;
   private static final String USE_ENCRYPTED_PASSWORD_KEY = "use_encrypted_password";
+  private static final String ENABLE_AUTO_CREATE_SCHEMA_KEY = "enable_auto_create_schema";
 
   private static final TSProtocolVersion CURRENT_RPC_VERSION =
       TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3;
@@ -1633,8 +1634,11 @@ public TSOpenSessionResp openSession(TSOpenSessionReq req) throws TException {
             parseUseEncryptedPassword(req));
     TSStatus tsStatus = RpcUtils.getStatus(openSessionResp.getCode(), openSessionResp.getMessage());
 
-    if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode() && database.isPresent()) {
-      clientSession.setDatabaseName(database.get().toLowerCase(Locale.ENGLISH));
+    if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      if (database.isPresent()) {
+        clientSession.setDatabaseName(database.get().toLowerCase(Locale.ENGLISH));
+      }
+      clientSession.setEnableAutoCreateSchema(parseEnableAutoCreateSchema(req));
     }
     TSOpenSessionResp resp = new TSOpenSessionResp(tsStatus, CURRENT_RPC_VERSION);
     Map<String, String> configuration = new HashMap<>();
@@ -1678,6 +1682,13 @@ private boolean parseUseEncryptedPassword(TSOpenSessionReq req) {
         && Boolean.parseBoolean(configuration.get(USE_ENCRYPTED_PASSWORD_KEY));
   }
 
+  private boolean parseEnableAutoCreateSchema(TSOpenSessionReq req) {
+    Map<String, String> configuration = req.configuration;
+    return configuration == null
+        || !configuration.containsKey(ENABLE_AUTO_CREATE_SCHEMA_KEY)
+        || Boolean.parseBoolean(configuration.get(ENABLE_AUTO_CREATE_SCHEMA_KEY));
+  }
+
   @Override
   public TSStatus closeSession(TSCloseSessionReq req) {
     return new TSStatus(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java

@@ -368,7 +368,8 @@ public void validateInsertNodeMeasurements(
     final List<Integer> missingMeasurementIndices = new ArrayList<>();
 
     final boolean isAutoCreateSchemaEnabled =
-        IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled();
+        IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
+            && context.getSession().isEnableAutoCreateSchema();
 
     // First round validate, check existing schema
     if (table == null) {

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/common/SessionInfo.java

@@ -46,12 +46,17 @@ public class SessionInfo {
 
   private ClientVersion version = ClientVersion.V_1_0;
 
+  // Only used during statement analysis on the coordinator. Analysis is complete before the
+  // SessionInfo is sent to execution nodes, so this field does not need serialization.
+  private final transient boolean enableAutoCreateSchema;
+
   public SessionInfo(long sessionId, UserEntity userEntity, ZoneId zoneId) {
     this.sessionId = sessionId;
     this.userEntity = userEntity;
     this.zoneId = zoneId;
     this.databaseName = null;
     this.sqlDialect = SqlDialect.TREE;
+    this.enableAutoCreateSchema = true;
   }
 
   public SessionInfo(
@@ -60,7 +65,7 @@ public SessionInfo(
       ZoneId zoneId,
       @Nullable String databaseName,
       SqlDialect sqlDialect) {
-    this(sessionId, userEntity, zoneId, ClientVersion.V_1_0, databaseName, sqlDialect);
+    this(sessionId, userEntity, zoneId, ClientVersion.V_1_0, databaseName, sqlDialect, true);
   }
 
   public SessionInfo(
@@ -70,12 +75,24 @@ public SessionInfo(
       ClientVersion version,
       @Nullable String databaseName,
       SqlDialect sqlDialect) {
+    this(sessionId, userEntity, zoneId, version, databaseName, sqlDialect, true);
+  }
+
+  public SessionInfo(
+      long sessionId,
+      UserEntity userEntity,
+      ZoneId zoneId,
+      ClientVersion version,
+      @Nullable String databaseName,
+      SqlDialect sqlDialect,
+      boolean enableAutoCreateSchema) {
     this.sessionId = sessionId;
     this.userEntity = userEntity;
     this.zoneId = zoneId;
     this.version = version;
     this.databaseName = databaseName;
     this.sqlDialect = sqlDialect;
+    this.enableAutoCreateSchema = enableAutoCreateSchema;
   }
 
   @TestOnly
@@ -144,6 +161,10 @@ public SqlDialect getSqlDialect() {
     return sqlDialect;
   }
 
+  public boolean isEnableAutoCreateSchema() {
+    return enableAutoCreateSchema;
+  }
+
   public static SessionInfo deserializeFrom(final ByteBuffer buffer) {
     final long sessionId = ReadWriteIOUtils.readLong(buffer);
     final long userId = ReadWriteIOUtils.readLong(buffer);