Embed schema files

schema/cassandra/embed.go

@@ -0,0 +1,26 @@
+// Copyright (c) 2019 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package cassandra
+
+import "embed"
+
+//go:embed cadence/* visibility/*
+var SchemaFS embed.FS

schema/mysql/embed.go

@@ -0,0 +1,26 @@
+// Copyright (c) 2019 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package mysql
+
+import "embed"
+
+//go:embed v57/cadence/* v57/visibility/*
+var SchemaFS embed.FS

schema/postgres/embed.go

@@ -0,0 +1,26 @@
+// Copyright (c) 2019 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package postgres
+
+import "embed"
+
+//go:embed cadence/* visibility/*
+var SchemaFS embed.FS

tools/common/schema/handler.go

@@ -22,6 +22,7 @@ package schema
 
 import (
 	"fmt"
+	"os"
 
 	"github.com/urfave/cli"
 )
@@ -86,7 +87,11 @@ func Update(cli *cli.Context, db SchemaClient) error {
 
 func newUpdateConfig(cli *cli.Context) (*UpdateConfig, error) {
 	config := new(UpdateConfig)
-	config.SchemaDir = cli.String(CLIOptSchemaDir)
+	schemaDir := cli.String(CLIOptSchemaDir)
+	if len(schemaDir) == 0 {
+		return nil, NewConfigError("missing " + flag(CLIOptSchemaDir) + " argument ")
+	}
+	config.SchemaFS = os.DirFS(schemaDir)
 	config.IsDryRun = cli.Bool(CLIOptDryrun)
 	config.TargetVersion = cli.String(CLIOptTargetVersion)
 
@@ -129,8 +134,8 @@ func validateSetupConfig(config *SetupConfig) error {
 }
 
 func validateUpdateConfig(config *UpdateConfig) error {
-	if len(config.SchemaDir) == 0 {
-		return NewConfigError("missing " + flag(CLIOptSchemaDir) + " argument ")
+	if config.SchemaFS == nil {
+		return fmt.Errorf("schema file system cannot be nil")
 	}
 	if len(config.TargetVersion) > 0 {
 		ver, err := parseValidateVersion(config.TargetVersion)

tools/common/schema/setuptask.go

@@ -20,7 +20,10 @@
 
 package schema
 
-import "log"
+import (
+	"log"
+	"os"
+)
 
 // SetupTask represents a task
 // that sets up cassandra schema on
@@ -57,7 +60,11 @@ func (task *SetupTask) Run() error {
 	}
 
 	if len(config.SchemaFilePath) > 0 {
-		stmts, err := ParseFile(config.SchemaFilePath)
+		file, err := os.Open(config.SchemaFilePath)
+		if err != nil {
+			return err
+		}
+		stmts, err := ParseFile(file)
 		if err != nil {
 			return err
 		}

tools/common/schema/test/dbtest.go

@@ -88,7 +88,7 @@ func (tb *DBTestBase) RunParseFileTest(content string) {
 
 	_, err = cqlFile.WriteString(content)
 	tb.NoError(err)
-	stmts, err := schema.ParseFile(cqlFile.Name())
+	stmts, err := schema.ParseFile(cqlFile)
 	tb.Nil(err)
 	tb.Equal(2, len(stmts), "wrong number of sql statements")
 }

tools/common/schema/types.go

@@ -22,6 +22,7 @@ package schema
 
 import (
 	"fmt"
+	"io/fs"
 	"regexp"
 )
 
@@ -36,7 +37,7 @@ type (
 	UpdateConfig struct {
 		DBName        string
 		TargetVersion string
-		SchemaDir     string
+		SchemaFS      fs.FS
 		IsDryRun      bool
 	}
 	// SetupConfig holds the config

tools/common/schema/updatetask.go

@@ -28,7 +28,8 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"io"
+	"io/fs"
 	"log"
 	"sort"
 	"strings"
@@ -190,7 +191,7 @@ func (task *UpdateTask) buildChangeSet(currVer string) ([]changeSet, error) {
 
 	config := task.config
 
-	verDirs, err := readSchemaDir(config.SchemaDir, currVer, config.TargetVersion)
+	verDirs, err := readSchemaDir(config.SchemaFS, currVer, config.TargetVersion)
 	if err != nil {
 		return nil, fmt.Errorf("error listing schema dir:%v", err.Error())
 	}
@@ -199,9 +200,7 @@ func (task *UpdateTask) buildChangeSet(currVer string) ([]changeSet, error) {
 
 	for _, vd := range verDirs {
 
-		dirPath := config.SchemaDir + "/" + vd
-
-		m, e := readManifest(dirPath)
+		m, e := readManifest(config.SchemaFS, vd)
 		if e != nil {
 			return nil, fmt.Errorf("error processing manifest for version %v:%v", vd, e.Error())
 		}
@@ -217,7 +216,7 @@ func (task *UpdateTask) buildChangeSet(currVer string) ([]changeSet, error) {
 				vd, m.CurrVersion)
 		}
 
-		stmts, e := task.parseSQLStmts(dirPath, m)
+		stmts, e := task.parseSQLStmts(vd, m)
 		if e != nil {
 			return nil, e
 		}
@@ -243,7 +242,11 @@ func (task *UpdateTask) parseSQLStmts(dir string, manifest *manifest) ([]string,
 
 	for _, file := range manifest.SchemaUpdateCqlFiles {
 		path := dir + "/" + file
-		stmts, err := ParseFile(path)
+		f, err := task.config.SchemaFS.Open(path)
+		if err != nil {
+			return nil, fmt.Errorf("error opening file %v, err=%v", path, err)
+		}
+		stmts, err := ParseFile(f)
 		if err != nil {
 			return nil, fmt.Errorf("error parsing file %v, err=%v", path, err)
 		}
@@ -273,15 +276,19 @@ func validateCQLStmts(stmts []string) error {
 	return nil
 }
 
-func readManifest(dirPath string) (*manifest, error) {
-
-	filePath := dirPath + "/" + manifestFileName
-	jsonStr, err := ioutil.ReadFile(filePath)
+func readManifest(fileSystem fs.FS, subdir string) (*manifest, error) {
+	fsys, err := fs.Sub(fileSystem, subdir)
+	if err != nil {
+		return nil, err
+	}
+	file, err := fsys.Open(manifestFileName)
+	if err != nil {
+		return nil, err
+	}
+	jsonBlob, err := io.ReadAll(file)
 	if err != nil {
 		return nil, err
 	}
-
-	jsonBlob := []byte(jsonStr)
 
 	var manifest manifest
 	err = json.Unmarshal(jsonBlob, &manifest)
@@ -322,12 +329,11 @@ func readManifest(dirPath string) (*manifest, error) {
 // this method has an assumption that the subdirs containing the
 // schema changes will be of the form vx.x, where x.x is the version
 // returns error when
-//   - startVer <= endVer
+//   - startVer >= endVer
 //   - endVer is empty and no subdirs have version >= startVer
 //   - endVer is non-empty and subdir with version == endVer is not found
-func readSchemaDir(dir string, startVer string, endVer string) ([]string, error) {
-
-	subdirs, err := ioutil.ReadDir(dir)
+func readSchemaDir(fileSystem fs.FS, startVer string, endVer string) ([]string, error) {
+	subdirs, err := fs.ReadDir(fileSystem, ".")
 	if err != nil {
 		return nil, err
 	}

tools/common/schema/updatetask_test.go

@@ -21,12 +21,17 @@
 package schema
 
 import (
+	"io/fs"
 	"io/ioutil"
 	"os"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+
+	"github.com/uber/cadence/schema/cassandra"
+	"github.com/uber/cadence/schema/mysql"
+	"github.com/uber/cadence/schema/postgres"
 )
 
 type UpdateTaskTestSuite struct {
@@ -58,46 +63,84 @@ func (s *UpdateTaskTestSuite) TestReadSchemaDir() {
 		s.NoError(os.Mkdir(squashDir+"/"+d, os.FileMode(0444)))
 	}
 
-	_, err := readSchemaDir(tmpDir, "11.0", "11.2")
+	_, err := readSchemaDir(os.DirFS(tmpDir), "11.0", "11.2")
 	s.Error(err)
-	_, err = readSchemaDir(tmpDir, "0.5", "10.3")
+	_, err = readSchemaDir(os.DirFS(tmpDir), "0.5", "10.3")
 	s.Error(err)
-	_, err = readSchemaDir(tmpDir, "1.5", "1.5")
+	_, err = readSchemaDir(os.DirFS(tmpDir), "1.5", "1.5")
 	s.Error(err)
-	_, err = readSchemaDir(tmpDir, "1.5", "0.5")
+	_, err = readSchemaDir(os.DirFS(tmpDir), "1.5", "0.5")
 	s.Error(err)
-	_, err = readSchemaDir(tmpDir, "10.3", "")
+	_, err = readSchemaDir(os.DirFS(tmpDir), "10.3", "")
 	s.Error(err)
-	_, err = readSchemaDir(emptyDir, "11.0", "")
+	_, err = readSchemaDir(os.DirFS(emptyDir), "11.0", "")
 	s.Error(err)
-	_, err = readSchemaDir(emptyDir, "10.1", "")
+	_, err = readSchemaDir(os.DirFS(emptyDir), "10.1", "")
 	s.Error(err)
 
-	ans, err := readSchemaDir(tmpDir, "0.4", "10.2")
+	ans, err := readSchemaDir(os.DirFS(tmpDir), "0.4", "10.2")
 	s.NoError(err)
 	s.Equal([]string{"v0.5", "v1.5", "v2.5", "v3.5", "v10.2"}, ans)
 
-	ans, err = readSchemaDir(tmpDir, "0.5", "3.5")
+	ans, err = readSchemaDir(os.DirFS(tmpDir), "0.5", "3.5")
 	s.NoError(err)
 	s.Equal([]string{"v1.5", "v2.5", "v3.5"}, ans)
 
-	ans, err = readSchemaDir(tmpDir, "10.2", "")
+	ans, err = readSchemaDir(os.DirFS(tmpDir), "10.2", "")
 	s.NoError(err)
 	s.Equal(0, len(ans))
 
-	ans, err = readSchemaDir(squashDir, "0.4", "10.2")
+	ans, err = readSchemaDir(os.DirFS(squashDir), "0.4", "10.2")
 	s.NoError(err)
 	s.Equal([]string{"v0.5", "s0.5-10.2"}, ans)
 
-	ans, err = readSchemaDir(squashDir, "0.5", "3.5")
+	ans, err = readSchemaDir(os.DirFS(squashDir), "0.5", "3.5")
 	s.NoError(err)
 	s.Equal([]string{"v1.5", "s1.5-3.5"}, ans)
 
-	ans, err = readSchemaDir(squashDir, "10.2", "")
+	ans, err = readSchemaDir(os.DirFS(squashDir), "10.2", "")
 	s.NoError(err)
 	s.Empty(ans)
 }
 
+func (s *UpdateTaskTestSuite) TestReadSchemaDirFromEmbeddings() {
+	fsys, err := fs.Sub(cassandra.SchemaFS, "cadence/versioned")
+	s.NoError(err)
+	ans, err := readSchemaDir(fsys, "0.30", "")
+	s.NoError(err)
+	s.Equal([]string{"v0.31", "v0.32", "v0.33", "v0.34"}, ans)
+
+	fsys, err = fs.Sub(cassandra.SchemaFS, "visibility/versioned")
+	s.NoError(err)
+	ans, err = readSchemaDir(fsys, "0.6", "")
+	s.NoError(err)
+	s.Equal([]string{"v0.7", "v0.8"}, ans)
+
+	fsys, err = fs.Sub(mysql.SchemaFS, "v57/cadence/versioned")
+	s.NoError(err)
+	ans, err = readSchemaDir(fsys, "0.3", "")
+	s.NoError(err)
+	s.Equal([]string{"v0.4", "v0.5"}, ans)
+
+	fsys, err = fs.Sub(mysql.SchemaFS, "v57/visibility/versioned")
+	s.NoError(err)
+	ans, err = readSchemaDir(fsys, "0.5", "")
+	s.NoError(err)
+	s.Equal([]string{"v0.6"}, ans)
+
+	fsys, err = fs.Sub(postgres.SchemaFS, "cadence/versioned")
+	s.NoError(err)
+	ans, err = readSchemaDir(fsys, "0.3", "")
+	s.NoError(err)
+	s.Equal([]string{"v0.4"}, ans)
+
+	fsys, err = fs.Sub(postgres.SchemaFS, "visibility/versioned")
+	s.NoError(err)
+	ans, err = readSchemaDir(fsys, "0.5", "")
+	s.NoError(err)
+	s.Equal([]string{"v0.6"}, ans)
+}
+
 func (s *UpdateTaskTestSuite) TestReadManifest() {
 
 	tmpDir := s.T().TempDir()
@@ -289,7 +332,7 @@ func (s *UpdateTaskTestSuite) runReadManifestTest(dir, input, currVer, minVer, d
 	err := ioutil.WriteFile(file, []byte(input), os.FileMode(0644))
 	s.Nil(err)
 
-	m, err := readManifest(dir)
+	m, err := readManifest(os.DirFS(dir), ".")
 	if isErr {
 		s.Error(err)
 		return