diff --git a/tests/pytest/insert/insert_drop.py b/tests/pytest/insert/insert_drop.py
new file mode 100644
index 0000000000000000000000000000000000000000..5ccaa5540bb30b650f732632fb6262c124783dfe
--- /dev/null
+++ b/tests/pytest/insert/insert_drop.py
@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+
+import sys
+from util.log import *
+from util.cases import *
+from util.sql import *
+import threading
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), logSql)
+
+    def genMultiThreadSeq(self, sql_list):
+        tlist = list()
+        for insert_sql in sql_list:
+            t = threading.Thread(target=tdSql.execute, args=(insert_sql,))
+            tlist.append(t)
+        return tlist
+
+    def multiThreadRun(self, tlist):
+        for t in tlist:
+            t.start()
+        for t in tlist:
+            t.join()
+
+    def run(self):
+        tdSql.prepare()
+        tdSql.execute('create database if not exists test;')
+        tdSql.execute('create table test.stb (ts timestamp, c11 int, c12 float ) TAGS(t11 int, t12 int );')
+        tdSql.execute('create table test.tb using test.stb TAGS (1, 1);')
+        sql_list = list()
+        for i in range(5):
+            sql = f'insert into test.tb values (now-{i}m, {i}, {i});'
+            sql_list.append(sql)
+        sql_list.append(f'drop database test;')
+        tlist = self.genMultiThreadSeq(sql_list)
+        self.multiThreadRun(tlist)
+        tdSql.query(f'show databases')
+        
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())