Add LOAD INTO VALUES (#11633)

Provide a quick way to insert data,  example is as follows:
```sql
load data  inline format='csv', data='1\n2\n' into table t1;

load data  inline format='csv', data=$XXX$
1
2
$XXX$
into table t1;
```

Approved by: @ouyuanning, @iamlinjunhong, @m-schen, @aunjgr, @heni02, @fengttt

pkg/sql/colexec/external/external.go

@@ -36,6 +36,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/common/util"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/bytejson"
 	"github.com/matrixorigin/matrixone/pkg/container/nulls"
@@ -97,7 +98,7 @@ func Prepare(proc *process.Process, arg any) error {
 	}
 	param.IgnoreLineTag = int(param.Extern.Tail.IgnoredLines)
 	param.IgnoreLine = param.IgnoreLineTag
-	if len(param.FileList) == 0 {
+	if len(param.FileList) == 0 && param.Extern.ScanType != tree.INLINE {
 		logutil.Warnf("no such file '%s'", param.Extern.Filepath)
 		param.Fileparam.End = true
 	}
@@ -139,7 +140,7 @@ func Call(idx int, proc *process.Process, arg any, isFirst bool, isLast bool) (p
 		proc.SetInputBatch(nil)
 		return process.ExecStop, nil
 	}
-	if param.plh == nil {
+	if param.plh == nil && param.Extern.ScanType != tree.INLINE {
 		if param.Fileparam.FileIndex >= len(param.FileList) {
 			proc.SetInputBatch(nil)
 			return process.ExecStop, nil
@@ -282,6 +283,9 @@ func FilterFileList(ctx context.Context, node *plan.Node, proc *process.Process,
 }
 
 func readFile(param *ExternalParam, proc *process.Process) (io.ReadCloser, error) {
+	if param.Extern.ScanType == tree.INLINE {
+		return io.NopCloser(bytes.NewReader(util.UnsafeStringToBytes(param.Extern.Data))), nil
+	}
 	if param.Extern.Local {
 		return io.NopCloser(proc.LoadLocalReader), nil
 	}

pkg/sql/compile/compile.go

@@ -29,6 +29,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	mokafka "github.com/matrixorigin/matrixone/pkg/stream/adapter/kafka"
+
 	"github.com/google/uuid"
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/cnservice/cnclient"
@@ -1397,9 +1399,21 @@ func (c *Compile) compileExternScan(ctx context.Context, n *plan.Node) ([]*Scope
 		ID2Addr[i] = mcpu - tmp
 	}
 	param := &tree.ExternParam{}
-	err := json.Unmarshal([]byte(n.TableDef.Createsql), param)
-	if err != nil {
-		return nil, err
+	if n.ExternScan == nil || n.ExternScan.Type != tree.INLINE {
+		err := json.Unmarshal([]byte(n.TableDef.Createsql), param)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		param.ScanType = int(n.ExternScan.Type)
+		param.Data = n.ExternScan.Data
+		param.Format = n.ExternScan.Format
+		param.Tail = new(tree.TailParameter)
+		param.Tail.IgnoredLines = n.ExternScan.IgnoredLines
+		param.Tail.Fields = &tree.Fields{
+			Terminated: n.ExternScan.Terminated,
+			EnclosedBy: n.ExternScan.EnclosedBy[0],
+		}
 	}
 	if param.ScanType == tree.S3 {
 		if err := plan2.InitS3Param(param); err != nil {
@@ -1418,6 +1432,8 @@ func (c *Compile) compileExternScan(ctx context.Context, n *plan.Node) ([]*Scope
 				ID2Addr[i] = mcpu - tmp
 			}
 		}
+	} else if param.ScanType == tree.INLINE {
+		return c.compileExternValueScan(n, param)
 	} else {
 		if err := plan2.InitInfileParam(param); err != nil {
 			return nil, err
@@ -1426,6 +1442,7 @@ func (c *Compile) compileExternScan(ctx context.Context, n *plan.Node) ([]*Scope
 
 	param.FileService = c.proc.FileService
 	param.Ctx = c.ctx
+	var err error
 	var fileList []string
 	var fileSize []int64
 	if !param.Local {
@@ -1463,7 +1480,6 @@ func (c *Compile) compileExternScan(ctx context.Context, n *plan.Node) ([]*Scope
 
 		return []*Scope{ret}, nil
 	}
-
 	if param.Parallel && (external.GetCompressType(param, fileList[0]) != tree.NOCOMPRESS || param.Local) {
 		return c.compileExternScanParallel(n, param, fileList, fileSize)
 	}
@@ -1511,6 +1527,29 @@ func (c *Compile) compileExternScan(ctx context.Context, n *plan.Node) ([]*Scope
 	return ss, nil
 }
 
+func (c *Compile) compileExternValueScan(n *plan.Node, param *tree.ExternParam) ([]*Scope, error) {
+	ss := make([]*Scope, ncpu)
+	for i := 0; i < ncpu; i++ {
+		ss[i] = c.constructLoadMergeScope()
+	}
+	s := c.constructScopeForExternal(c.addr, false)
+	s.appendInstruction(vm.Instruction{
+		Op:      vm.External,
+		Idx:     c.anal.curr,
+		IsFirst: c.anal.isFirst,
+		Arg:     constructExternal(n, param, c.ctx, nil, nil, nil),
+	})
+	_, arg := constructDispatchLocalAndRemote(0, ss, c.addr)
+	arg.FuncId = dispatch.SendToAnyLocalFunc
+	s.appendInstruction(vm.Instruction{
+		Op:  vm.Dispatch,
+		Arg: arg,
+	})
+	ss[0].PreScopes = append(ss[0].PreScopes, s)
+	c.anal.isFirst = false
+	return ss, nil
+}
+
 // construct one thread to read the file data, then dispatch to mcpu thread to get the filedata for insert
 func (c *Compile) compileExternScanParallel(n *plan.Node, param *tree.ExternParam, fileList []string, fileSize []int64) ([]*Scope, error) {
 	param.Parallel = false

pkg/sql/parsers/dialect/mysql/keywords.go

@@ -210,6 +210,7 @@ func init() {
 		"in":                         IN,
 		"index":                      INDEX,
 		"indexes":                    INDEXES,
+		"inline":                     INLINE,
 		"infile":                     INFILE,
 		"inout":                      INOUT,
 		"inner":                      INNER,