Merge pull request #357 from illia-li/il/fix-gen_udt_columns

fix(UDTType) added JSON processing
scylladb · Jun 15, 2023 · 2a1bb3b · 2a1bb3b
2 parents ad3f856 + 789d5cb
commit 2a1bb3b
Show file tree

Hide file tree

Showing 21 changed files with 142 additions and 75 deletions.
diff --git a/cmd/gemini/schema.json b/cmd/gemini/schema.json
@@ -29,14 +29,14 @@
         {
           "name": "col0",
           "type": {
-            "types": {
+            "type_name": "udt_672245080",
+            "frozen": true,
+            "coltypes": {
               "udt_672245080_0": "ascii",
               "udt_672245080_1": "boolean",
               "udt_672245080_2": "bigint",
               "udt_672245080_3": "blob"
-            },
-            "type_name": "udt_672245080",
-            "frozen": true
+            }
           }
         },
         {

diff --git a/pkg/joberror/joberror.go b/pkg/joberror/joberror.go
@@ -25,6 +25,7 @@ type JobError struct {
 	Timestamp time.Time `json:"timestamp"`
 	Message   string    `json:"message"`
 	Query     string    `json:"query"`
+	StmtType  string    `json:"stmt-type"`
 }
 
 type ErrorList struct {

diff --git a/pkg/joberror/joberror_test.go b/pkg/joberror/joberror_test.go
@@ -69,11 +69,12 @@ func TestParallel(t *testing.T) {
 }
 
 func TestErrorSerialization(t *testing.T) {
-	expected := []byte(`{"timestamp":"2020-02-01T00:00:00Z","message":"Some Message","query":"Some Query"}`)
+	expected := []byte(`{"timestamp":"2020-02-01T00:00:00Z","message":"Some Message","query":"Some Query","stmt-type":"Some Type"}`)
 	result, err := json.Marshal(joberror.JobError{
 		Timestamp: time.Date(2020, 02, 01, 0, 0, 0, 0, time.UTC),
 		Message:   "Some Message",
 		Query:     "Some Query",
+		StmtType:  "Some Type",
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -86,12 +87,13 @@ func TestErrorSerialization(t *testing.T) {
 
 func TestErrorListSerialization(t *testing.T) {
 	//nolint:lll
-	expected := []byte(`[{"timestamp":"2020-02-01T00:00:00Z","message":"Some Message 0","query":"Some Query 0"},{"timestamp":"2020-02-02T00:00:00Z","message":"Some Message 1","query":"Some Query 1"},{"timestamp":"2020-02-03T00:00:00Z","message":"Some Message 2","query":"Some Query 2"},{"timestamp":"2020-02-04T00:00:00Z","message":"Some Message 3","query":"Some Query 3"},{"timestamp":"2020-02-05T00:00:00Z","message":"Some Message 4","query":"Some Query 4"},{"timestamp":"2020-02-06T00:00:00Z","message":"Some Message 5","query":"Some Query 5"},{"timestamp":"2020-02-07T00:00:00Z","message":"Some Message 6","query":"Some Query 6"},{"timestamp":"2020-02-08T00:00:00Z","message":"Some Message 7","query":"Some Query 7"},{"timestamp":"2020-02-09T00:00:00Z","message":"Some Message 8","query":"Some Query 8"},{"timestamp":"2020-02-10T00:00:00Z","message":"Some Message 9","query":"Some Query 9"}]`)
+	expected := []byte(`[{"timestamp":"2020-02-01T00:00:00Z","message":"Some Message 0","query":"Some Query 0","stmt-type":"Some Stmt Type 0"},{"timestamp":"2020-02-02T00:00:00Z","message":"Some Message 1","query":"Some Query 1","stmt-type":"Some Stmt Type 1"},{"timestamp":"2020-02-03T00:00:00Z","message":"Some Message 2","query":"Some Query 2","stmt-type":"Some Stmt Type 2"},{"timestamp":"2020-02-04T00:00:00Z","message":"Some Message 3","query":"Some Query 3","stmt-type":"Some Stmt Type 3"},{"timestamp":"2020-02-05T00:00:00Z","message":"Some Message 4","query":"Some Query 4","stmt-type":"Some Stmt Type 4"},{"timestamp":"2020-02-06T00:00:00Z","message":"Some Message 5","query":"Some Query 5","stmt-type":"Some Stmt Type 5"},{"timestamp":"2020-02-07T00:00:00Z","message":"Some Message 6","query":"Some Query 6","stmt-type":"Some Stmt Type 6"},{"timestamp":"2020-02-08T00:00:00Z","message":"Some Message 7","query":"Some Query 7","stmt-type":"Some Stmt Type 7"},{"timestamp":"2020-02-09T00:00:00Z","message":"Some Message 8","query":"Some Query 8","stmt-type":"Some Stmt Type 8"},{"timestamp":"2020-02-10T00:00:00Z","message":"Some Message 9","query":"Some Query 9","stmt-type":"Some Stmt Type 9"}]`)
 	lst := joberror.NewErrorList(1000)
 	baseDate := time.Date(2020, 02, 01, 0, 0, 0, 0, time.UTC)
 	for y := 0; y < 10; y++ {
 		lst.AddError(&joberror.JobError{
 			Timestamp: baseDate.AddDate(0, 0, y),
+			StmtType:  "Some Stmt Type " + strconv.Itoa(y),
 			Message:   "Some Message " + strconv.Itoa(y),
 			Query:     "Some Query " + strconv.Itoa(y),
 		})

diff --git a/pkg/jobs/gen_ddl_stmt.go b/pkg/jobs/gen_ddl_stmt.go
@@ -73,7 +73,8 @@ func genAddColumnStmt(t *typedef.Table, keyspace string, column *typedef.ColumnD
 		},
 	})
 	return &typedef.Stmts{
-		List: stmts,
+		List:      stmts,
+		QueryType: typedef.AddColumnStatementType,
 		PostStmtHook: func() {
 			t.Columns = append(t.Columns, column)
 			t.ResetQueryCache()
@@ -124,7 +125,8 @@ func genDropColumnStmt(t *typedef.Table, keyspace string, column *typedef.Column
 		},
 	})
 	return &typedef.Stmts{
-		List: stmts,
+		List:      stmts,
+		QueryType: typedef.DropColumnStatementType,
 		PostStmtHook: func() {
 			t.Columns = t.Columns.Remove(column)
 			t.ResetQueryCache()

diff --git a/pkg/jobs/gen_mutate_stmt.go b/pkg/jobs/gen_mutate_stmt.go
@@ -174,7 +174,7 @@ func genInsertJSONStmt(
 		StmtCache: &typedef.StmtCache{
 			Query:     builder,
 			Types:     []typedef.Type{typedef.TYPE_TEXT},
-			QueryType: typedef.InsertStatement,
+			QueryType: typedef.InsertJSONStatementType,
 		},
 		ValuesWithToken: valuesWithToken,
 		Values:          []interface{}{string(jsonString)},

diff --git a/pkg/jobs/jobs.go b/pkg/jobs/jobs.go
@@ -243,6 +243,7 @@ func validationJob(
 			if err := validation(ctx, schemaConfig, table, s, stmt, g, globalStatus, logger); err != nil {
 				globalStatus.AddReadError(&joberror.JobError{
 					Timestamp: time.Now(),
+					StmtType:  stmt.QueryType.ToString(),
 					Message:   "Validation failed: " + err.Error(),
 					Query:     stmt.PrettyCQL(),
 				})
@@ -341,6 +342,7 @@ func ddl(
 		if err = s.Mutate(ctx, ddlStmt.Query); err != nil {
 			globalStatus.AddWriteError(&joberror.JobError{
 				Timestamp: time.Now(),
+				StmtType:  ddlStmts.QueryType.ToString(),
 				Message:   "DDL failed: " + err.Error(),
 				Query:     ddlStmt.PrettyCQL(),
 			})
@@ -394,6 +396,7 @@ func mutation(
 	if err = s.Mutate(ctx, mutateQuery, mutateValues...); err != nil {
 		globalStatus.AddWriteError(&joberror.JobError{
 			Timestamp: time.Now(),
+			StmtType:  mutateStmt.QueryType.ToString(),
 			Message:   "Mutation failed: " + err.Error(),
 			Query:     mutateStmt.PrettyCQL(),
 		})

diff --git a/pkg/jobs/test_expected_data/ddl/drop_column.json b/pkg/jobs/test_expected_data/ddl/drop_column.json
@@ -7,7 +7,7 @@
       "Names": "[]",
       "Values": "[]",
       "Types": "",
-      "QueryType": "8"
+      "QueryType": "9"
     }
   ],
   "pkAll_ckAll_colAll_delLast": [
@@ -18,7 +18,7 @@
       "Names": "[]",
       "Values": "[]",
       "Types": "",
-      "QueryType": "8"
+      "QueryType": "9"
     }
   ]
 }
diff --git a/pkg/jobs/test_expected_data/mutate/insert_j.json b/pkg/jobs/test_expected_data/mutate/insert_j.json
@@ -7,7 +7,7 @@
       "Names": "[]",
       "Values": "[{\"pk0\":1}]",
       "Types": " text",
-      "QueryType": "5"
+      "QueryType": "6"
     }
   ],
   "pk1_ck1_col1": [
@@ -18,7 +18,7 @@
       "Names": "[]",
       "Values": "[{\"ck0\":\"1969-12-31\",\"col0\":\"1969-12-31\",\"pk0\":1}]",
       "Types": " text",
-      "QueryType": "5"
+      "QueryType": "6"
     }
   ],
   "pk3_ck3_col5": [
@@ -29,7 +29,7 @@
       "Names": "[]",
       "Values": "[{\"ck0\":\"01\",\"ck1\":\"1969-12-31\",\"ck2\":\"0.001\",\"col0\":\"00\",\"col1\":\"1969-12-31\",\"col2\":\"0x3030\",\"col3\":1,\"col4\":1.110223e-16,\"pk0\":1,\"pk1\":1.110223e-16,\"pk2\":\"1.1.1.1\"}]",
       "Types": " text",
-      "QueryType": "5"
+      "QueryType": "6"
     }
   ],
   "pkAll_ckAll_colAll": [
@@ -40,7 +40,7 @@
       "Names": "[]",
       "Values": "[{\"ck0\":\"01\",\"ck1\":1,\"ck10\":0,\"ck11\":\"00\",\"ck12\":\"1969-12-31T00:00:01Z\",\"ck13\":\"00000001-0000-1000-8000-3132372e302e\",\"ck14\":0,\"ck15\":\"00000001-0000-1000-8000-3132372e302e\",\"ck16\":\"00\",\"ck17\":1,\"ck18\":-86399000000000,\"ck2\":\"0x3030\",\"ck3\":false,\"ck4\":\"1969-12-31\",\"ck5\":\"0.001\",\"ck6\":1.1102230246251565e-16,\"ck7\":1.110223e-16,\"ck8\":\"1.1.1.1\",\"ck9\":0,\"col0\":\"1m0s\",\"col1\":\"01\",\"col10\":0,\"col11\":0,\"col12\":\"00\",\"col13\":\"1969-12-31T00:00:01Z\",\"col14\":\"00000001-0000-1000-8000-3132372e302e\",\"col15\":0,\"col16\":\"00000001-0000-1000-8000-3132372e302e\",\"col17\":\"00\",\"col18\":1,\"col19\":-86399000000000,\"col2\":1,\"col3\":\"0x3030\",\"col4\":false,\"col5\":\"1969-12-31\",\"col6\":\"0.001\",\"col7\":1.1102230246251565e-16,\"col8\":1.110223e-16,\"col9\":\"1.1.1.1\",\"pk0\":\"01\",\"pk1\":1,\"pk10\":0,\"pk11\":\"00\",\"pk12\":\"1969-12-31T00:00:01Z\",\"pk13\":\"00000001-0000-1000-8000-3132372e302e\",\"pk14\":0,\"pk15\":\"00000001-0000-1000-8000-3132372e302e\",\"pk16\":\"00\",\"pk17\":1,\"pk18\":-86399000000000,\"pk2\":\"0x3030\",\"pk3\":false,\"pk4\":\"1969-12-31\",\"pk5\":\"0.001\",\"pk6\":1.1102230246251565e-16,\"pk7\":1.110223e-16,\"pk8\":\"1.1.1.1\",\"pk9\":0}]",
       "Types": " text",
-      "QueryType": "5"
+      "QueryType": "6"
     }
   ]
 }
diff --git a/pkg/jobs/test_expected_data/mutate/update.json b/pkg/jobs/test_expected_data/mutate/update.json
@@ -7,7 +7,7 @@
       "Names": "[pk0]",
       "Values": "[1]",
       "Types": " bigint",
-      "QueryType": "6"
+      "QueryType": "7"
     }
   ],
   "pk1_ck1_col1": [
@@ -18,7 +18,7 @@
       "Names": "[col0 pk0 ck0]",
       "Values": "[1969-12-31 1 1969-12-31]",
       "Types": " date bigint date",
-      "QueryType": "6"
+      "QueryType": "7"
     }
   ],
   "pk1_ck1_col1cr": [
@@ -29,7 +29,7 @@
       "Names": "[pk0 ck0]",
       "Values": "[1 1969-12-31]",
       "Types": " bigint date",
-      "QueryType": "6"
+      "QueryType": "7"
     }
   ],
   "pk3_ck3_col3cr": [
@@ -40,7 +40,7 @@
       "Names": "[pk0 pk1 pk2 ck0 ck1 ck2]",
       "Values": "[1 1.110223e-16 1.1.1.1 01 1969-12-31 0.001]",
       "Types": " bigint float inet ascii date decimal",
-      "QueryType": "6"
+      "QueryType": "7"
     }
   ],
   "pk3_ck3_col5": [
@@ -51,7 +51,7 @@
       "Names": "[col0 col1 col2 col3 col4 pk0 pk1 pk2 ck0 ck1 ck2]",
       "Values": "[01 1969-12-31 3030 1 1.110223e-16 1 1.110223e-16 1.1.1.1 00 1969-12-31 0.001]",
       "Types": " ascii date blob bigint float bigint float inet ascii date decimal",
-      "QueryType": "6"
+      "QueryType": "7"
     }
   ],
   "pkAll_ckAll_colAll": [
@@ -62,7 +62,7 @@
       "Names": "[col0 col1 col2 col3 col4 col5 col6 col7 col8 col9 col10 col11 col12 col13 col14 col15 col16 col17 col18 col19 pk0 pk1 pk2 pk3 pk4 pk5 pk6 pk7 pk8 pk9 pk10 pk11 pk12 pk13 pk14 pk15 pk16 pk17 pk18 ck0 ck1 ck2 ck3 ck4 ck5 ck6 ck7 ck8 ck9 ck10 ck11 ck12 ck13 ck14 ck15 ck16 ck17 ck18]",
       "Values": "[1m0s 01 1 3030 false 1969-12-31 0.001 1.1102230246251565e-16 1.110223e-16 1.1.1.1 0 0 00 1969-12-31 00:00:01 +0000 UTC 00000001-0000-1000-8000-3132372e302e 0 00000001-0000-1000-8000-3132372e302e 00 1 -86399000000000 01 1 3030 false 1969-12-31 0.001 1.1102230246251565e-16 1.110223e-16 1.1.1.1 0 0 00 1969-12-31 00:00:01 +0000 UTC 00000001-0000-1000-8000-3132372e302e 0 00000001-0000-1000-8000-3132372e302e 00 1 -86399000000000 01 1 3030 false 1969-12-31 0.001 1.1102230246251565e-16 1.110223e-16 1.1.1.1 0 0 00 1969-12-31 00:00:01 +0000 UTC 00000001-0000-1000-8000-3132372e302e 0 00000001-0000-1000-8000-3132372e302e 00 1 -86399000000000]",
       "Types": " duration ascii bigint blob boolean date decimal double float inet int smallint text timestamp timeuuid tinyint uuid varchar varint time ascii bigint blob boolean date decimal double float inet int smallint text timestamp timeuuid tinyint uuid varchar varint time ascii bigint blob boolean date decimal double float inet int smallint text timestamp timeuuid tinyint uuid varchar varint time",
-      "QueryType": "6"
+      "QueryType": "7"
     }
   ]
 }
diff --git a/pkg/querycache/querycache.go b/pkg/querycache/querycache.go
@@ -123,7 +123,7 @@ func genInsertStmtCache(
 	return &typedef.StmtCache{
 		Query:     builder,
 		Types:     allTypes,
-		QueryType: typedef.InsertStatement,
+		QueryType: typedef.InsertStatementType,
 	}
 }
 
@@ -164,7 +164,7 @@ func genUpdateStmtCache(s *typedef.Schema, t *typedef.Table) *typedef.StmtCache
 	return &typedef.StmtCache{
 		Query:     builder,
 		Types:     allTypes,
-		QueryType: typedef.Updatetatement,
+		QueryType: typedef.UpdateStatementType,
 	}
 }
 

diff --git a/pkg/status/status_test.go b/pkg/status/status_test.go
@@ -29,7 +29,7 @@ import (
 func TestSerialization(t *testing.T) {
 	t.Parallel()
 	//nolint:lll
-	expected := []byte(`{"errors":[{"timestamp":"2020-02-01T00:00:00Z","message":"Some Message 0","query":"Some Query 0"},{"timestamp":"2020-02-02T00:00:00Z","message":"Some Message 1","query":"Some Query 1"},{"timestamp":"2020-02-03T00:00:00Z","message":"Some Message 2","query":"Some Query 2"},{"timestamp":"2020-02-04T00:00:00Z","message":"Some Message 3","query":"Some Query 3"},{"timestamp":"2020-02-05T00:00:00Z","message":"Some Message 4","query":"Some Query 4"},{"timestamp":"2020-03-01T00:00:00Z","message":"Some Message 0","query":"Some Query 0"},{"timestamp":"2020-03-02T00:00:00Z","message":"Some Message 1","query":"Some Query 1"},{"timestamp":"2020-03-03T00:00:00Z","message":"Some Message 2","query":"Some Query 2"},{"timestamp":"2020-03-04T00:00:00Z","message":"Some Message 3","query":"Some Query 3"},{"timestamp":"2020-03-05T00:00:00Z","message":"Some Message 4","query":"Some Query 4"}],"write_ops":10,"write_errors":5,"read_ops":5,"read_errors":5}`)
+	expected := []byte(`{"errors":[{"timestamp":"2020-02-01T00:00:00Z","message":"Some Message 0","query":"Some Query 0","stmt-type":"Some Query Type 0"},{"timestamp":"2020-02-02T00:00:00Z","message":"Some Message 1","query":"Some Query 1","stmt-type":"Some Query Type 1"},{"timestamp":"2020-02-03T00:00:00Z","message":"Some Message 2","query":"Some Query 2","stmt-type":"Some Query Type 2"},{"timestamp":"2020-02-04T00:00:00Z","message":"Some Message 3","query":"Some Query 3","stmt-type":"Some Query Type 3"},{"timestamp":"2020-02-05T00:00:00Z","message":"Some Message 4","query":"Some Query 4","stmt-type":"Some Query Type 4"},{"timestamp":"2020-03-01T00:00:00Z","message":"Some Message 0","query":"Some Query 0","stmt-type":"Some Query Type 0"},{"timestamp":"2020-03-02T00:00:00Z","message":"Some Message 1","query":"Some Query 1","stmt-type":"Some Query Type 1"},{"timestamp":"2020-03-03T00:00:00Z","message":"Some Message 2","query":"Some Query 2","stmt-type":"Some Query Type 2"},{"timestamp":"2020-03-04T00:00:00Z","message":"Some Message 3","query":"Some Query 3","stmt-type":"Some Query Type 3"},{"timestamp":"2020-03-05T00:00:00Z","message":"Some Message 4","query":"Some Query 4","stmt-type":"Some Query Type 4"}],"write_ops":10,"write_errors":5,"read_ops":5,"read_errors":5}`)
 	st := status.NewGlobalStatus(10)
 	st.WriteOps.Store(10)
 	st.ReadOps.Store(5)
@@ -38,6 +38,7 @@ func TestSerialization(t *testing.T) {
 	for y := 0; y < 5; y++ {
 		st.AddReadError(&joberror.JobError{
 			Timestamp: baseDate.AddDate(0, 0, y),
+			StmtType:  "Some Query Type " + strconv.Itoa(y),
 			Message:   "Some Message " + strconv.Itoa(y),
 			Query:     "Some Query " + strconv.Itoa(y),
 		})
@@ -47,6 +48,7 @@ func TestSerialization(t *testing.T) {
 	for y := 0; y < 5; y++ {
 		st.AddWriteError(&joberror.JobError{
 			Timestamp: baseDate.AddDate(0, 0, y),
+			StmtType:  "Some Query Type " + strconv.Itoa(y),
 			Message:   "Some Message " + strconv.Itoa(y),
 			Query:     "Some Query " + strconv.Itoa(y),
 		})

diff --git a/pkg/store/cqlstore.go b/pkg/store/cqlstore.go
@@ -140,8 +140,8 @@ func ignore(err error) bool {
 		return true
 	}
 	//nolint:errorlint
-	switch err {
-	case context.Canceled, context.DeadlineExceeded:
+	switch {
+	case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
 		return true
 	default:
 		return false

diff --git a/pkg/typedef/bag.go b/pkg/typedef/bag.go
@@ -87,6 +87,15 @@ func (ct *BagType) GenValue(r *rand.Rand, p *PartitionRangeConfig) []interface{}
 	return []interface{}{out}
 }
 
+func (ct *BagType) GenJSONValue(r *rand.Rand, p *PartitionRangeConfig) interface{} {
+	count := r.Intn(9) + 1
+	out := make([]interface{}, count)
+	for i := 0; i < count; i++ {
+		out[i] = ct.Type.GenJSONValue(r, p)
+	}
+	return out
+}
+
 func (ct *BagType) LenValue() int {
 	return 1
 }

diff --git a/pkg/typedef/columns.go b/pkg/typedef/columns.go
@@ -16,7 +16,6 @@ package typedef
 
 import (
 	"encoding/json"
-	"fmt"
 
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
@@ -96,31 +95,7 @@ func (c Columns) Remove(column *ColumnDef) Columns {
 
 func (c Columns) ToJSONMap(values map[string]interface{}, r *rand.Rand, p *PartitionRangeConfig) map[string]interface{} {
 	for _, k := range c {
-		switch t := k.Type.(type) {
-		case SimpleType:
-			if t != TYPE_BLOB {
-				values[k.Name] = t.GenValue(r, p)[0]
-				continue
-			}
-			v, ok := t.GenValue(r, p)[0].(string)
-			if ok {
-				values[k.Name] = "0x" + v
-			}
-		case *TupleType:
-			vv := t.GenValue(r, p)
-			for i, val := range vv {
-				if t.Types[i] == TYPE_BLOB {
-					v, ok := val.(string)
-					if ok {
-						v = "0x" + v
-					}
-					vv[i] = v
-				}
-			}
-			values[k.Name] = vv
-		default:
-			panic(fmt.Sprintf("unknown type: %s", t.Name()))
-		}
+		values[k.Name] = k.Type.GenJSONValue(r, p)
 	}
 	return values
 }

diff --git a/pkg/typedef/const.go b/pkg/typedef/const.go
@@ -30,10 +30,12 @@ const (
 	SelectByIndexStatementType
 	SelectFromMaterializedViewStatementType
 	DeleteStatementType
-	InsertStatement
-	Updatetatement
+	InsertStatementType
+	InsertJSONStatementType
+	UpdateStatementType
 	AlterColumnStatementType
 	DropColumnStatementType
+	AddColumnStatementType
 )
 
 //nolint:revive

diff --git a/pkg/typedef/interfaces.go b/pkg/typedef/interfaces.go
@@ -25,6 +25,7 @@ type Type interface {
 	CQLHolder() string
 	CQLPretty(string, []interface{}) (string, int)
 	GenValue(*rand.Rand, *PartitionRangeConfig) []interface{}
+	GenJSONValue(*rand.Rand, *PartitionRangeConfig) interface{}
 	LenValue() int
 	Indexable() bool
 	CQLType() gocql.TypeInfo