Skip to content

Shreya/transformation #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
VivekY1098 wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Shreya/transformation #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d34353f
Advanced transaformation integration for forward migration
shreyakhajanchi Jul 4, 2024
7035acc
added test cases
shreyakhajanchi Jul 7, 2024
42cdec6
flaky test solved
shreyakhajanchi Jul 8, 2024
e591352
minor fixes
shreyakhajanchi Jul 9, 2024
bd8b549
merge master
shreyakhajanchi Jul 9, 2024
91f323d
ng build
shreyakhajanchi Jul 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 34 additions & 32 deletions profiles/source_profile.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ import (

type SourceProfileType int


const (
SourceProfileTypeUnset = iota
SourceProfileTypeFile
Expand All @@ -54,7 +53,7 @@ type SourceProfileDialectInterface interface {
NewSourceProfileConnectionOracle(params map[string]string, g utils.GetUtilInfoInterface) (SourceProfileConnectionOracle, error)
}

type SourceProfileDialectImpl struct {}
type SourceProfileDialectImpl struct{}

// Interface to create new source profiles for different input types
type NewSourceProfileInterface interface {
Expand Down Expand Up @@ -101,11 +100,11 @@ const (
)

type SourceProfileConnectionCloudSQLMySQL struct {
User string
Db string
User string
Db string
InstanceName string
Project string
Region string
Project string
Region string
}

func (spd *SourceProfileDialectImpl) NewSourceProfileConnectionCloudSQLMySQL(params map[string]string, g utils.GetUtilInfoInterface) (SourceProfileConnectionCloudSQLMySQL, error) {
Expand Down Expand Up @@ -202,11 +201,11 @@ func (spd *SourceProfileDialectImpl) NewSourceProfileConnectionMySQL(params map[
}

type SourceProfileConnectionCloudSQLPostgreSQL struct {
User string
Db string
User string
Db string
InstanceName string
Project string
Region string
Project string
Region string
}

func (spd *SourceProfileDialectImpl) NewSourceProfileConnectionCloudSQLPostgreSQL(params map[string]string, g utils.GetUtilInfoInterface) (SourceProfileConnectionCloudSQLPostgreSQL, error) {
Expand Down Expand Up @@ -468,9 +467,9 @@ type SourceProfileConnection struct {
}

type SourceProfileConnectionCloudSQL struct {
Ty SourceProfileConnectionTypeCloudSQL
Mysql SourceProfileConnectionCloudSQLMySQL
Pg SourceProfileConnectionCloudSQLPostgreSQL
Ty SourceProfileConnectionTypeCloudSQL
Mysql SourceProfileConnectionCloudSQLMySQL
Pg SourceProfileConnectionCloudSQLPostgreSQL
}

func (nsp *NewSourceProfileImpl) NewSourceProfileConnection(source string, params map[string]string, s SourceProfileDialectInterface) (SourceProfileConnection, error) {
Expand Down Expand Up @@ -571,16 +570,16 @@ type DirectConnectionConfig struct {

type DatastreamConnProfileSource struct {
Name string `json:"name"`
Host string `json:"host"`
User string `json:"user"`
Port string `json:"port"`
Password string `json:"password"`
Location string `json:"location"`
Host string `json:"host"`
User string `json:"user"`
Port string `json:"port"`
Password string `json:"password"`
Location string `json:"location"`
}

type DatastreamConnProfileTarget struct {
Name string `json:"name"`
Location string `json:"location"`
Location string `json:"location"`
}

type DatastreamConfig struct {
Expand All @@ -607,18 +606,21 @@ type DataflowConfig struct {
AdditionalUserLabels string `json:"additionalUserLabels"`
KmsKeyName string `json:"kmsKeyName"`
GcsTemplatePath string `json:"gcsTemplatePath"`
CustomJarPath string `json:"customJarPath"`
CustomClassName string `json:"customClassName"`
CustomParameter string `json:"customParameter"`
}

type DataShard struct {
DataShardId string `json:"dataShardId"`
DataShardId string `json:"dataShardId"`
SrcConnectionProfile DatastreamConnProfileSource `json:"srcConnectionProfile"`
DstConnectionProfile DatastreamConnProfileTarget `json:"dstConnectionProfile"`
DatastreamConfig DatastreamConfig `json:"datastreamConfig"`
GcsConfig GcsConfig `json:"gcsConfig"`
DataflowConfig DataflowConfig `json:"dataflowConfig"`
TmpDir string `json:"tmpDir"`
StreamLocation string `json:"streamLocation"`
LogicalShards []LogicalShard `json:"databases"`
DatastreamConfig DatastreamConfig `json:"datastreamConfig"`
GcsConfig GcsConfig `json:"gcsConfig"`
DataflowConfig DataflowConfig `json:"dataflowConfig"`
TmpDir string `json:"tmpDir"`
StreamLocation string `json:"streamLocation"`
LogicalShards []LogicalShard `json:"databases"`
}

type LogicalShard struct {
Expand Down Expand Up @@ -690,13 +692,13 @@ func NewSourceProfileCsv(params map[string]string) SourceProfileCsv {
}

type SourceProfile struct {
Driver string
Ty SourceProfileType
File SourceProfileFile
Conn SourceProfileConnection
Driver string
Ty SourceProfileType
File SourceProfileFile
Conn SourceProfileConnection
ConnCloudSQL SourceProfileConnectionCloudSQL
Config SourceProfileConfig
Csv SourceProfileCsv
Config SourceProfileConfig
Csv SourceProfileCsv
}

// UseTargetSchema returns true if the driver expects an existing schema
Expand Down
15 changes: 15 additions & 0 deletions streaming/streaming.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,9 @@ type DataflowCfg struct {
KmsKeyName string `json:"kmsKeyName"`
GcsTemplatePath string `json:"gcsTemplatePath"`
DbNameToShardIdMap map[string]string `json:"dbNameToShardIdMap"`
CustomJarPath string `json:"customJarPath"`
CustomClassName string `json:"customClassName"`
CustomParameter string `json:"customParameter"`
}

type StreamingCfg struct {
Expand Down Expand Up @@ -763,6 +766,15 @@ func LaunchDataflowJob(ctx context.Context, migrationProjectId string, targetPro
KmsKeyName: dataflowCfg.KmsKeyName,
},
}

if dataflowCfg.CustomClassName != "" && dataflowCfg.CustomJarPath != "" {
launchParameters.Parameters["transformationJarPath"] = dataflowCfg.CustomJarPath
launchParameters.Parameters["transformationClassName"] = dataflowCfg.CustomClassName
launchParameters.Parameters["transformationCustomParameters"] = dataflowCfg.CustomParameter
} else if (dataflowCfg.CustomClassName != "" && dataflowCfg.CustomJarPath == "") || (dataflowCfg.CustomClassName == "" && dataflowCfg.CustomJarPath != "") {
return internal.DataflowOutput{}, fmt.Errorf("specify both the custom class name and custom jar GCS path, or specify neither")
}

req := &dataflowpb.LaunchFlexTemplateRequest{
ProjectId: dataflowProjectId,
LaunchParameter: launchParameters,
Expand Down Expand Up @@ -836,6 +848,9 @@ func CreateStreamingConfig(pl profiles.DataShard) StreamingCfg {
AdditionalUserLabels: inputDataflowConfig.AdditionalUserLabels,
KmsKeyName: inputDataflowConfig.KmsKeyName,
GcsTemplatePath: inputDataflowConfig.GcsTemplatePath,
CustomJarPath: inputDataflowConfig.CustomJarPath,
CustomClassName: inputDataflowConfig.CustomClassName,
CustomParameter: inputDataflowConfig.CustomParameter,
}
//create src and dst datastream from pl receiver object
datastreamCfg := DatastreamCfg{
Expand Down
92 changes: 92 additions & 0 deletions streaming/streaming_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -431,3 +431,95 @@ func TestGetUpdateDataStreamLRORetryBackoff(t *testing.T) {
assert.Equal(t, backoff.MaxInterval, DEFAULT_DATASTREAM_LRO_POLL_MAX_DELAY)
assert.Equal(t, backoff.MaxElapsedTime, DEFAULT_DATASTREAM_LRO_POLL_MAX_ELAPSED_TIME)
}

func TestCreateStreamingConfig(t *testing.T) {
// Mock data
inputDataflowConfig := profiles.DataflowConfig{
ProjectId: "project-id",
Location: "us-central1",
Network: "network",
VpcHostProjectId: "vpc-host-project-id",
Subnetwork: "subnetwork",
MaxWorkers: "10",
NumWorkers: "5",
ServiceAccountEmail: "service-account-email",
MachineType: "machine-type",
AdditionalUserLabels: "",
KmsKeyName: "kms-key-name",
GcsTemplatePath: "gcs-template-path",
CustomJarPath: "custom-jar-path",
CustomClassName: "custom-class-name",
CustomParameter: "custom-parameter",
}

inputDatastreamConfig := profiles.DatastreamConfig{
MaxConcurrentBackfillTasks: "3",
MaxConcurrentCdcTasks: "5",
}

inputSrcConnProfile := profiles.DatastreamConnProfileSource{
Location: "us-central1",
Name: "src-conn-profile",
}

inputDstConnProfile := profiles.DatastreamConnProfileTarget{
Name: "dst-conn-profile",
Location: "us-central1",
}

inputGcsConfig := profiles.GcsConfig{
TtlInDays: 7,
TtlInDaysSet: true,
}

inputDataShard := profiles.DataShard{
DataflowConfig: inputDataflowConfig,
StreamLocation: "us-central1",
DatastreamConfig: inputDatastreamConfig,
SrcConnectionProfile: inputSrcConnProfile,
DstConnectionProfile: inputDstConnProfile,
GcsConfig: inputGcsConfig,
TmpDir: "gs://my-bucket/tmp/",
DataShardId: "data-shard-id",
}

// Expected output
expectedStreamingCfg := StreamingCfg{
DatastreamCfg: DatastreamCfg{
StreamLocation: "us-central1",
MaxConcurrentBackfillTasks: "3",
MaxConcurrentCdcTasks: "5",
SourceConnectionConfig: SrcConnCfg{Location: "us-central1", Name: "src-conn-profile"},
DestinationConnectionConfig: DstConnCfg{Name: "dst-conn-profile", Location: "us-central1"},
},
GcsCfg: GcsCfg{
TtlInDays: 7,
TtlInDaysSet: true,
},
DataflowCfg: DataflowCfg{
ProjectId: "project-id",
Location: "us-central1",
Network: "network",
VpcHostProjectId: "vpc-host-project-id",
Subnetwork: "subnetwork",
MaxWorkers: "10",
NumWorkers: "5",
ServiceAccountEmail: "service-account-email",
MachineType: "machine-type",
AdditionalUserLabels: "",
KmsKeyName: "kms-key-name",
GcsTemplatePath: "gcs-template-path",
CustomJarPath: "custom-jar-path",
CustomClassName: "custom-class-name",
CustomParameter: "custom-parameter",
},
TmpDir: "gs://my-bucket/tmp/",
DataShardId: "data-shard-id",
}

// Call function
actualStreamingCfg := CreateStreamingConfig(inputDataShard)

// Compare expected and actual output
assert.Equal(t, expectedStreamingCfg, actualStreamingCfg, "The streaming configuration should match the expected configuration")
}
2 changes: 1 addition & 1 deletion ui/dist/ui/index.html
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion ui/dist/ui/main.d5cd3b3eb1566c78.js → ui/dist/ui/main.afdaa2bee8c4a595.js
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions ui/src/app/app.constants.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,9 @@ export const Dataflow = {
Location: 'dataflowLocation',
GcsTemplatePath: 'gcsTemplatePath',
IsDataflowConfigSet: 'isDataflowConfigSet',
CustomJarPath: 'customJarPath',
CustomClassName: 'customClassName',
CustomParameter: 'customParameter'
}

export const ColLength = {
Expand Down
25 changes: 25 additions & 0 deletions ui/src/app/components/dataflow-form/dataflow-form.component.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,31 @@
<h2>Tune Dataflow (Optional)</h2>
<h5>This form is optional and should only be edited to tune runtime environment for Dataflow.</h5>

<mat-expansion-panel>
<mat-expansion-panel-header class="mat-row" matTooltip="Edit to run custom transformations">
<span>Custom Transformations</span>
</mat-expansion-panel-header>
<div>
<mat-form-field class="full-width" appearance="outline" matTooltip="Specify the GCS path for custom jar"
[matTooltipPosition]="'right'">
<mat-label>Custom JAR GCS Path</mat-label>
<input matInput placeholder="Custom JAR GCS Path" type="text" formControlName="customJarPath"/>
</mat-form-field>
<br>
<mat-form-field class="full-width" appearance="outline" matTooltip="Specify the fully classified class name"
[matTooltipPosition]="'right'">
<mat-label>Custom Class Name</mat-label>
<input matInput placeholder="Fully Classified Class Name" type="text" formControlName="customClassName"/>
</mat-form-field>
<br>
<mat-form-field class="full-width" appearance="outline" matTooltip="Specify the custom parameter required by the JAR"
[matTooltipPosition]="'right'">
<mat-label>Custom Parameter</mat-label>
<input matInput placeholder="Custom Parameters" type="text" formControlName="customParameter"/>
</mat-form-field>
</div>
</mat-expansion-panel>
Comment on lines +6 to +29

@entelligence-ai-pr-reviews entelligence-ai-pr-reviews Bot Apr 1, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The formControlName attributes are used but there's no parent <form> or <ng-form> element wrapping these form fields, which will cause Angular to throw errors.

<br>
<mat-expansion-panel>
<mat-expansion-panel-header class="mat-row" matTooltip="Edit to run Dataflow in a VPC">
<span>Networking</span>
Expand Down
Loading