Fix mo stream bugs (matrixorigin#13117)

1. Fix the data conversion bugs 2. Fix drop source bug 3. Fix options bugs when creating dt with options Approved by: @iamlinjunhong, @daviszhen, @ouyuanning, @nnsgmsone, @sukki37
nojomohr · Dec 4, 2023 · 87b1ce9 · 87b1ce9
1 parent fd9cf7a
commit 87b1ce9
Show file tree

Hide file tree

Showing 13 changed files with 555 additions and 116 deletions.
diff --git a/pkg/frontend/connector.go b/pkg/frontend/connector.go
@@ -48,9 +48,17 @@ func (mce *MysqlCmdExecutor) handleCreateDynamicTable(ctx context.Context, st *t
 		return moerr.NewNoSuchTable(ctx, dbName, tableName)
 	}
 	options := make(map[string]string)
+	for _, option := range st.DTOptions {
+		switch opt := option.(type) {
+		case *tree.CreateSourceWithOption:
+			key := string(opt.Key)
+			val := opt.Val.(*tree.NumVal).OrigString()
+			options[key] = val
+		}
+	}
+
 	ses := mce.GetSession()
 
-	//get query optimizer and execute Optimize
 	generatedPlan, err := buildPlan(ctx, ses, ses.GetTxnCompileCtx(), st.AsSource)
 	if err != nil {
 		return err

diff --git a/pkg/sql/compile/ddl.go b/pkg/sql/compile/ddl.go
@@ -1427,7 +1427,10 @@ func (s *Scope) DropTable(c *Compile) error {
 	dbName := qry.GetDatabase()
 	tblName := qry.GetTable()
 	isView := qry.GetIsView()
-
+	var isSource = false
+	if qry.TableDef != nil {
+		isSource = qry.TableDef.TableType == catalog.SystemSourceRel
+	}
 	var dbSource engine.Database
 	var rel engine.Relation
 	var err error
@@ -1462,7 +1465,7 @@ func (s *Scope) DropTable(c *Compile) error {
 		isTemp = true
 	}
 
-	if !isTemp && !isView && c.proc.TxnOperator.Txn().IsPessimistic() {
+	if !isTemp && !isView && !isSource && c.proc.TxnOperator.Txn().IsPessimistic() {
 		var err error
 		if e := lockMoTable(c, dbName, tblName, lock.LockMode_Exclusive); e != nil {
 			if !moerr.IsMoErrCode(e, moerr.ErrTxnNeedRetry) &&

diff --git a/pkg/sql/parsers/dialect/mysql/mysql_sql.go b/pkg/sql/parsers/dialect/mysql/mysql_sql.go
diff --git a/pkg/sql/parsers/dialect/mysql/mysql_sql.y b/pkg/sql/parsers/dialect/mysql/mysql_sql.y
@@ -6442,7 +6442,7 @@ create_table_stmt:
             IfNotExists: $4,
             Table: *$5,
             AsSource: $7,
-            Options: $8,
+            DTOptions: $8,
         }
     }
 load_param_opt_2:

diff --git a/pkg/sql/parsers/dialect/mysql/mysql_sql_test.go b/pkg/sql/parsers/dialect/mysql/mysql_sql_test.go
@@ -731,6 +731,9 @@ var (
 	}, {
 		input:  "create dynamic table t as select a from t1",
 		output: "create dynamic table t as select a from t1",
+	}, {
+		input:  "create dynamic table t as select a from t1 with (\"type\"='kafka')",
+		output: "create dynamic table t as select a from t1 with (type = kafka)",
 	}, {
 		input:  "create external table t (a int) infile 'data.txt'",
 		output: "create external table t (a int) infile 'data.txt'",

diff --git a/pkg/sql/parsers/tree/create.go b/pkg/sql/parsers/tree/create.go
@@ -156,6 +156,7 @@ type CreateTable struct {
 	Param           *ExternParam
 	AsSource        *Select
 	IsDynamicTable  bool
+	DTOptions       []TableOption
 }
 
 func (node *CreateTable) Format(ctx *FmtCtx) {
@@ -185,6 +186,16 @@ func (node *CreateTable) Format(ctx *FmtCtx) {
 	if node.IsDynamicTable {
 		ctx.WriteString(" as ")
 		node.AsSource.Format(ctx)
+
+		if node.DTOptions != nil {
+			prefix := " with ("
+			for _, t := range node.DTOptions {
+				ctx.WriteString(prefix)
+				t.Format(ctx)
+				prefix = ", "
+			}
+			ctx.WriteByte(')')
+		}
 	} else {
 
 		ctx.WriteString(" (")
@@ -198,7 +209,7 @@ func (node *CreateTable) Format(ctx *FmtCtx) {
 		ctx.WriteByte(')')
 	}
 
-	if node.Options != nil {
+	if node.Options != nil && !node.IsDynamicTable {
 		prefix := " "
 		for _, t := range node.Options {
 			ctx.WriteString(prefix)

diff --git a/pkg/sql/plan/explain/explain_node.go b/pkg/sql/plan/explain/explain_node.go
@@ -59,7 +59,7 @@ func (ndesc *NodeDescribeImpl) GetNodeBasicInfo(ctx context.Context, options *Ex
 	case plan.Node_EXTERNAL_SCAN:
 		pname = ExternalScan
 	case plan.Node_SOURCE_SCAN:
-		pname = "Stream Scan"
+		pname = "Source Scan"
 	case plan.Node_MATERIAL_SCAN:
 		pname = "Material Scan"
 	case plan.Node_PROJECT: