Skip to content

Commit 0974786

Browse files
akashthawaitccakashthawaitcc
committed
Merge remote-tracking branch 'upstream/master'
2 parents ef3c414 + 534145a commit 0974786

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

62 files changed

+1541
-315
lines changed

‎cmd/schema.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
2828
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
2929
"github.com/GoogleCloudPlatform/spanner-migration-tool/conversion"
30+
"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
3031
"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
3132
"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
3233
"github.com/GoogleCloudPlatform/spanner-migration-tool/proto/migration"
@@ -134,7 +135,16 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
134135
return subcommands.ExitFailure
135136
}
136137
} else {
137-
conv, err = convImpl.SchemaConv(cmd.project, sourceProfile, targetProfile, &ioHelper, &conversion.SchemaFromSourceImpl{})
138+
ctx := context.Background()
139+
ddlVerifier, err := expressions_api.NewDDLVerifierImpl(ctx, "", "")
140+
if err != nil {
141+
logger.Log.Error(fmt.Sprintf("error trying create ddl verifier: %v", err))
142+
return subcommands.ExitFailure
143+
}
144+
sfs := &conversion.SchemaFromSourceImpl{
145+
DdlVerifier: ddlVerifier,
146+
}
147+
conv, err = convImpl.SchemaConv(cmd.project, sourceProfile, targetProfile, &ioHelper, sfs)
138148
if err != nil {
139149
return subcommands.ExitFailure
140150
}

‎cmd/schema_and_data.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
2828
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
2929
"github.com/GoogleCloudPlatform/spanner-migration-tool/conversion"
30+
"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
3031
"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
3132
"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
3233
"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
@@ -136,7 +137,15 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
136137
dbURI string
137138
)
138139
convImpl := &conversion.ConvImpl{}
139-
conv, err = convImpl.SchemaConv(cmd.project, sourceProfile, targetProfile, &ioHelper, &conversion.SchemaFromSourceImpl{})
140+
ddlVerifier, err := expressions_api.NewDDLVerifierImpl(ctx, "", "")
141+
if err != nil {
142+
logger.Log.Error(fmt.Sprintf("error trying create ddl verifier: %v", err))
143+
return subcommands.ExitFailure
144+
}
145+
sfs := &conversion.SchemaFromSourceImpl{
146+
DdlVerifier: ddlVerifier,
147+
}
148+
conv, err = convImpl.SchemaConv(cmd.project, sourceProfile, targetProfile, &ioHelper, sfs)
140149
if err != nil {
141150
panic(err)
142151
}

‎common/constants/constants.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -123,11 +123,12 @@ const (
123123
DLQ_GCS string = "dlq"
124124

125125
// VerifyExpresions API
126-
CHECK_EXPRESSION = "CHECK"
127-
DEFAUT_EXPRESSION = "DEFAULT"
128-
DEFAULT_GENERATED = "DEFAULT_GENERATED"
129-
TEMP_DB = "smt-staging-db"
130-
126+
CHECK_EXPRESSION = "CHECK"
127+
DEFAULT_EXPRESSION = "DEFAULT"
128+
DEFAULT_GENERATED = "DEFAULT_GENERATED"
129+
TEMP_DB = "smt-staging-db"
130+
DB_URI = "projects/%s/instances/%s/databases/%s"
131+
131132
// Regex for matching database collation
132133
DB_COLLATION_REGEX = `(_[a-zA-Z0-9]+\\|\\)`
133134
)

‎common/utils/utils.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ import (
4141
"cloud.google.com/go/storage"
4242
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
4343
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/parse"
44+
"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
4445
"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
4546
"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/common"
4647
"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/spanner"
@@ -445,7 +446,16 @@ func GetLegacyModeSupportedDrivers() []string {
445446
func ReadSpannerSchema(ctx context.Context, conv *internal.Conv, client *sp.Client) error {
446447
infoSchema := spanner.InfoSchemaImpl{Client: client, Ctx: ctx, SpDialect: conv.SpDialect}
447448
processSchema := common.ProcessSchemaImpl{}
448-
err := processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, internal.AdditionalSchemaAttributes{IsSharded: false}, &common.SchemaToSpannerImpl{}, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
449+
expressionVerificationAccessor, _ := expressions_api.NewExpressionVerificationAccessorImpl(ctx, conv.SpProjectId, conv.SpInstanceId)
450+
ddlVerifier, err := expressions_api.NewDDLVerifierImpl(ctx, conv.SpProjectId, conv.SpInstanceId)
451+
if err != nil {
452+
return fmt.Errorf("error trying create ddl verifier: %v", err)
453+
}
454+
schemaToSpanner := common.SchemaToSpannerImpl{
455+
DdlV: ddlVerifier,
456+
ExpressionVerificationAccessor: expressionVerificationAccessor,
457+
}
458+
err = processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, internal.AdditionalSchemaAttributes{IsSharded: false}, &schemaToSpanner, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
449459
if err != nil {
450460
return fmt.Errorf("error trying to read and convert spanner schema: %v", err)
451461
}

‎conversion/conversion.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ import (
3737
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
3838
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/task"
3939
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
40+
"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
4041
"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
4142
"github.com/GoogleCloudPlatform/spanner-migration-tool/internal/reports"
4243
"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
@@ -79,7 +80,8 @@ func (ci *ConvImpl) SchemaConv(migrationProjectId string, sourceProfile profiles
7980
case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER, constants.ORACLE:
8081
return schemaFromSource.schemaFromDatabase(migrationProjectId, sourceProfile, targetProfile, &GetInfoImpl{}, &common.ProcessSchemaImpl{})
8182
case constants.PGDUMP, constants.MYSQLDUMP:
82-
return schemaFromSource.SchemaFromDump(sourceProfile.Driver, targetProfile.Conn.Sp.Dialect, ioHelper, &ProcessDumpByDialectImpl{})
83+
expressionVerificationAccessor, _ := expressions_api.NewExpressionVerificationAccessorImpl(context.Background(), targetProfile.Conn.Sp.Project, targetProfile.Conn.Sp.Instance)
84+
return schemaFromSource.SchemaFromDump(targetProfile.Conn.Sp.Project, targetProfile.Conn.Sp.Instance, sourceProfile.Driver, targetProfile.Conn.Sp.Dialect, ioHelper, &ProcessDumpByDialectImpl{ExpressionVerificationAccessor: expressionVerificationAccessor})
8385
default:
8486
return nil, fmt.Errorf("schema conversion for driver %s not supported", sourceProfile.Driver)
8587
}

‎conversion/conversion_from_source.go‎

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
2727
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/metrics"
2828
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
29+
"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
2930
"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
3031
"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
3132
"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
@@ -38,10 +39,12 @@ import (
3839

3940
type SchemaFromSourceInterface interface {
4041
schemaFromDatabase(migrationProjectId string, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, getInfo GetInfoInterface, processSchema common.ProcessSchemaInterface) (*internal.Conv, error)
41-
SchemaFromDump(driver string, spDialect string, ioHelper *utils.IOStreams, processDump ProcessDumpByDialectInterface) (*internal.Conv, error)
42+
SchemaFromDump(SpProjectId string, SpInstanceId string, driver string, spDialect string, ioHelper *utils.IOStreams, processDump ProcessDumpByDialectInterface) (*internal.Conv, error)
4243
}
4344

44-
type SchemaFromSourceImpl struct{}
45+
type SchemaFromSourceImpl struct {
46+
DdlVerifier expressions_api.DDLVerifier
47+
}
4548

4649
type DataFromSourceInterface interface {
4750
dataFromDatabase(ctx context.Context, migrationProjectId string, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client, getInfo GetInfoInterface, dataFromDb DataFromDatabaseInterface, snapshotMigration SnapshotMigrationInterface) (*writer.BatchWriter, error)
@@ -99,10 +102,17 @@ func (sads *SchemaFromSourceImpl) schemaFromDatabase(migrationProjectId string,
99102
additionalSchemaAttributes := internal.AdditionalSchemaAttributes{
100103
IsSharded: isSharded,
101104
}
102-
return conv, processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, additionalSchemaAttributes, &common.SchemaToSpannerImpl{}, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
105+
106+
ctx := context.Background()
107+
expressionVerificationAccessor, _ := expressions_api.NewExpressionVerificationAccessorImpl(ctx, conv.SpProjectId, conv.SpInstanceId)
108+
schemaToSpanner := common.SchemaToSpannerImpl{
109+
DdlV: sads.DdlVerifier,
110+
ExpressionVerificationAccessor: expressionVerificationAccessor,
111+
}
112+
return conv, processSchema.ProcessSchema(conv, infoSchema, common.DefaultWorkers, additionalSchemaAttributes, &schemaToSpanner, &common.UtilsOrderImpl{}, &common.InfoSchemaImpl{})
103113
}
104114

105-
func (sads *SchemaFromSourceImpl) SchemaFromDump(driver string, spDialect string, ioHelper *utils.IOStreams, processDump ProcessDumpByDialectInterface) (*internal.Conv, error) {
115+
func (sads *SchemaFromSourceImpl) SchemaFromDump(SpProjectId string, SpInstanceId string, driver string, spDialect string, ioHelper *utils.IOStreams, processDump ProcessDumpByDialectInterface) (*internal.Conv, error) {
106116
f, n, err := getSeekable(ioHelper.In)
107117
if err != nil {
108118
utils.PrintSeekError(driver, err, ioHelper.Out)
@@ -112,6 +122,7 @@ func (sads *SchemaFromSourceImpl) SchemaFromDump(driver string, spDialect string
112122
ioHelper.BytesRead = n
113123
conv := internal.MakeConv()
114124
conv.SpDialect = spDialect
125+
conv.Source = driver
115126
p := internal.NewProgress(n, "Generating schema", internal.Verbose(), false, int(internal.SchemaCreationInProgress))
116127
r := internal.NewReader(bufio.NewReader(f), p)
117128
conv.SetSchemaMode() // Build schema and ignore data in dump.

‎conversion/conversion_from_source_test.go‎

Lines changed: 54 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,14 @@ import (
1818
"fmt"
1919
"testing"
2020

21+
"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
2122
"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
2223
"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/common"
2324
"github.com/GoogleCloudPlatform/spanner-migration-tool/sources/mysql"
2425
"github.com/stretchr/testify/assert"
2526
"github.com/stretchr/testify/mock"
2627
)
2728

28-
2929
func TestSchemaFromDatabase(t *testing.T) {
3030
targetProfile := profiles.TargetProfile{
3131
Conn: profiles.TargetProfileConnection{
@@ -65,88 +65,88 @@ func TestSchemaFromDatabase(t *testing.T) {
6565
sourceProfileCloudDefault := profiles.SourceProfile{}
6666
// Avoid getting/setting env variables in the unit tests.
6767
testCases := []struct {
68-
name string
69-
sourceProfile profiles.SourceProfile
70-
getInfoError error
71-
processSchemaError error
72-
errorExpected bool
68+
name string
69+
sourceProfile profiles.SourceProfile
70+
getInfoError error
71+
processSchemaError error
72+
errorExpected bool
7373
}{
7474
{
75-
name: "successful source profile config for bulk migration",
76-
sourceProfile: sourceProfileConfigBulk,
77-
getInfoError: nil,
75+
name: "successful source profile config for bulk migration",
76+
sourceProfile: sourceProfileConfigBulk,
77+
getInfoError: nil,
7878
processSchemaError: nil,
79-
errorExpected: false,
79+
errorExpected: false,
8080
},
8181
{
82-
name: "source profile config for bulk migration: get info error",
83-
sourceProfile: sourceProfileConfigBulk,
84-
getInfoError: fmt.Errorf("error"),
82+
name: "source profile config for bulk migration: get info error",
83+
sourceProfile: sourceProfileConfigBulk,
84+
getInfoError: fmt.Errorf("error"),
8585
processSchemaError: nil,
86-
errorExpected: true,
86+
errorExpected: true,
8787
},
8888
{
89-
name: "source profile config for bulk migration: process schema error",
90-
sourceProfile: sourceProfileConfigBulk,
91-
getInfoError: nil,
89+
name: "source profile config for bulk migration: process schema error",
90+
sourceProfile: sourceProfileConfigBulk,
91+
getInfoError: nil,
9292
processSchemaError: fmt.Errorf("error"),
93-
errorExpected: true,
93+
errorExpected: true,
9494
},
9595
{
96-
name: "successful source profile config for dataflow migration",
97-
sourceProfile: sourceProfileConfigDataflow,
98-
getInfoError: nil,
96+
name: "successful source profile config for dataflow migration",
97+
sourceProfile: sourceProfileConfigDataflow,
98+
getInfoError: nil,
9999
processSchemaError: nil,
100-
errorExpected: false,
100+
errorExpected: false,
101101
},
102102
{
103-
name: "source profile config for dataflow migration: get info error",
104-
sourceProfile: sourceProfileConfigDataflow,
105-
getInfoError: fmt.Errorf("error"),
103+
name: "source profile config for dataflow migration: get info error",
104+
sourceProfile: sourceProfileConfigDataflow,
105+
getInfoError: fmt.Errorf("error"),
106106
processSchemaError: nil,
107-
errorExpected: true,
107+
errorExpected: true,
108108
},
109109
{
110-
name: "source profile config for dms migration",
111-
sourceProfile: sourceProfileConfigDms,
112-
getInfoError: nil,
110+
name: "source profile config for dms migration",
111+
sourceProfile: sourceProfileConfigDms,
112+
getInfoError: nil,
113113
processSchemaError: nil,
114-
errorExpected: true,
114+
errorExpected: true,
115115
},
116116
{
117-
name: "invalid source profile config",
118-
sourceProfile: sourceProfileConfigInvalid,
119-
getInfoError: nil,
117+
name: "invalid source profile config",
118+
sourceProfile: sourceProfileConfigInvalid,
119+
getInfoError: nil,
120120
processSchemaError: nil,
121-
errorExpected: true,
121+
errorExpected: true,
122122
},
123123
{
124-
name: "successful source profile cloud sql",
125-
sourceProfile: sourceProfileCloudSql,
126-
getInfoError: nil,
124+
name: "successful source profile cloud sql",
125+
sourceProfile: sourceProfileCloudSql,
126+
getInfoError: nil,
127127
processSchemaError: nil,
128-
errorExpected: false,
128+
errorExpected: false,
129129
},
130130
{
131-
name: "source profile cloud sql: get info error",
132-
sourceProfile: sourceProfileCloudSql,
133-
getInfoError: fmt.Errorf("error"),
131+
name: "source profile cloud sql: get info error",
132+
sourceProfile: sourceProfileCloudSql,
133+
getInfoError: fmt.Errorf("error"),
134134
processSchemaError: nil,
135-
errorExpected: true,
135+
errorExpected: true,
136136
},
137137
{
138-
name: "successful source profile default",
139-
sourceProfile: sourceProfileCloudDefault,
140-
getInfoError: nil,
138+
name: "successful source profile default",
139+
sourceProfile: sourceProfileCloudDefault,
140+
getInfoError: nil,
141141
processSchemaError: nil,
142-
errorExpected: false,
142+
errorExpected: false,
143143
},
144144
{
145-
name: "source profile default: get info error",
146-
sourceProfile: sourceProfileCloudDefault,
147-
getInfoError: fmt.Errorf("error"),
145+
name: "source profile default: get info error",
146+
sourceProfile: sourceProfileCloudDefault,
147+
getInfoError: fmt.Errorf("error"),
148148
processSchemaError: nil,
149-
errorExpected: true,
149+
errorExpected: true,
150150
},
151151
}
152152

@@ -159,8 +159,10 @@ func TestSchemaFromDatabase(t *testing.T) {
159159
gim.On("GetInfoSchema", "migration-project-id", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(mysql.InfoSchemaImpl{}, tc.getInfoError)
160160
ps.On("ProcessSchema", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(tc.processSchemaError)
161161

162-
s := SchemaFromSourceImpl{}
162+
s := SchemaFromSourceImpl{
163+
DdlVerifier: &expressions_api.MockDDLVerifier{},
164+
}
163165
_, err := s.schemaFromDatabase("migration-project-id", tc.sourceProfile, targetProfile, &gim, &ps)
164166
assert.Equal(t, tc.errorExpected, err != nil, tc.name)
165167
}
166-
}
168+
}

‎conversion/conversion_helper.go‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import (
2929
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/constants"
3030
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/metrics"
3131
"github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils"
32+
"github.com/GoogleCloudPlatform/spanner-migration-tool/expressions_api"
3233
"github.com/GoogleCloudPlatform/spanner-migration-tool/internal"
3334
"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
3435
"github.com/GoogleCloudPlatform/spanner-migration-tool/profiles"
@@ -41,17 +42,21 @@ import (
4142
"google.golang.org/protobuf/proto"
4243
)
4344

44-
type ProcessDumpByDialectInterface interface{
45+
type ProcessDumpByDialectInterface interface {
4546
ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error
4647
}
4748

48-
type ProcessDumpByDialectImpl struct{}
49+
type ProcessDumpByDialectImpl struct {
50+
ExpressionVerificationAccessor expressions_api.ExpressionVerificationAccessor
51+
DdlVerifier expressions_api.DDLVerifier
52+
}
4953

50-
type PopulateDataConvInterface interface{
54+
type PopulateDataConvInterface interface {
5155
populateDataConv(conv *internal.Conv, config writer.BatchWriterConfig, client *sp.Client) *writer.BatchWriter
5256
}
5357

5458
type PopulateDataConvImpl struct{}
59+
5560
// getSeekable returns a seekable file (with same content as f) and the size of the content (in bytes).
5661
func getSeekable(f *os.File) (*os.File, int64, error) {
5762
_, err := f.Seek(0, 0)
@@ -88,15 +93,14 @@ func getSeekable(f *os.File) (*os.File, int64, error) {
8893
func (pdd *ProcessDumpByDialectImpl) ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
8994
switch driver {
9095
case constants.MYSQLDUMP:
91-
return common.ProcessDbDump(conv, r, mysql.DbDumpImpl{})
96+
return common.ProcessDbDump(conv, r, mysql.DbDumpImpl{}, pdd.DdlVerifier, pdd.ExpressionVerificationAccessor)
9297
case constants.PGDUMP:
93-
return common.ProcessDbDump(conv, r, postgres.DbDumpImpl{})
98+
return common.ProcessDbDump(conv, r, postgres.DbDumpImpl{}, pdd.DdlVerifier, pdd.ExpressionVerificationAccessor)
9499
default:
95100
return fmt.Errorf("process dump for driver %s not supported", driver)
96101
}
97102
}
98103

99-
100104
func (pdc *PopulateDataConvImpl) populateDataConv(conv *internal.Conv, config writer.BatchWriterConfig, client *sp.Client) *writer.BatchWriter {
101105
rows := int64(0)
102106
config.Write = func(m []*sp.Mutation) error {
@@ -130,7 +134,6 @@ func (pdc *PopulateDataConvImpl) populateDataConv(conv *internal.Conv, config wr
130134
return batchWriter
131135
}
132136

133-
134137
func connectionConfig(sourceProfile profiles.SourceProfile) (interface{}, error) {
135138
switch sourceProfile.Driver {
136139
// For PG and MYSQL, When called as part of the subcommand flow, host/user/db etc will
@@ -199,4 +202,4 @@ func getDynamoDBClientConfig() (*aws.Config, error) {
199202
cfg.Endpoint = aws.String(endpointOverride)
200203
}
201204
return &cfg, nil
202-
}
205+
}

0 commit comments

Comments
 (0)