Repository: matrixorigin/matrixone (go)

Issue:
[Bug]: data race occurred during bvt test 
### Is there an existing issue for the same bug?

- [X] I have checked the existing issues.

### Branch Name

2.0-dev

### Commit ID

0b8641a

### Other Environment Information

```Markdown
- Hardware parameters:
- OS type:
- Others:
```


### Actual Behavior


https://github.com/matrixorigin/matrixone/actions/runs/11968300034/job/33367831087

```
==================
WARNING: DATA RACE
Read at 0x00c0333492f0 by goroutine 90348:
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.fulltextIndexTokenizePrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/fulltext_tokenize.go:83 +0x448
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:151 +0x2289
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).ApplyPrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:172 +0x48
  github.com/matrixorigin/matrixone/pkg/sql/colexec/apply.(*Apply).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/apply/apply.go:54 +0x315
  github.com/matrixorigin/matrixone/pkg/vm.Prepare.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:37 +0x5a
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:570 +0x237
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.HandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:574 +0x87
  github.com/matrixorigin/matrixone/pkg/vm.Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:36 +0xfe
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:65 +0x244
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).RunWithReader()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:59 +0x2d1
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:197 +0xfe5
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).MergeRun.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:292 +0xc4
  github.com/panjf2000/ants/v2.(*goWorker).run.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:67 +0x18f

Previous write at 0x00c0333492f0 by goroutine 90705:
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.fulltextIndexTokenizePrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/fulltext_tokenize.go:83 +0x4f4
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:151 +0x2289
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).ApplyPrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:172 +0x48
  github.com/matrixorigin/matrixone/pkg/sql/colexec/apply.(*Apply).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/apply/apply.go:54 +0x315
  github.com/matrixorigin/matrixone/pkg/vm.Prepare.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:37 +0x5a
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:570 +0x237
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.HandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:574 +0x87
  github.com/matrixorigin/matrixone/pkg/vm.Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:36 +0xfe
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:65 +0x244
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).RunWithReader()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:59 +0x2d1
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:197 +0xfe5
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).MergeRun.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:292 +0xc4
  github.com/panjf2000/ants/v2.(*goWorker).run.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:67 +0x18f

Goroutine 90348 (running) created at:
  github.com/panjf2000/ants/v2.(*goWorker).run()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:48 +0xe9
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:337 +0xef
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:348 +0x2a6
  github.com/panjf2000/ants/v2.(*Pool).Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:222 +0xa6
  github.com/panjf2000/ants/v2.Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/ants.go:103 +0x4e
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).MergeRun()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:288 +0x6dc
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile.go:331 +0xb46
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).runOnce.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile.go:497 +0x190
  github.com/panjf2000/ants/v2.(*goWorker).run.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:67 +0x18f

Goroutine 90705 (running) created at:
  github.com/panjf2000/ants/v2.(*goWorker).run()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:48 +0xe9
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:337 +0xef
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:348 +0x2a6
  github.com/panjf2000/ants/v2.(*Pool).Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:222 +0xa6
  github.com/panjf2000/ants/v2.Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/ants.go:103 +0x4e
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).MergeRun()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:288 +0x6dc
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile.go:331 +0xb46
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).runOnce.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile.go:497 +0x190
  github.com/panjf2000/ants/v2.(*goWorker).run.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:67 +0x18f
==================
==================
WARNING: DATA RACE
Write at 0x00c0333492f0 by goroutine 90348:
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.fulltextIndexTokenizePrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/fulltext_tokenize.go:83 +0x4f4
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:151 +0x2289
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).ApplyPrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:172 +0x48
  github.com/matrixorigin/matrixone/pkg/sql/colexec/apply.(*Apply).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/apply/apply.go:54 +0x315
  github.com/matrixorigin/matrixone/pkg/vm.Prepare.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:37 +0x5a
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:570 +0x237
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.HandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:574 +0x87
  github.com/matrixorigin/matrixone/pkg/vm.Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:36 +0xfe
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:65 +0x244
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).RunWithReader()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:59 +0x2d1
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:197 +0xfe5
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).MergeRun.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:292 +0xc4
  github.com/panjf2000/ants/v2.(*goWorker).run.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:67 +0x18f

Previous write at 0x00c0333492f0 by goroutine 90346:
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.fulltextIndexTokenizePrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/fulltext_tokenize.go:83 +0x4f4
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:151 +0x2289
  github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function.(*TableFunction).ApplyPrepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/table_function/table_function.go:172 +0x48
  github.com/matrixorigin/matrixone/pkg/sql/colexec/apply.(*Apply).Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/colexec/apply/apply.go:54 +0x315
  github.com/matrixorigin/matrixone/pkg/vm.Prepare.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:37 +0x5a
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:570 +0x237
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.doHandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:566 +0x17e
  github.com/matrixorigin/matrixone/pkg/vm.HandleAllOp()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/types.go:574 +0x87
  github.com/matrixorigin/matrixone/pkg/vm.Prepare()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/vm.go:36 +0xfe
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:65 +0x244
  github.com/matrixorigin/matrixone/pkg/vm/pipeline.(*Pipeline).RunWithReader()
      /home/runner/_work/matrixone/matrixone/head/pkg/vm/pipeline/pipeline.go:59 +0x2d1
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:197 +0xfe5
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).MergeRun.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:292 +0xc4
  github.com/panjf2000/ants/v2.(*goWorker).run.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:67 +0x18f

Goroutine 90348 (running) created at:
  github.com/panjf2000/ants/v2.(*goWorker).run()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:48 +0xe9
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:337 +0xef
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:348 +0x2a6
  github.com/panjf2000/ants/v2.(*Pool).Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:222 +0xa6
  github.com/panjf2000/ants/v2.Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/ants.go:103 +0x4e
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Scope).MergeRun()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/scope.go:288 +0x6dc
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile.go:331 +0xb46
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).runOnce.func1()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile.go:497 +0x190
  github.com/panjf2000/ants/v2.(*goWorker).run.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:67 +0x18f

Goroutine 90346 (running) created at:
  github.com/panjf2000/ants/v2.(*goWorker).run()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/worker.go:48 +0xe9
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker.func1()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:337 +0xef
  github.com/panjf2000/ants/v2.(*Pool).retrieveWorker()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:348 +0x2a6
  github.com/panjf2000/ants/v2.(*Pool).Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/pool.go:222 +0xa6
  github.com/panjf2000/ants/v2.Submit()
      /home/runner/go/pkg/mod/github.com/panjf2000/ants/v2@v2.7.4/ants.go:103 +0x4e
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).runOnce()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile.go:487 +0x43c
  github.com/matrixorigin/matrixone/pkg/sql/compile.(*Compile).Run()
      /home/runner/_work/matrixone/matrixone/head/pkg/sql/compile/compile2.go:245 +0x160f
  github.com/matrixorigin/matrixone/pkg/frontend.executeStatusStmt()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/status_stmt.go:141 +0x188e
  github.com/matrixorigin/matrixone/pkg/frontend.executeStmt()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:2799 +0x2ec8
  github.com/matrixorigin/matrixone/pkg/frontend.dispatchStmt()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:2647 +0x6f6
  github.com/matrixorigin/matrixone/pkg/frontend.executeStmtWithIncrStmt()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:2616 +0x9b5
  github.com/matrixorigin/matrixone/pkg/frontend.executeStmtWithWorkspace()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:2560 +0xeee
  github.com/matrixorigin/matrixone/pkg/frontend.executeStmtWithTxn()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:2451 +0x384
  github.com/matrixorigin/matrixone/pkg/frontend.executeStmtWithResponse()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:2414 +0xa46
  github.com/matrixorigin/matrixone/pkg/frontend.doComQuery()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:3048 +0x4b04
  github.com/matrixorigin/matrixone/pkg/frontend.ExecRequest()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/mysql_cmd_executor.go:3125 +0x12e4
  github.com/matrixorigin/matrixone/pkg/frontend.(*Routine).handleRequest()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/routine.go:298 +0x10f1
  github.com/matrixorigin/matrixone/pkg/frontend.(*RoutineManager).Handler()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/routine_manager.go:385 +0x875
  github.com/matrixorigin/matrixone/pkg/frontend.(*MOServer).handleRequest()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/server.go:516 +0x484
  github.com/matrixorigin/matrixone/pkg/frontend.(*MOServer).handleMessage()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/server.go:484 +0x10a
  github.com/matrixorigin/matrixone/pkg/frontend.(*MOServer).handleLoop()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/server.go:212 +0x91
  github.com/matrixorigin/matrixone/pkg/frontend.(*MOServer).handleConn()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/server.go:208 +0x8fa
  github.com/matrixorigin/matrixone/pkg/frontend.(*MOServer).startAccept.gowrap2()
      /home/runner/_work/matrixone/matrixone/head/pkg/frontend/server.go:177 +0x9e
==================
```



### Expected Behavior

_No response_

### Steps to Reproduce

```Markdown
run ut with race enabled
```


### Additional information

_No response_

Pull request description:
SQL executor with streaming and fulltext index with pushdown limit
## What type of PR is this?

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #20213 #20175 #20311 #20157

## What this PR does / why we need it:

1. SQL Executor with streaming mode
2. bug fix #20311 
3. bug fix alter table drop column with fulltext index

Patch:
diff --git a/pkg/sql/colexec/table_function/fulltext.go b/pkg/sql/colexec/table_function/fulltext.go
index 4c8b367da..be1e0bb48 100644
--- a/pkg/sql/colexec/table_function/fulltext.go
+++ b/pkg/sql/colexec/table_function/fulltext.go
@@ -17,7 +17,6 @@ package table_function
 import (
 	"fmt"
 	"strings"
-	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/common/fulltext"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
@@ -38,10 +37,16 @@ const (
 
 type fulltextState struct {
 	inited      bool
-	result_chan chan map[any]float32
 	errors      chan error
-	done        chan bool
+	stream_chan chan executor.Result
+	score_array []map[any]float32
+	n_doc_id    int
+	last_doc_id any
+	n_result    uint64
+	sql_closed  bool
+	sacc        *fulltext.SearchAccum
 	limit       uint64
+	nrows       int
 
 	// holding output batch
 	batch *batch.Batch
@@ -58,33 +63,20 @@ func (u *fulltextState) free(tf *TableFunction, proc *process.Process, pipelineF
 		u.batch.Clean(proc.Mp())
 	}
 
-	close(u.done)
-}
-
-func (u *fulltextState) call(tf *TableFunction, proc *process.Process) (vm.CallResult, error) {
-
-	var scoremap map[any]float32
-	var ok bool
-
-	u.batch.CleanOnlyData()
-
-	select {
-	case <-proc.Ctx.Done():
-		return vm.CancelResult, nil
-	case err, ok := <-u.errors:
-		if ok {
-			// error
-			return vm.CancelResult, err
-		} else {
-			// channel closed
-			return vm.CancelResult, nil
-		}
-	case scoremap, ok = <-u.result_chan:
-		if !ok {
-			return vm.CancelResult, nil
+	for {
+		select {
+		case res, ok := <-u.stream_chan:
+			if !ok {
+				return
+			}
+			res.Close()
+		case <-proc.Ctx.Done():
+			return
 		}
 	}
+}
 
+func (u *fulltextState) returnResult(proc *process.Process, scoremap map[any]float32) (vm.CallResult, error) {
 	// return result
 	if u.batch.VectorCount() == 1 {
 		// only doc_id returned
@@ -116,12 +108,46 @@ func (u *fulltextState) call(tf *TableFunction, proc *process.Process) (vm.CallR
 	}
 
 	u.batch.SetRowCount(len(scoremap))
+	u.n_result += uint64(len(scoremap))
 
 	if u.batch.RowCount() == 0 {
 		return vm.CancelResult, nil
 	}
 
 	return vm.CallResult{Status: vm.ExecNext, Batch: u.batch}, nil
+
+}
+
+func (u *fulltextState) call(tf *TableFunction, proc *process.Process) (vm.CallResult, error) {
+
+	var scoremap map[any]float32
+	u.batch.CleanOnlyData()
+
+	// number of result more than pushdown limit and exit
+	if u.limit > 0 && u.n_result >= u.limit {
+		return vm.CancelResult, nil
+	}
+
+	// return scoremap when array is not empty
+	if len(u.score_array) > 0 {
+		scoremap, u.score_array = u.score_array[0], u.score_array[1:]
+		return u.returnResult(proc, scoremap)
+	}
+
+	// array is empty, try to get batch from SQL executor
+	for !u.sql_closed {
+		sql_closed, err := getResults(u, proc, u.sacc)
+		if err != nil {
+			return vm.CancelResult, err
+		}
+		u.sql_closed = sql_closed
+
+		if len(u.score_array) > 0 {
+			scoremap, u.score_array = u.score_array[0], u.score_array[1:]
+			return u.returnResult(proc, scoremap)
+		}
+	}
+	return vm.CancelResult, nil
 }
 
 // start calling tvf on nthRow and put the result in u.batch.  Note that current unnest impl will
@@ -130,9 +156,9 @@ func (u *fulltextState) start(tf *TableFunction, proc *process.Process, nthRow i
 
 	if !u.inited {
 		u.batch = tf.createResultBatch()
-		u.result_chan = make(chan map[any]float32, 2)
 		u.errors = make(chan error)
-		u.done = make(chan bool)
+		u.stream_chan = make(chan executor.Result, 8)
+		u.score_array = make([]map[any]float32, 0, 512)
 		u.inited = true
 	}
 
@@ -170,10 +196,6 @@ func fulltextIndexScanPrepare(proc *process.Process, tableFunction *TableFunctio
 	tableFunction.ctr.executorsForArgs, err = colexec.NewExpressionExecutorsFromPlanExpressions(proc, tableFunction.Args)
 	tableFunction.ctr.argVecs = make([]*vector.Vector, len(tableFunction.Args))
 
-	for i := range tableFunction.Attrs {
-		tableFunction.Attrs[i] = strings.ToUpper(tableFunction.Attrs[i])
-	}
-
 	if tableFunction.Limit != nil {
 		if cExpr, ok := tableFunction.Limit.Expr.(*plan.Expr_Lit); ok {
 			if c, ok := cExpr.Lit.Value.(*plan.Literal_U64Val); ok {
@@ -184,7 +206,9 @@ func fulltextIndexScanPrepare(proc *process.Process, tableFunction *TableFunctio
 	return st, err
 }
 
-func ft_runSql(proc *process.Process, sql string) (executor.Result, error) {
+var ft_runSql = ft_runSql_fn
+
+func ft_runSql_fn(proc *process.Process, sql string) (executor.Result, error) {
 	v, ok := moruntime.ServiceRuntime(proc.GetService()).GetGlobalVariables(moruntime.InternalSQLExecutor)
 	if !ok {
 		panic("missing lock service")
@@ -201,6 +225,27 @@ func ft_runSql(proc *process.Process, sql string) (executor.Result, error) {
 	return exec.Exec(proc.GetTopContext(), sql, opts)
 }
 
+var ft_runSql_streaming = ft_runSql_streaming_fn
+
+// run SQL in WithStreaming() and pass the channel to SQL executor
+func ft_runSql_streaming_fn(proc *process.Process, sql string, stream_chan chan executor.Result) (executor.Result, error) {
+	v, ok := moruntime.ServiceRuntime(proc.GetService()).GetGlobalVariables(moruntime.InternalSQLExecutor)
+	if !ok {
+		panic("missing lock service")
+	}
+	exec := v.(executor.SQLExecutor)
+	opts := executor.Options{}.
+		// All runSql and runSqlWithResult is a part of input sql, can not incr statement.
+		// All these sub-sql's need to be rolled back and retried en masse when they conflict in pessimistic mode
+		WithDisableIncrStatement().
+		WithTxn(proc.GetTxnOperator()).
+		WithDatabase(proc.GetSessionInfo().Database).
+		WithTimeZone(proc.GetSessionInfo().TimeZone).
+		WithAccountID(proc.GetSessionInfo().AccountId).
+		WithStreaming(stream_chan)
+	return exec.Exec(proc.GetTopContext(), sql, opts)
+}
+
 // run SQL to get the (doc_id, pos) of all patterns (words) in the search string
 func runWordStats(u *fulltextState, proc *process.Process, s *fulltext.SearchAccum) (executor.Result, error) {
 	var union []string
@@ -243,7 +288,7 @@ func runWordStats(u *fulltextState, proc *process.Process, s *fulltext.SearchAcc
 	}
 	//logutil.Infof("SQL is %s", sql)
 
-	res, err := ft_runSql(proc, sql)
+	res, err := ft_runSql_streaming(proc, sql, u.stream_chan)
 	if err != nil {
 		return executor.Result{}, err
 	}
@@ -251,131 +296,99 @@ func runWordStats(u *fulltextState, proc *process.Process, s *fulltext.SearchAcc
 	return res, nil
 }
 
-func getResults(u *fulltextState, proc *process.Process, s *fulltext.SearchAccum, res executor.Result) {
+// get one batch and evaluate the result in mini batches with size 8192
+func getResults(u *fulltextState, proc *process.Process, s *fulltext.SearchAccum) (stream_closed bool, err error) {
 
-	defer res.Close()
-	defer close(u.result_chan)
-	var n_result uint64 = 0
-	n_doc_id := 0
-	var last_doc_id any = nil
-	for _, bat := range res.Batches {
-
-		if len(bat.Vecs) != 3 {
-			u.errors <- moerr.NewInternalError(proc.Ctx, "output vector columns not match")
-			return
+	// first receive the batch and calculate the scoremap
+	// We don't need to calculate mini-batch?????
+	var res executor.Result
+	var ok bool
+	select {
+	case res, ok = <-u.stream_chan:
+		if !ok {
+			// channel closed and evaluate the rest of result
+			scoremap, err := s.Eval()
+			if err != nil {
+				return false, err
+			}
+			if len(scoremap) > 0 {
+				u.score_array = append(u.score_array, scoremap)
+			}
+
+			return true, nil
 		}
+	case err = <-u.errors:
+		return false, err
+	case <-proc.Ctx.Done():
+		return false, moerr.NewInternalError(proc.Ctx, "context cancelled")
+	}
 
-		for i := 0; i < bat.RowCount(); i++ {
-			// doc_id any
-			doc_id := vector.GetAny(bat.Vecs[0], i)
+	bat := res.Batches[0]
+	defer res.Close()
 
-			bytes, ok := doc_id.([]byte)
-			if ok {
-				// change it to string
-				key := string(bytes)
-				doc_id = key
-			}
+	if len(bat.Vecs) != 3 {
+		return false, moerr.NewInternalError(proc.Ctx, "output vector columns not match")
+	}
 
-			// pos int32
-			pos := vector.GetFixedAtWithTypeCheck[int32](bat.Vecs[1], i)
+	u.nrows += bat.RowCount()
 
-			// word string
-			word := bat.Vecs[2].GetStringAt(i)
+	for i := 0; i < bat.RowCount(); i++ {
+		// doc_id any
+		doc_id := vector.GetAny(bat.Vecs[0], i)
 
-			if last_doc_id == nil {
-				last_doc_id = doc_id
-			} else if last_doc_id != doc_id {
-				// new doc_id
-				last_doc_id = doc_id
-				n_doc_id++
-			}
+		bytes, ok := doc_id.([]byte)
+		if ok {
+			// change it to string
+			key := string(bytes)
+			doc_id = key
+		}
 
-			//logutil.Infof("WORD:%s, DOC_ID:%v, POS:%d, LAST: %v", word, doc_id, pos, last_doc_id)
-			if n_doc_id >= 8192 {
-				scoremap, err := s.Eval()
-				if err != nil {
-					u.errors <- err
-					return
-				}
-
-				// check context cancel
-				select {
-				case <-proc.Ctx.Done():
-					return
-				case <-u.done:
-					return
-				default:
-				}
-
-				if len(scoremap) > 0 {
-					for len(u.result_chan) == cap(u.result_chan) {
-						select {
-						case <-proc.Ctx.Done():
-							return
-						case <-u.done:
-							return
-						default:
-							time.Sleep(10 * time.Millisecond)
-						}
-					}
-					u.result_chan <- scoremap
-				}
-
-				clear(s.WordAccums)
-				n_doc_id = 0
-
-				// check limit
-				n_result += uint64(len(scoremap))
-				if u.limit > 0 && n_result >= u.limit {
-					return
-				}
-			}
+		// pos int32
+		pos := vector.GetFixedAtWithTypeCheck[int32](bat.Vecs[1], i)
 
-			w, ok := s.WordAccums[word]
-			if !ok {
-				s.WordAccums[word] = fulltext.NewWordAccum()
-				w = s.WordAccums[word]
-			}
-			_, ok = w.Words[doc_id]
-			if ok {
-				w.Words[doc_id].Position = append(w.Words[doc_id].Position, pos)
-				w.Words[doc_id].DocCount += 1
-			} else {
-				positions := make([]int32, 0, 128)
-				positions = append(positions, pos)
-				w.Words[doc_id] = &fulltext.Word{DocId: doc_id, Position: positions, DocCount: 1}
-			}
-		}
-	}
+		// word string
+		word := bat.Vecs[2].GetStringAt(i)
 
-	scoremap, err := s.Eval()
-	if err != nil {
-		u.errors <- err
-		return
-	}
+		if u.last_doc_id == nil {
+			u.last_doc_id = doc_id
+		} else if u.last_doc_id != doc_id {
+			// new doc_id
+			u.last_doc_id = doc_id
+			u.n_doc_id++
+		}
 
-	// check context cancel
-	select {
-	case <-proc.Ctx.Done():
-		return
-	case <-u.done:
-		return
-	default:
-	}
+		//logutil.Infof("WORD:%s, DOC_ID:%v, POS:%d, LAST: %v", word, doc_id, pos, last_doc_id)
+		if u.n_doc_id >= 8192 {
+			scoremap, err := s.Eval()
+			if err != nil {
+				return false, err
+			}
 
-	if len(scoremap) > 0 {
-		for len(u.result_chan) == cap(u.result_chan) {
-			select {
-			case <-proc.Ctx.Done():
-				return
-			case <-u.done:
-				return
-			default:
-				time.Sleep(10 * time.Millisecond)
+			if len(scoremap) > 0 {
+				u.score_array = append(u.score_array, scoremap)
 			}
+
+			clear(s.WordAccums)
+			u.n_doc_id = 0
+		}
+
+		w, ok := s.WordAccums[word]
+		if !ok {
+			s.WordAccums[word] = fulltext.NewWordAccum()
+			w = s.WordAccums[word]
+		}
+		_, ok = w.Words[doc_id]
+		if ok {
+			w.Words[doc_id].Position = append(w.Words[doc_id].Position, pos)
+			w.Words[doc_id].DocCount += 1
+		} else {
+			positions := make([]int32, 0, 128)
+			positions = append(positions, pos)
+			w.Words[doc_id] = &fulltext.Word{DocId: doc_id, Position: positions, DocCount: 1}
 		}
-		u.result_chan <- scoremap
 	}
+
+	return false, nil
 }
 
 // Run SQL to get number of records in source table
@@ -419,13 +432,16 @@ func fulltextIndexMatch(u *fulltextState, proc *process.Process, tableFunction *
 	}
 	s.Nrow = nrow
 
-	// get the statistic of search string ([]Pattern) and store in SearchAccum
-	res, err := runWordStats(u, proc, s)
-	if err != nil {
-		return err
-	}
+	u.sacc = s
 
-	go getResults(u, proc, s, res)
+	go func() {
+		// get the statistic of search string ([]Pattern) and store in SearchAccum
+		_, err = runWordStats(u, proc, u.sacc)
+		if err != nil {
+			u.errors <- err
+			return
+		}
+	}()
 
 	return nil
 }
diff --git a/pkg/sql/colexec/table_function/fulltext_tokenize.go b/pkg/sql/colexec/table_function/fulltext_tokenize.go
index 7491ba538..621f056c1 100644
--- a/pkg/sql/colexec/table_function/fulltext_tokenize.go
+++ b/pkg/sql/colexec/table_function/fulltext_tokenize.go
@@ -16,7 +16,6 @@ package table_function
 
 import (
 	"encoding/json"
-	"strings"
 
 	"github.com/matrixorigin/matrixone/pkg/common/datalink"
 	"github.com/matrixorigin/matrixone/pkg/common/fulltext"
@@ -96,9 +95,6 @@ func fulltextIndexTokenizePrepare(proc *process.Process, arg *TableFunction) (tv
 	arg.ctr.executorsForArgs, err = colexec.NewExpressionExecutorsFromPlanExpressions(proc, arg.Args)
 	arg.ctr.argVecs = make([]*vector.Vector, len(arg.Args))
 
-	for i := range arg.Attrs {
-		arg.Attrs[i] = strings.ToUpper(arg.Attrs[i])
-	}
 	return st, err
 
 }
diff --git a/pkg/sql/colexec/table_function/stage.go b/pkg/sql/colexec/table_function/stage.go
index 0eaa96753..e218cbd0b 100644
--- a/pkg/sql/colexec/table_function/stage.go
+++ b/pkg/sql/colexec/table_function/stage.go
@@ -38,9 +38,6 @@ func stageListPrepare(proc *process.Process, tableFunction *TableFunction) (tvfS
 	var err error
 	tableFunction.ctr.executorsForArgs, err = colexec.NewExpressionExecutorsFromPlanExpressions(proc, tableFunction.Args)
 	tableFunction.ctr.argVecs = make([]*vector.Vector, len(tableFunction.Args))
-	for i := range tableFunction.Attrs {
-		tableFunction.Attrs[i] = strings.ToUpper(tableFunction.Attrs[i])
-	}
 	return &stagelistState{}, err
 }
 
diff --git a/pkg/sql/compile/sql_executor.go b/pkg/sql/compile/sql_executor.go
index 4a7eeabe8..d24f6d247 100644
--- a/pkg/sql/compile/sql_executor.go
+++ b/pkg/sql/compile/sql_executor.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/common/buffer"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
@@ -334,6 +335,12 @@ func (exec *txnExecutor) Exec(
 	})
 
 	result := executor.NewResult(exec.s.mp)
+
+	stream_chan, streaming := exec.opts.Streaming()
+	if streaming {
+		defer close(stream_chan)
+	}
+
 	var batches []*batch.Batch
 	err = c.Compile(
 		exec.ctx,
@@ -347,9 +354,24 @@ func (exec *txnExecutor) Exec(
 				if err != nil {
 					return err
 				}
-				batches = append(batches, rows)
+				if streaming {
+					stream_result := executor.NewResult(exec.s.mp)
+					for len(stream_chan) == cap(stream_chan) {
+						select {
+						case <-proc.Ctx.Done():
+							return moerr.NewInternalError(proc.Ctx, "context cancelled")
+						default:
+							time.Sleep(1 * time.Millisecond)
+						}
+					}
+					stream_result.Batches = []*batch.Batch{rows}
+					stream_chan <- stream_result
+				} else {
+					batches = append(batches, rows)
+				}
 			}
 			return nil
+
 		})
 	if err != nil {
 		return executor.Result{}, err
diff --git a/pkg/sql/plan/build_alter_add_column.go b/pkg/sql/plan/build_alter_add_column.go
index d8d261e58..a6ca98acc 100644
--- a/pkg/sql/plan/build_alter_add_column.go
+++ b/pkg/sql/plan/build_alter_add_column.go
@@ -376,6 +376,7 @@ func handleDropColumnWithIndex(ctx context.Context, colName string, tbInfo *Tabl
 		indexInfo.Parts = RemoveIf[string](indexInfo.Parts, func(t string) bool {
 			return catalog.ResolveAlias(t) == colName
 		})
+
 		if indexInfo.Unique {
 			// handle unique index
 			if len(indexInfo.Parts) == 0 {
@@ -409,6 +410,10 @@ func handleDropColumnWithIndex(ctx context.Context, colName string, tbInfo *Tabl
 					// TODO: verify this
 					tbInfo.Indexes = append(tbInfo.Indexes[:i], tbInfo.Indexes[i+1:]...)
 				}
+			case catalog.MOIndexFullTextAlgo.ToString():
+				if len(indexInfo.Parts) == 0 {
+					tbInfo.Indexes = append(tbInfo.Indexes[:i], tbInfo.Indexes[i+1:]...)
+				}
 			}
 		}
 	}
diff --git a/pkg/util/executor/options.go b/pkg/util/executor/options.go
index 258682ec6..5bfad40d7 100644
--- a/pkg/util/executor/options.go
+++ b/pkg/util/executor/options.go
@@ -221,3 +221,13 @@ func (opts Options) LowerCaseTableNames() int64 {
 	}
 	return 1
 }
+
+func (opts Options) WithStreaming(stream_chan chan Result) Options {
+	opts.stream_chan = stream_chan
+	opts.streaming = true
+	return opts
+}
+
+func (opts Options) Streaming() (chan Result, bool) {
+	return opts.stream_chan, opts.streaming
+}
diff --git a/pkg/util/executor/types.go b/pkg/util/executor/types.go
index ef2f838b9..cb702ed27 100644
--- a/pkg/util/executor/types.go
+++ b/pkg/util/executor/types.go
@@ -62,6 +62,8 @@ type Options struct {
 	txnOpts                 []client.TxnOption
 	enableTrace             bool
 	lower                   *int64
+	streaming               bool
+	stream_chan             chan Result
 }
 
 // StatementOption statement execute option.


Test patch:
diff --git a/pkg/sql/colexec/table_function/fulltext_test.go b/pkg/sql/colexec/table_function/fulltext_test.go
new file mode 100644
index 000000000..090bd2886
--- /dev/null
+++ b/pkg/sql/colexec/table_function/fulltext_test.go
@@ -0,0 +1,279 @@
+// Copyright 2022 Matrix Origin
+//
+// Licensed 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 table_function
+
+import (
+	"testing"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/util/executor"
+	"github.com/matrixorigin/matrixone/pkg/vm"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+	"github.com/stretchr/testify/require"
+)
+
+type fulltextTestCase struct {
+	arg  *TableFunction
+	proc *process.Process
+}
+
+var (
+	ftdefaultAttrs = []string{"DOC_ID", "SCORE"}
+
+	ftdefaultColdefs = []*plan.ColDef{
+		// row_id type should be same as index type
+		{
+			Name: "DOC_ID",
+			Typ: plan.Type{
+				Id:          int32(types.T_int32),
+				NotNullable: false,
+			},
+		},
+		{
+			Name: "SCORE",
+			Typ: plan.Type{
+				Id:          int32(types.T_float32),
+				NotNullable: false,
+				Width:       4,
+			},
+		},
+	}
+)
+
+func newFTTestCase(m *mpool.MPool, attrs []string) fulltextTestCase {
+	proc := testutil.NewProcessWithMPool("", m)
+	colDefs := make([]*plan.ColDef, len(attrs))
+	for i := range attrs {
+		for j := range ftdefaultColdefs {
+			if attrs[i] == ftdefaultColdefs[j].Name {
+				colDefs[i] = ftdefaultColdefs[j]
+				break
+			}
+		}
+	}
+
+	ret := fulltextTestCase{
+		proc: proc,
+		arg: &TableFunction{
+			Attrs:    attrs,
+			Rets:     colDefs,
+			FuncName: "fulltext_index_scan",
+			OperatorBase: vm.OperatorBase{
+				OperatorInfo: vm.OperatorInfo{
+					Idx:     0,
+					IsFirst: false,
+					IsLast:  false,
+				},
+			},
+			Limit: &plan.Expr{
+				Typ: plan.Type{
+					Id: int32(types.T_uint64),
+				},
+				Expr: &plan.Expr_Lit{
+					Lit: &plan.Literal{
+						Value: &plan.Literal_U64Val{
+							U64Val: 0,
+						},
+					},
+				},
+			},
+		},
+	}
+	return ret
+}
+
+func fake_runSql(proc *process.Process, sql string) (executor.Result, error) {
+
+	// give count
+	return executor.Result{Mp: proc.Mp(), Batches: []*batch.Batch{makeCountBatchFT(proc)}}, nil
+}
+
+func fake_runSql_streaming(proc *process.Process, sql string, ch chan executor.Result) (executor.Result, error) {
+
+	defer close(ch)
+	res := executor.Result{Mp: proc.Mp(), Batches: []*batch.Batch{makeTextBatchFT(proc)}}
+	ch <- res
+	return executor.Result{}, nil
+}
+
+// argvec [src_tbl, index_tbl, pattern, mode int64]
+func TestFullTextCall(t *testing.T) {
+
+	ut := newFTTestCase(mpool.MustNewZero(), ftdefaultAttrs)
+
+	inbat := makeBatchFT(ut.proc)
+
+	ut.arg.Args = makeConstInputExprsFT()
+	//fmt.Printf("%v\n", ut.arg.Args)
+
+	// Prepare
+	err := ut.arg.Prepare(ut.proc)
+	require.Nil(t, err)
+
+	for i := range ut.arg.ctr.executorsForArgs {
+		ut.arg.ctr.argVecs[i], err = ut.arg.ctr.executorsForArgs[i].Eval(ut.proc, []*batch.Batch{inbat}, nil)
+		require.Nil(t, err)
+	}
+
+	// stub runSql function
+	ft_runSql = fake_runSql
+	ft_runSql_streaming = fake_runSql_streaming
+
+	// start
+	err = ut.arg.ctr.state.start(ut.arg, ut.proc, 0, nil)
+	require.Nil(t, err)
+
+	time.Sleep(1000 * time.Millisecond)
+
+	var result vm.CallResult
+
+	// first call receive data
+	for i := 0; i < 3; i++ {
+		result, err = ut.arg.ctr.state.call(ut.arg, ut.proc)
+		require.Nil(t, err)
+		require.Equal(t, result.Status, vm.ExecNext)
+		require.Equal(t, result.Batch.RowCount(), 8192)
+	}
+
+	result, err = ut.arg.ctr.state.call(ut.arg, ut.proc)
+	require.Nil(t, err)
+	require.Equal(t, result.Status, vm.ExecNext)
+	require.Equal(t, result.Batch.RowCount(), 1)
+	//fmt.Printf("ROW COUNT = %d  BATCH = %v\n", result.Batch.RowCount(), result.Batch)
+
+	// second call receive channel close
+	result, err = ut.arg.ctr.state.call(ut.arg, ut.proc)
+	require.Nil(t, err)
+	require.Equal(t, result.Status, vm.ExecStop)
+
+	// reset
+	ut.arg.ctr.state.reset(ut.arg, ut.proc)
+
+	// free
+	ut.arg.ctr.state.free(ut.arg, ut.proc, false, nil)
+}
+
+// create const input exprs
+func makeConstInputExprsFT() []*plan.Expr {
+
+	//ret := make([]*plan.Expr, 4)
+	ret := []*plan.Expr{{
+		Typ: plan.Type{
+			Id:    int32(types.T_varchar),
+			Width: 256,
+		},
+		Expr: &plan.Expr_Lit{
+			Lit: &plan.Literal{
+				Value: &plan.Literal_Sval{
+					Sval: "src_table",
+				},
+			},
+		},
+	}, {
+		Typ: plan.Type{
+			Id:    int32(types.T_varchar),
+			Width: 256,
+		},
+		Expr: &plan.Expr_Lit{
+			Lit: &plan.Literal{
+				Value: &plan.Literal_Sval{
+					Sval: "index_table",
+				},
+			},
+		},
+	}, {
+		Typ: plan.Type{
+			Id:    int32(types.T_varchar),
+			Width: 256,
+		},
+		Expr: &plan.Expr_Lit{
+			Lit: &plan.Literal{
+				Value: &plan.Literal_Sval{
+					Sval: "pattern",
+				},
+			},
+		},
+	}, {
+		Typ: plan.Type{
+			Id: int32(types.T_int64),
+		},
+		Expr: &plan.Expr_Lit{
+			Lit: &plan.Literal{
+				Value: &plan.Literal_I64Val{
+					I64Val: 0,
+				},
+			},
+		},
+	}}
+
+	return ret
+}
+
+// create input vector for arg (src_table, index_table, pattern, mode)
+func makeBatchFT(proc *process.Process) *batch.Batch {
+	bat := batch.NewWithSize(4)
+	bat.Vecs[0] = vector.NewVec(types.New(types.T_varchar, 256, 0))
+	bat.Vecs[1] = vector.NewVec(types.New(types.T_varchar, 256, 0))
+	bat.Vecs[2] = vector.NewVec(types.New(types.T_varchar, 256, 0))
+	bat.Vecs[3] = vector.NewVec(types.New(types.T_int32, 4, 0))
+
+	vector.AppendBytes(bat.Vecs[0], []byte("src_table"), false, proc.Mp())
+	vector.AppendBytes(bat.Vecs[1], []byte("idx_table"), false, proc.Mp())
+	vector.AppendBytes(bat.Vecs[2], []byte("pattern"), false, proc.Mp())
+	vector.AppendFixed[int64](bat.Vecs[3], int64(0), false, proc.Mp())
+
+	bat.SetRowCount(1)
+	return bat
+}
+
+// create count (int64)
+func makeCountBatchFT(proc *process.Process) *batch.Batch {
+	bat := batch.NewWithSize(1)
+	bat.Vecs[0] = vector.NewVec(types.New(types.T_int64, 8, 0))
+
+	vector.AppendFixed[int64](bat.Vecs[0], int64(100), false, proc.Mp())
+
+	bat.SetRowCount(1)
+	return bat
+}
+
+// create (doc_id, pos, text)
+func makeTextBatchFT(proc *process.Process) *batch.Batch {
+	bat := batch.NewWithSize(3)
+	bat.Vecs[0] = vector.NewVec(types.New(types.T_int32, 4, 0))     // doc_id
+	bat.Vecs[1] = vector.NewVec(types.New(types.T_int32, 4, 0))     // pos
+	bat.Vecs[2] = vector.NewVec(types.New(types.T_varchar, 256, 0)) // text
+
+	nitem := 8192*3 + 1
+	for i := 0; i < nitem; i++ {
+		// doc_id
+		vector.AppendFixed[int32](bat.Vecs[0], int32(i), false, proc.Mp())
+
+		// pos
+		vector.AppendFixed[int32](bat.Vecs[1], int32(i+1), false, proc.Mp())
+
+		// word
+		vector.AppendBytes(bat.Vecs[2], []byte("pattern"), false, proc.Mp())
+	}
+
+	bat.SetRowCount(nitem)
+	return bat
+}
diff --git a/pkg/sql/colexec/table_function/fulltext_tokenize_test.go b/pkg/sql/colexec/table_function/fulltext_tokenize_test.go
new file mode 100644
index 000000000..7b0fad980
--- /dev/null
+++ b/pkg/sql/colexec/table_function/fulltext_tokenize_test.go
@@ -0,0 +1,304 @@
+// Copyright 2022 Matrix Origin
+//
+// Licensed 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 table_function
+
+import (
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/vm"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+	"github.com/stretchr/testify/require"
+)
+
+type fulltextTokenizeTestCase struct {
+	arg  *TableFunction
+	proc *process.Process
+}
+
+var (
+	fttdefaultAttrs = []string{"DOC_ID", "POS", "WORD"}
+
+	fftdefaultColdefs = []*plan.ColDef{
+		// row_id type should be same as index type
+		{
+			Name: "DOC_ID",
+			Typ: plan.Type{
+				Id:          int32(types.T_int32),
+				NotNullable: false,
+			},
+		},
+		{
+			Name: "POS",
+			Typ: plan.Type{
+				Id:          int32(types.T_int32),
+				NotNullable: false,
+			},
+		},
+		{
+			Name: "WORD",
+			Typ: plan.Type{
+				Id:          int32(types.T_varchar),
+				NotNullable: false,
+				Width:       128,
+			},
+		},
+	}
+)
+
+func newFTTTestCase(m *mpool.MPool, attrs []string, param string) fulltextTokenizeTestCase {
+	proc := testutil.NewProcessWithMPool("", m)
+	colDefs := make([]*plan.ColDef, len(attrs))
+	for i := range attrs {
+		for j := range fftdefaultColdefs {
+			if attrs[i] == fftdefaultColdefs[j].Name {
+				colDefs[i] = fftdefaultColdefs[j]
+				break
+			}
+		}
+	}
+
+	ret := fulltextTokenizeTestCase{
+		proc: proc,
+		arg: &TableFunction{
+			Attrs:    attrs,
+			Rets:     colDefs,
+			FuncName: "fulltext_index_tokenize",
+			OperatorBase: vm.OperatorBase{
+				OperatorInfo: vm.OperatorInfo{
+					Idx:     0,
+					IsFirst: false,
+					IsLast:  false,
+				},
+			},
+			Params: []byte(param),
+		},
+	}
+	return ret
+}
+
+// argvec [src_tbl, index_tbl, pattern, mode int64]
+func TestFullTextTokenizeCall(t *testing.T) {
+
+	ut := newFTTTestCase(mpool.MustNewZero(), fttdefaultAttrs, "")
+
+	inbat := makeBatchFTT(ut.proc)
+
+	ut.arg.Args = makeConstInputExprsFTT()
+	//fmt.Printf("%v\n", ut.arg.Args)
+
+	// Prepare
+	err := ut.arg.Prepare(ut.proc)
+	require.Nil(t, err)
+
+	for i := range ut.arg.ctr.executorsForArgs {
+		ut.arg.ctr.argVecs[i], err = ut.arg.ctr.executorsForArgs[i].Eval(ut.proc, []*batch.Batch{inbat}, nil)
+		require.Nil(t, err)
+	}
+
+	// start
+	err = ut.arg.ctr.state.start(ut.arg, ut.proc, 0, nil)
+	require.Nil(t, err)
+
+	// first call receive data
+	result, err := ut.arg.ctr.state.call(ut.arg, ut.proc)
+	require.Nil(t, err)
+
+	require.Equal(t, result.Status, vm.ExecNext)
+
+	require.Equal(t, 4, result.Batch.RowCount())
+
+	// reset
+	ut.arg.ctr.state.reset(ut.arg, ut.proc)
+
+	// free
+	ut.arg.ctr.state.free(ut.arg, ut.proc, false, nil)
+}
+
+// argvec [src_tbl, index_tbl, pattern, mode int64]
+func TestFullTextTokenizeCallJSON(t *testing.T) {
+
+	ut := newFTTTestCase(mpool.MustNewZero(), fttdefaultAttrs, "{\"parser\":\"json\"}")
+
+	inbat := makeBatchJSONFTT(ut.proc)
+
+	ut.arg.Args = makeConstInputJSONExprsFTT()
+	//fmt.Printf("%v\n", ut.arg.Args)
+
+	// Prepare
+	err := ut.arg.Prepare(ut.proc)
+	require.Nil(t, err)
+
+	for i := range ut.arg.ctr.executorsForArgs {
+		ut.arg.ctr.argVecs[i], err = ut.arg.ctr.executorsForArgs[i].Eval(ut.proc, []*batch.Batch{inbat}, nil)
+		require.Nil(t, err)
+	}
+
+	// start
+	err = ut.arg.ctr.state.start(ut.arg, ut.proc, 0, nil)
+	require.Nil(t, err)
+
+	// first call receive data
+	result, err := ut.arg.ctr.state.call(ut.arg, ut.proc)
+	require.Nil(t, err)
+
+	require.Equal(t, result.Status, vm.ExecNext)
+
+	require.Equal(t, 1, result.Batch.RowCount())
+
+	// reset
+	ut.arg.ctr.state.reset(ut.arg, ut.proc)
+
+	// free
+	ut.arg.ctr.state.free(ut.arg, ut.proc, false, nil)
+}
+
+// argvec [src_tbl, index_tbl, pattern, mode int64]
+func TestFullTextTokenizeCallJSONValue(t *testing.T) {
+
+	ut := newFTTTestCase(mpool.MustNewZero(), fttdefaultAttrs, "{\"parser\":\"json_value\"}")
+
+	inbat := makeBatchJSONFTT(ut.proc)
+
+	ut.arg.Args = makeConstInputJSONExprsFTT()
+	//fmt.Printf("%v\n", ut.arg.Args)
+
+	// Prepare
+	err := ut.arg.Prepare(ut.proc)
+	require.Nil(t, err)
+
+	for i := range ut.arg.ctr.executorsForArgs {
+		ut.arg.ctr.argVecs[i], err = ut.arg.ctr.executorsForArgs[i].Eval(ut.proc, []*batch.Batch{inbat}, nil)
+		require.Nil(t, err)
+	}
+
+	// start
+	err = ut.arg.ctr.state.start(ut.arg, ut.proc, 0, nil)
+	require.Nil(t, err)
+
+	// first call receive data
+	result, err := ut.arg.ctr.state.call(ut.arg, ut.proc)
+	require.Nil(t, err)
+
+	require.Equal(t, result.Status, vm.ExecNext)
+
+	require.Equal(t, 1, result.Batch.RowCount())
+
+	// reset
+	ut.arg.ctr.state.reset(ut.arg, ut.proc)
+
+	// free
+	ut.arg.ctr.state.free(ut.arg, ut.proc, false, nil)
+}
+
+// create const input exprs
+func makeConstInputExprsFTT() []*plan.Expr {
+
+	ret := []*plan.Expr{
+		{
+			Typ: plan.Type{
+				Id: int32(types.T_int32),
+			},
+			Expr: &plan.Expr_Lit{
+				Lit: &plan.Literal{
+					Value: &plan.Literal_I32Val{
+						I32Val: 1,
+					},
+				},
+			},
+		},
+
+		{
+			Typ: plan.Type{
+				Id:    int32(types.T_varchar),
+				Width: 128,
+			},
+			Expr: &plan.Expr_Lit{
+				Lit: &plan.Literal{
+					Value: &plan.Literal_Sval{
+						Sval: "this is a text",
+					},
+				},
+			},
+		}}
+
+	return ret
+}
+
+// create input vector for arg (id, text)
+func makeBatchFTT(proc *process.Process) *batch.Batch {
+	bat := batch.NewWithSize(2)
+	bat.Vecs[0] = vector.NewVec(types.New(types.T_int32, 4, 0))
+	bat.Vecs[1] = vector.NewVec(types.New(types.T_varchar, 128, 0))
+
+	vector.AppendFixed[int32](bat.Vecs[0], int32(1), false, proc.Mp())
+	vector.AppendBytes(bat.Vecs[1], []byte("this is a text"), false, proc.Mp())
+
+	bat.SetRowCount(1)
+	return bat
+}
+
+// JSON
+// create const input exprs
+func makeConstInputJSONExprsFTT() []*plan.Expr {
+
+	ret := []*plan.Expr{
+		{
+			Typ: plan.Type{
+				Id: int32(types.T_int32),
+			},
+			Expr: &plan.Expr_Lit{
+				Lit: &plan.Literal{
+					Value: &plan.Literal_I32Val{
+						I32Val: 1,
+					},
+				},
+			},
+		},
+
+		{
+			Typ: plan.Type{
+				Id:    int32(types.T_varchar),
+				Width: 128,
+			},
+			Expr: &plan.Expr_Lit{
+				Lit: &plan.Literal{
+					Value: &plan.Literal_Sval{
+						Sval: "{\"a\":\"abcdedfghijklmnopqrstuvwxyz\"}",
+					},
+				},
+			},
+		}}
+
+	return ret
+}
+
+// create input vector for arg (id, text)
+func makeBatchJSONFTT(proc *process.Process) *batch.Batch {
+	bat := batch.NewWithSize(2)
+	bat.Vecs[0] = vector.NewVec(types.New(types.T_int32, 4, 0))
+	bat.Vecs[1] = vector.NewVec(types.New(types.T_varchar, 128, 0))
+
+	vector.AppendFixed[int32](bat.Vecs[0], int32(1), false, proc.Mp())
+	vector.AppendBytes(bat.Vecs[0], []byte("{\"a\":\"abcdedfghijklmnopqrstuvwxyz\"}"), false, proc.Mp())
+
+	bat.SetRowCount(1)
+	return bat
+}
diff --git a/test/distributed/cases/fulltext/fulltext.result b/test/distributed/cases/fulltext/fulltext.result
index 1270bdf1f..8ec4fd00c 100644
--- a/test/distributed/cases/fulltext/fulltext.result
+++ b/test/distributed/cases/fulltext/fulltext.result
@@ -466,3 +466,11 @@ delete from nation where n_nationkey = 0;
 select * from nation where match(n_comment) against('china');
 n_nationkey    n_name    n_regionkey    n_comment    n_dummy
 drop table nation;
+drop table if exists articles;
+create table articles (id int auto_increment not null primary key, title varchar, body text);
+create fulltext index ftidx on articles (title, body);
+alter table articles drop column title;
+drop table if exists articles;
+create table articles (id int auto_increment not null primary key, title varchar, body text);
+create fulltext index ftidx on articles (title);
+alter table articles drop column title;
diff --git a/test/distributed/cases/fulltext/fulltext.sql b/test/distributed/cases/fulltext/fulltext.sql
index 27790219c..fa8203e90 100644
--- a/test/distributed/cases/fulltext/fulltext.sql
+++ b/test/distributed/cases/fulltext/fulltext.sql
@@ -377,3 +377,15 @@ select * from nation where match(n_comment) against('china');
 
 drop table nation;
 
+
+-- error drop column with multiple index parts
+drop table if exists articles;
+create table articles (id int auto_increment not null primary key, title varchar, body text);
+create fulltext index ftidx on articles (title, body);
+alter table articles drop column title;
+
+-- drop column success
+drop table if exists articles;
+create table articles (id int auto_increment not null primary key, title varchar, body text);
+create fulltext index ftidx on articles (title);
+alter table articles drop column title;