Implements store infrastructure for cockroachDB

Provides cockroachdb support but due to some limitation from cockroachDB, we could not implement it further. Development so far, - Implements store methods - Updates DDL generator to create/update tables and indexes - Provides sqlFunctionHandler, fieldToColumnTypeCaster handle, and workaround for LIKE operator
2021-04-12 12:09:35 +05:30
parent cfde435385
commit 109e23fc6a
6 changed files with 497 additions and 20 deletions
@@ -0,0 +1,116 @@
+package cockroach
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"github.com/Masterminds/squirrel"
+	"github.com/cortezaproject/corteza-server/pkg/ql"
+	"github.com/cortezaproject/corteza-server/store"
+	"github.com/cortezaproject/corteza-server/store/rdbms"
+	"github.com/cortezaproject/corteza-server/store/rdbms/instrumentation"
+	"github.com/lib/pq"
+	"github.com/ngrok/sqlmw"
+	"go.uber.org/zap"
+	"net/url"
+	"strings"
+)
+
+type (
+	Store struct {
+		*rdbms.Store
+	}
+)
+
+const (
+	storeTypeName      = "cockroachdb"
+	storeTypeWithDebug = "cockroachdb+debug"
+)
+
+// For now, CockroachDB implementation is disabled due to some limitation from cockroachDB!
+//
+//	Here is why could not work on this further:
+//	* Prefix/Expression Indexes.
+//		I.e. we are not able to use "lower" expression while creating/updating indexes
+//	* Not able to use Prefix/Expression into upsert.
+//		I.e. also not able to use "lower"(for most case) expression in upsert
+//
+// @note we could enable this once cockroachDB overcome above limitations :)
+func init() {
+	store.Register(Connect, storeTypeName, storeTypeWithDebug)
+	sql.Register(storeTypeWithDebug, sqlmw.Driver(new(pq.Driver), instrumentation.Debug()))
+}
+
+func Connect(ctx context.Context, dsn string) (store.Storer, error) {
+	var (
+		err error
+		cfg *rdbms.Config
+
+		s = new(Store)
+	)
+
+	if cfg, err = ProcDataSourceName(dsn); err != nil {
+		return nil, err
+	}
+
+	cfg.PlaceholderFormat = squirrel.Dollar
+	cfg.ErrorHandler = errorHandler
+	cfg.SqlFunctionHandler = sqlFunctionHandler
+	cfg.CastModuleFieldToColumnType = fieldToColumnTypeCaster
+
+	if s.Store, err = rdbms.Connect(ctx, cfg); err != nil {
+		return nil, err
+	}
+
+	ql.QueryEncoder = &QueryEncoder{}
+
+	return s, nil
+}
+
+func (s *Store) Upgrade(ctx context.Context, log *zap.Logger) (err error) {
+	if err = (&rdbms.Schema{}).Upgrade(ctx, NewUpgrader(log, s)); err != nil {
+		return fmt.Errorf("can not upgrade cockroach schema: %w", err)
+	}
+
+	return nil
+}
+
+// ProcDataSourceName validates given DSN and ensures
+// params are present and correct
+func ProcDataSourceName(dsn string) (c *rdbms.Config, err error) {
+	const (
+		validScheme = "postgres"
+	)
+	var (
+		scheme string
+		u      *url.URL
+	)
+
+	if u, err = url.Parse(dsn); err != nil {
+		return nil, err
+	}
+
+	if strings.HasPrefix(dsn, storeTypeName) {
+		scheme = validScheme
+		u.Scheme = validScheme
+	}
+
+	return &rdbms.Config{
+		DriverName:     scheme,
+		DataSourceName: u.String(),
+		DBName:         strings.Trim(u.Path, "/"),
+	}, nil
+}
+
+func errorHandler(err error) error {
+	if err != nil {
+		if implErr, ok := err.(*pq.Error); ok {
+			switch implErr.Code.Name() {
+			case "unique_violation":
+				return store.ErrNotUnique.Wrap(implErr)
+			}
+		}
+	}
+
+	return err
+}
@@ -0,0 +1,19 @@
+package cockroach
+
+import "github.com/cortezaproject/corteza-server/pkg/ql"
+
+type (
+	// QueryEncoder provides query parts encoding rules for CockroachDB
+	// see ql.QueryEncoder for mor info
+	QueryEncoder struct{}
+)
+
+var _ ql.Encoder = &QueryEncoder{}
+
+func (QueryEncoder) CaseInsensitiveLike(neg bool) string {
+	if neg {
+		return "NOT ILIKE"
+	} else {
+		return "ILIKE"
+	}
+}
@@ -0,0 +1,22 @@
+package cockroach
+
+import (
+	"fmt"
+	"github.com/cortezaproject/corteza-server/pkg/ql"
+	"strings"
+)
+
+func sqlFunctionHandler(f ql.Function) (ql.ASTNode, error) {
+	switch strings.ToUpper(f.Name) {
+	case "QUARTER", "YEAR":
+		return ql.MakeFormattedNode(fmt.Sprintf("EXTRACT(%s FROM %%s::date)", f.Name), f.Arguments...), nil
+	case "DATE_FORMAT":
+		return ql.MakeFormattedNode("TO_CHAR(%s, %s)", f.Arguments...), nil
+	case "DATE":
+		return ql.MakeFormattedNode("%s::DATE", f.Arguments...), nil
+	case "DATE_ADD", "DATE_SUB", "STD":
+		return nil, fmt.Errorf("%q function is currently unsupported in PostgreSQL store backend", f.Name)
+	}
+
+	return f, nil
+}
@@ -0,0 +1,40 @@
+package cockroach
+
+import (
+	"fmt"
+	"github.com/cortezaproject/corteza-server/store/rdbms"
+)
+
+// fieldToColumnTypeCaster handles special ComposeModule field query representations
+// @todo Not as elegant as it should be but it'll do the trick until the #2 store iteration
+//
+// Return parameters:
+//   * full cast: query column + datatype cast
+//   * field cast tpl: fmt template to get query column
+//   * type cast tpl: fmt template to cast the compared to value
+func fieldToColumnTypeCaster(field rdbms.ModuleFieldTypeDetector, ident string) (string, string, string, error) {
+	fcp := "rv_%s.value"
+	fcpRef := "rv_%s.ref"
+
+	switch true {
+	case field.IsBoolean():
+		tcp := "%s NOT IN ('', '0', 'false', 'f',  'FALSE', 'F')"
+		fc := fmt.Sprintf(fcp, ident)
+		return fmt.Sprintf(tcp, fc), fcp, tcp, nil
+	case field.IsNumeric():
+		tcp := "CAST('0' || %s AS NUMERIC)"
+		fc := fmt.Sprintf(fcp, ident)
+		return fmt.Sprintf(tcp, fc), fcp, tcp, nil
+	case field.IsDateTime():
+		tcp := "CAST(%s AS TIMESTAMP)"
+		fc := fmt.Sprintf(fcp, ident)
+		return fmt.Sprintf(tcp, fc), fcp, tcp, nil
+	case field.IsRef():
+		tcp := "%s"
+		fc := fmt.Sprintf(fcpRef, ident)
+		return fmt.Sprintf(tcp, fc), fcpRef, tcp, nil
+	}
+	tcp := "%s"
+	fc := fmt.Sprintf(fcp, ident)
+	return fmt.Sprintf(tcp, fc), fcp, tcp, nil
+}
@@ -0,0 +1,267 @@
+package cockroach
+
+// PostgreSQL specific prefixes, sql
+// templates, functions and other helpers
+
+import (
+	"context"
+	"fmt"
+	"github.com/cortezaproject/corteza-server/store/rdbms"
+	"github.com/cortezaproject/corteza-server/store/rdbms/ddl"
+	"go.uber.org/zap"
+)
+
+type (
+	upgrader struct {
+		s   *Store
+		log *zap.Logger
+		ddl *ddl.Generator
+	}
+)
+
+// NewUpgrader returns PostgreSQL schema upgrader
+func NewUpgrader(log *zap.Logger, store *Store) *upgrader {
+	var g = &upgrader{store, log, ddl.NewGenerator(log)}
+
+	// All modifications we need for the DDL generator
+	// to properly support CockroachDB dialect
+
+	g.ddl.AddTemplate("index-fields", `
+	({{ range $n, $f := . -}}
+	{{ if $n }}, {{ end }}
+	{{- .Field | trimExpression}}
+	{{- if .Desc }} DESC{{ end }}
+	{{- end }})
+	`)
+	g.ddl.AddTemplate("if-not-exists-clause", "IF NOT EXISTS")
+
+	return g
+}
+
+// Before runs before all tables are upgraded
+func (u upgrader) Before(ctx context.Context) error {
+	return rdbms.GenericUpgrades(u.log, u).Before(ctx)
+}
+
+// After runs after all tables are upgraded
+func (u upgrader) After(ctx context.Context) error {
+	return rdbms.GenericUpgrades(u.log, u).After(ctx)
+}
+
+// CreateTable is triggered for every table defined in the rdbms package
+//
+// It checks if table is missing and creates it, otherwise
+// it runs
+func (u upgrader) CreateTable(ctx context.Context, t *ddl.Table) (err error) {
+	var exists bool
+	if exists, err = u.TableExists(ctx, t.Name); err != nil {
+		return
+	}
+
+	if !exists {
+		if err = u.Exec(ctx, u.ddl.CreateTable(t)); err != nil {
+			return err
+		}
+
+		for _, i := range t.Indexes {
+			if err = u.Exec(ctx, u.ddl.CreateIndex(i)); err != nil {
+				return fmt.Errorf("could not create index %s on table %s: %w", i.Name, i.Table, err)
+			}
+		}
+	} else {
+		return u.upgradeTable(ctx, t)
+	}
+
+	return nil
+}
+
+func (u upgrader) Exec(ctx context.Context, sql string, aa ...interface{}) error {
+	_, err := u.s.DB().ExecContext(ctx, sql, aa...)
+	return err
+}
+
+// upgradeTable applies any necessary changes connected to that specific table
+func (u upgrader) upgradeTable(ctx context.Context, t *ddl.Table) error {
+	g := rdbms.GenericUpgrades(u.log, u)
+
+	switch t.Name {
+	default:
+		return g.Upgrade(ctx, t)
+	}
+}
+
+func (u upgrader) TableExists(ctx context.Context, table string) (bool, error) {
+	//const tableSchemaName = "public"
+	var exists bool
+	// @todo use pq.NewError(regex?!) -> ADDED IF NOT EXISTS
+	//if err := u.s.DB().GetContext(ctx, &exists, fmt.Sprintf("SELECT * FROM %[1]s.%[2]s", tableSchemaName, table)); err != nil {
+	//	if err.Error() == fmt.Sprintf("pq: relation \"%[1]s.%[2]s\" does not exist", tableSchemaName, table) {
+	//		exists = false
+	//	} else {
+	//		return false, fmt.Errorf("could not check if table exists: %w", err)
+	//	}
+	//}
+
+	return exists, nil
+}
+
+func (u upgrader) DropTable(ctx context.Context, table string) (dropped bool, err error) {
+	var exists bool
+	exists, err = u.TableExists(ctx, table)
+	if err != nil || !exists {
+		return false, err
+	}
+
+	err = u.Exec(ctx, fmt.Sprintf(`DROP TABLE %s`, table))
+	if err != nil {
+		return false, err
+	}
+
+	return true, nil
+}
+
+func (u upgrader) TableSchema(ctx context.Context, table string) (ddl.Columns, error) {
+	return nil, fmt.Errorf("pending implementation")
+}
+
+// AddColumn adds column to table
+func (u upgrader) AddColumn(ctx context.Context, table string, col *ddl.Column) (added bool, err error) {
+	err = func() error {
+		var columns ddl.Columns
+		if columns, err = u.getColumns(ctx, table); err != nil {
+			return err
+		}
+
+		if columns.Get(col.Name) != nil {
+			return nil
+		}
+
+		if err = u.Exec(ctx, u.ddl.AddColumn(table, col)); err != nil {
+			return err
+		}
+
+		added = true
+		return nil
+	}()
+
+	if err != nil {
+		return false, fmt.Errorf("could not add column %q to %q: %w", col.Name, table, err)
+	}
+
+	return
+}
+
+// DropColumn drops column from table
+func (u upgrader) DropColumn(ctx context.Context, table, column string) (dropped bool, err error) {
+	err = func() error {
+		var columns ddl.Columns
+		if columns, err = u.getColumns(ctx, table); err != nil {
+			return err
+		}
+
+		if columns.Get(column) == nil {
+			return nil
+		}
+
+		if err = u.Exec(ctx, u.ddl.DropColumn(table, column)); err != nil {
+			return err
+		}
+
+		dropped = true
+		return nil
+	}()
+
+	if err != nil {
+		return false, fmt.Errorf("could not drop column %q from %q: %w", column, table, err)
+	}
+
+	return
+}
+
+// RenameColumn renames column on a table
+func (u upgrader) RenameColumn(ctx context.Context, table, oldName, newName string) (changed bool, err error) {
+	err = func() error {
+		if oldName == newName {
+			return nil
+		}
+
+		var columns ddl.Columns
+		if columns, err = u.getColumns(ctx, table); err != nil {
+			return err
+		}
+
+		if columns.Get(oldName) == nil {
+			// Old column does not exist anymore
+
+			if columns.Get(newName) == nil {
+				return fmt.Errorf("old and new columns are missing")
+			}
+
+			return nil
+		}
+
+		if columns.Get(newName) != nil {
+			return fmt.Errorf("new column already exists")
+
+		}
+
+		if err = u.Exec(ctx, u.ddl.RenameColumn(table, oldName, newName)); err != nil {
+			return err
+		}
+
+		changed = true
+		return nil
+	}()
+
+	if err != nil {
+		return false, fmt.Errorf("could not rename column %q on table %q to %q: %w", oldName, table, newName, err)
+	}
+
+	return
+}
+
+func (u upgrader) AddPrimaryKey(ctx context.Context, table string, ind *ddl.Index) (added bool, err error) {
+	if err = u.Exec(ctx, u.ddl.AddPrimaryKey(table, ind)); err != nil {
+		return false, fmt.Errorf("could not add primary key to table %s: %w", table, err)
+	}
+
+	return true, nil
+}
+
+// loads and returns all tables columns
+func (u upgrader) getColumns(ctx context.Context, table string) (out ddl.Columns, err error) {
+	type (
+		col struct {
+			Name       string `db:"column_name"`
+			IsNullable bool   `db:"is_nullable"`
+			DataType   string `db:"data_type"`
+		}
+	)
+
+	var (
+		lookup = `SELECT column_name,
+                         is_nullable = 'YES' AS is_nullable,
+                         data_type
+                    FROM information_schema.columns 
+                   WHERE table_catalog = $1 
+                     AND table_name = $2`
+
+		cols []*col
+	)
+
+	if err = u.s.DB().SelectContext(ctx, &cols, lookup, u.s.Config().DBName, table); err != nil {
+		return nil, err
+	}
+
+	out = make([]*ddl.Column, len(cols))
+	for i := range cols {
+		out[i] = &ddl.Column{
+			Name: cols[i].Name,
+			//Type:         ddl.ColumnType{},
+			IsNull: cols[i].IsNullable,
+			//DefaultValue: "",
+		}
+	}
+
+	return out, nil
+}
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"fmt"
 	"go.uber.org/zap"
+	"regexp"
 	"strings"
 	"text/template"
 )
@@ -21,22 +22,22 @@ type (
 const (
 	// table creation
 	genericCreateTable = `
-CREATE TABLE {{ .Name }} (
-{{ range $n, $c := .Columns -}}
-{{ if $n }}, {{ else }}  {{ end }}{{ template "create-table-column" . }}
-{{ end -}}
-{{- if .PrimaryKey }}
-, PRIMARY KEY {{ template "index-fields" .PrimaryKey.Fields }}
-{{- end }}
-) {{- template "create-table-suffix" . -}}
-`
+	CREATE TABLE {{template "if-not-exists-clause" .}} {{ .Name }} (
+	{{ range $n, $c := .Columns -}}
+	{{ if $n }}, {{ else }}  {{ end }}{{ template "create-table-column" . }}
+	{{ end -}}
+	{{- if .PrimaryKey }}
+	, PRIMARY KEY {{ template "index-fields" .PrimaryKey.Fields }}
+	{{- end }}
+	) {{- template "create-table-suffix" . -}}
+	`
 	genericCreateTableSuffix = ``

 	genericCreateTableColumn = `
 	{{- .Name }} {{ columnType .Type }}
 	{{- if not .IsNull }} NOT NULL{{end}}
 	{{- if .DefaultValue }} DEFAULT {{ .DefaultValue }} {{end}}
-`
+	`

 	genericAddColumn     = `ALTER TABLE {{ .Table }} ADD {{ template "create-table-column" .Column }}`
 	genericAddPrimaryKey = `ALTER TABLE {{ .Table }} ADD CONSTRAINT PRIMARY KEY {{ template "index-fields" .PrimaryKey.Fields }}`
@@ -44,21 +45,23 @@ CREATE TABLE {{ .Name }} (
 	genericRenameColumn  = `ALTER TABLE {{ .Table }} RENAME COLUMN {{ .OldName }} TO {{ .NewName }}`

 	// index creation
-	genericCreateIndex = `CREATE {{ if .Unique }}UNIQUE {{ end }}INDEX {{ template "index-name" . }} ON {{ .Table }} {{ template "index-fields" .Fields }}{{ template "index-condition" . }}`
+	genericCreateIndex = `CREATE {{ if .Unique }}UNIQUE {{ end }}INDEX {{ template "if-not-exists-clause" . }} {{ template "index-name" . }} ON {{ .Table }} {{ template "index-fields" .Fields }}{{ template "index-condition" . }}`

 	genericIndexName      = `{{ .Table }}_{{ .Name }}`
 	genericIndexCondition = `{{- if .Condition }} WHERE ({{ .Condition }}){{ end }}`

 	// index creation
 	genericIndexFields = `
-({{ range $n, $f := . -}}
-	{{ if $n }}, {{ end }}
-	{{- if .Expr}}({{ end }}
-	{{- .Field }}
-	{{- if .Expr}}){{ end }}
-	{{- if .Desc }} DESC{{ end }}
-{{- end }})
-`
+	({{ range $n, $f := . -}}
+		{{ if $n }}, {{ end }}
+		{{- if .Expr}}({{ end }}
+		{{- .Field }}
+		{{- if .Expr}}){{ end }}
+		{{- if .Desc }} DESC{{ end }}
+	{{- end }})
+	`
+	// table/index exist or not clause
+	genericIfNotExistsClause = ``
 )

 func NewGenerator(log *zap.Logger) *Generator {
@@ -75,6 +78,7 @@ func NewGenerator(log *zap.Logger) *Generator {
 	g.AddTemplate("index-condition", genericIndexCondition)
 	g.AddTemplate("index-name", genericIndexName)
 	g.AddTemplate("index-fields", genericIndexFields)
+	g.AddTemplate("if-not-exists-clause", genericIfNotExistsClause)

 	return g
 }
@@ -93,7 +97,16 @@ func (g *Generator) AddTemplateFunc(name string, fn interface{}) {
 }

 func (g *Generator) AddTemplate(name, tpl string) {
-	template.Must(g.tpl.New(name).Parse(strings.TrimSpace(tpl)))
+	funcMap := template.FuncMap{
+		"trimExpression": func(s string) string {
+			re := regexp.MustCompile(`^.*\((\w+)\).*$`)
+			if str := re.FindAllStringSubmatch(s, 1); len(str) > 0 && len(str[0]) > 0 && len(str[0][1]) > 0 {
+				return str[0][1]
+			}
+			return s
+		},
+	}
+	template.Must(g.tpl.Funcs(funcMap).New(name).Parse(strings.TrimSpace(tpl)))
 }

 func (g *Generator) CreateTable(t *Table) string {