diff --git a/go/adbc/driver/salesforce/connection.go b/go/adbc/driver/salesforce/connection.go index ce5b7f2d99..a89ccd3911 100644 --- a/go/adbc/driver/salesforce/connection.go +++ b/go/adbc/driver/salesforce/connection.go @@ -144,7 +144,7 @@ func (c *connectionImpl) setupRequired(authType string) error { // Finalizes the connectionImpl fields for the authentication flow func (c *connectionImpl) finalize(ctx context.Context, config *api.AuthConfig) error { - c.client = api.NewClient(config, c.version) + c.client = api.NewClientWithLogger(config, c.version, c.Logger.WithGroup("client")) // Authenticate and get token err := c.client.Authenticate(ctx) diff --git a/go/adbc/driver/salesforce/driver.go b/go/adbc/driver/salesforce/driver.go index 4cfa3e1f28..9f2e6fbe9b 100644 --- a/go/adbc/driver/salesforce/driver.go +++ b/go/adbc/driver/salesforce/driver.go @@ -52,6 +52,8 @@ const ( // Stmt options OptionStringDLOCategory = "adbc.salesforce.dc.dlo.category" OptionStringDLOPrimaryKey = "adbc.salesforce.dc.dlo.primary_key" + OptionStringDLOMaterialized = "adbc.salesforce.dc.dlo.materialized" + OptionStringDLOWriteMode = "adbc.salesforce.dc.dlo.write_mode" OptionsStringTargetDLO = "adbc.salesforce.dc.dlo.target_dlo" OptionIntDataTransformRunTimeout = "adbc.salesforce.dc.data_transform_run_timeout" diff --git a/go/adbc/driver/salesforce/gosalesforce/api/client.go b/go/adbc/driver/salesforce/gosalesforce/api/client.go index f48d4cfe1c..ec07619991 100644 --- a/go/adbc/driver/salesforce/gosalesforce/api/client.go +++ b/go/adbc/driver/salesforce/gosalesforce/api/client.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "io" + "log/slog" "math/rand" "net/http" "net/url" @@ -24,10 +25,16 @@ type Client struct { accessToken *Token cdpToken *Token version string + + logger *slog.Logger } -// NewClient creates a new authentication client func NewClient(config *AuthConfig, version string) *Client { + return NewClientWithLogger(config, version, nil) +} + +// NewClient creates a new authentication client +func NewClientWithLogger(config *AuthConfig, version string, logger *slog.Logger) *Client { if config == nil { config = DefaultAuthConfig() } @@ -36,10 +43,15 @@ func NewClient(config *AuthConfig, version string) *Client { Timeout: config.Timeout, } + if logger == nil { + logger = slog.Default() + } + return &Client{ config: config, httpClient: httpClient, version: version, + logger: logger, } } @@ -92,7 +104,7 @@ func (c *Client) executeHTTPRequest(ctx context.Context, req *http.Request) (*ht // handleErrorResponse handles common error response patterns func handleErrorResponse(statusCode int, body []byte, errorType string) error { - var errorResp map[string]interface{} + var errorResp map[string]any if json.Unmarshal(body, &errorResp) == nil { if errorMsg, ok := errorResp["message"].(string); ok { return &SfdcError{ @@ -527,7 +539,7 @@ func (c *Client) requestAccessToken(ctx context.Context, tokenURL string, data u if resp.StatusCode != http.StatusOK { // Handle special case for token endpoint error format - var errorResp map[string]interface{} + var errorResp map[string]any if json.Unmarshal(body, &errorResp) == nil { if errorMsg, ok := errorResp["error_description"].(string); ok { return nil, &SfdcError{ diff --git a/go/adbc/driver/salesforce/gosalesforce/api/http.go b/go/adbc/driver/salesforce/gosalesforce/api/http.go index 16c6e2f0f1..a6266c4432 100644 --- a/go/adbc/driver/salesforce/gosalesforce/api/http.go +++ b/go/adbc/driver/salesforce/gosalesforce/api/http.go @@ -49,6 +49,40 @@ func PostJSON[RequestType any, ResponseType any](c *Client, ctx context.Context, return executeJSONRequest[ResponseType](c, ctx, req) } +// PutJSON performs a PUT request with JSON request/response handling +// RequestType is the request payload type, ResponseType is the response payload type +func PutJSON[RequestType any, ResponseType any](c *Client, ctx context.Context, path string, request *RequestType) (*ResponseType, error) { + // Marshal request to JSON + reqBody, err := json.Marshal(request) + if err != nil { + return nil, fmt.Errorf("failed to marshal request: %w", err) + } + + req, err := createRequest(c, ctx, "PUT", path, strings.NewReader(string(reqBody))) + if err != nil { + return nil, err + } + + return executeJSONRequest[ResponseType](c, ctx, req) +} + +// PatchJSON performs a PATCH request with JSON request/response handling +// RequestType is the request payload type, ResponseType is the response payload type +func PatchJSON[RequestType any, ResponseType any](c *Client, ctx context.Context, path string, request *RequestType) (*ResponseType, error) { + // Marshal request to JSON + reqBody, err := json.Marshal(request) + if err != nil { + return nil, fmt.Errorf("failed to marshal request: %w", err) + } + + req, err := createRequest(c, ctx, "PATCH", path, strings.NewReader(string(reqBody))) + if err != nil { + return nil, err + } + + return executeJSONRequest[ResponseType](c, ctx, req) +} + // DeleteJSON performs a DELETE request func DeleteJSON(c *Client, ctx context.Context, path string) error { req, err := createRequest(c, ctx, "DELETE", path, nil) diff --git a/go/adbc/driver/salesforce/gosalesforce/api/transform.go b/go/adbc/driver/salesforce/gosalesforce/api/transform.go index d6b3787b3b..5ce036efd8 100644 --- a/go/adbc/driver/salesforce/gosalesforce/api/transform.go +++ b/go/adbc/driver/salesforce/gosalesforce/api/transform.go @@ -2,6 +2,7 @@ package api import ( "context" + "errors" "fmt" ) @@ -28,6 +29,53 @@ func (c *Client) CreateDataTransform(ctx context.Context, request *CreateDataTra return PostJSON[CreateDataTransformRequest, DataTransform](c, ctx, "data-transforms", request) } +// ValidateDataTransform creates a new data transform in Data Cloud +// reference: https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=createDataTransform +func (c *Client) ValidateDataTransform(ctx context.Context, request *CreateDataTransformRequest) (*DataTransformValidation, error) { + // Validate required fields + if request.Name == "" { + return nil, &SfdcError{ + Code: 400, + Message: "Data transform name cannot be empty", + Type: "invalid_request", + } + } + + if request.Label == "" { + return nil, &SfdcError{ + Code: 400, + Message: "Data transform label cannot be empty", + Type: "invalid_request", + } + } + + return PostJSON[CreateDataTransformRequest, DataTransformValidation](c, ctx, "data-transforms-validation", request) +} + +// UpdateDataTransform creates a new data transform in Data Cloud +// reference: https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=updateDataTransform +func (c *Client) UpdateDataTransform(ctx context.Context, request *CreateDataTransformRequest) (*DataTransform, error) { + // Validate required fields + if request.Name == "" { + return nil, &SfdcError{ + Code: 400, + Message: "Data transform name cannot be empty", + Type: "invalid_request", + } + } + + if request.Label == "" { + return nil, &SfdcError{ + Code: 400, + Message: "Data transform label cannot be empty", + Type: "invalid_request", + } + } + + path := fmt.Sprintf("data-transforms/%s", request.Name) + return PutJSON[CreateDataTransformRequest, DataTransform](c, ctx, path, request) +} + // CreateDataTransform creates a new data transform in Data Cloud // reference: https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=getDataTransform func (c *Client) GetDataTransform(ctx context.Context, dataTransformNameOrId string) (*DataTransform, error) { @@ -89,7 +137,21 @@ func (c *Client) RefreshDataTransformStatus(ctx context.Context, dataTransformNa } path := fmt.Sprintf("data-transforms/%s/actions/refresh-status", dataTransformNameOrId) - return PostJSON[interface{}, DataCloudActionResponse](c, ctx, path, nil) + return PostJSON[any, DataCloudActionResponse](c, ctx, path, nil) +} + +func actionMustSucceed(resp *DataCloudActionResponse, err error) error { + if resp != nil && !resp.Success { + return errors.Join( + fmt.Errorf("action unsuccessful due to: %v", resp.Errors), + err, + ) + } + return err +} + +func (c *Client) MustRefreshDataTransformStatus(ctx context.Context, dataTransformNameOrId string) error { + return actionMustSucceed(c.RefreshDataTransformStatus(ctx, dataTransformNameOrId)) } // RunDataTransform runs a data transform @@ -105,7 +167,10 @@ func (c *Client) RunDataTransform(ctx context.Context, dataTransformNameOrId str } path := fmt.Sprintf("data-transforms/%s/actions/run", dataTransformNameOrId) - return PostJSON[interface{}, DataCloudActionResponse](c, ctx, path, nil) + return PostJSON[any, DataCloudActionResponse](c, ctx, path, nil) +} +func (c *Client) MustRunDataTransform(ctx context.Context, dataTransformNameOrId string) error { + return actionMustSucceed(c.RunDataTransform(ctx, dataTransformNameOrId)) } // CancelDataTransform cancels a data transform @@ -121,7 +186,10 @@ func (c *Client) CancelDataTransform(ctx context.Context, dataTransformNameOrId } path := fmt.Sprintf("data-transforms/%s/actions/cancel", dataTransformNameOrId) - return PostJSON[interface{}, DataCloudActionResponse](c, ctx, path, nil) + return PostJSON[any, DataCloudActionResponse](c, ctx, path, nil) +} +func (c *Client) MustCancelDataTransform(ctx context.Context, dataTransformNameOrId string) error { + return actionMustSucceed(c.CancelDataTransform(ctx, dataTransformNameOrId)) } // DeleteDataTransform deletes a data transform @@ -157,12 +225,13 @@ func NewBatchDataTransformRequest(name, label string, nodes map[string]DbtDataTr } // NewDbtDataTransformNode creates a new dbt-style data transform node -func NewDbtDataTransformNode(name, relationName, compiledCode string, materialized string, dependsOn map[string]interface{}) DbtDataTransformNode { +func NewDbtDataTransformNode(name, relationName, compiledCode string, materialized, writeMode string, dependsOn map[string]any) DbtDataTransformNode { return DbtDataTransformNode{ Name: name, RelationName: relationName, Config: DbtDataTransformNodeConfig{ Materialized: materialized, + WriteMode: writeMode, }, CompiledCode: compiledCode, DependsOn: dependsOn, @@ -180,6 +249,6 @@ func NewSimpleDbtDataTransformNode(name, relationName, sql string) DbtDataTransf Materialized: "table", }, CompiledCode: sql, - DependsOn: make(map[string]interface{}), + DependsOn: make(map[string]any), } } diff --git a/go/adbc/driver/salesforce/gosalesforce/api/types.go b/go/adbc/driver/salesforce/gosalesforce/api/types.go index 6c21825c2e..a43d7fbe67 100644 --- a/go/adbc/driver/salesforce/gosalesforce/api/types.go +++ b/go/adbc/driver/salesforce/gosalesforce/api/types.go @@ -77,14 +77,14 @@ type SqlQueryRequest struct { // SqlParameter represents a parameter in a SQL query type SqlParameter struct { - Type string `json:"type"` - Name string `json:"name"` - Value interface{} `json:"value"` + Type string `json:"type"` + Name string `json:"name"` + Value any `json:"value"` } // SqlQueryResponse represents the response from a SQL query type SqlQueryResponse struct { - Data [][]interface{} `json:"data"` + Data [][]any `json:"data"` Metadata []SqlQueryMetadata `json:"metadata"` Status SqlQueryStatus `json:"status"` ReturnedRows int64 `json:"returnedRows"` @@ -162,7 +162,7 @@ type QueryV2Request struct { // QueryV2Response represents the response from the v2 query API // reference: https://developer.salesforce.com/docs/data/data-cloud-query-guide/references/data-cloud-query-api-reference/c360a-api-query-v2.html type QueryV2Response struct { - Data [][]interface{} `json:"data,omitempty"` + Data [][]any `json:"data,omitempty"` Metadata map[string]QueryV2Metadata `json:"metadata,omitempty"` Done bool `json:"done"` NextBatchId *string `json:"nextBatchId,omitempty"` @@ -170,7 +170,7 @@ type QueryV2Response struct { QueryId string `json:"queryId,omitempty"` StartTime string `json:"startTime,omitempty"` EndTime string `json:"endTime,omitempty"` - ArrowStream interface{} `json:"arrowStream,omitempty"` + ArrowStream any `json:"arrowStream,omitempty"` } // QueryV2Metadata represents metadata for a v2 query result column @@ -191,7 +191,7 @@ type MetadataEntity struct { DisplayName string `json:"displayName"` Category string `json:"category,omitempty"` Fields []MetadataField `json:"fields,omitempty"` - Indexes []interface{} `json:"indexes,omitempty"` + Indexes []any `json:"indexes,omitempty"` Relationships []MetadataRelationship `json:"relationships,omitempty"` PrimaryKeys []MetadataPrimaryKey `json:"primaryKeys,omitempty"` ReferenceModelEntityDeveloperName string `json:"referenceModelEntityDeveloperName,omitempty"` @@ -321,22 +321,50 @@ const ( DataTransformStatusProcessing DataTransformStatus = "Processing" ) -// DataTransformLastRunStatus represents the status of the last run -type DataTransformLastRunStatus string +// Though it's documented that the status value is Camel case +// see `status` from https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=getDataTransform +// it's actually returned as all upper case + +// helper +func (s DataTransformStatus) isEq(other DataTransformStatus) bool { + return strings.EqualFold(string(s), string(other)) +} + +func (s DataTransformStatus) IsActive() bool { return s.isEq(DataTransformStatusActive) } +func (s DataTransformStatus) IsError() bool { return s.isEq(DataTransformStatusError) } +func (s DataTransformStatus) IsProcessing() bool { return s.isEq(DataTransformStatusProcessing) } +func (s DataTransformStatus) IsDeleting() bool { return s.isEq(DataTransformStatusDeleting) } + +// DataTransformRunStatus represents the status of the last run +type DataTransformRunStatus string // TODO: revisit these values, the actual returns are different from the documentation // see LastRunStatus from https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=getDataTransform const ( - DataTransformLastRunStatusCanceled DataTransformLastRunStatus = "Canceled" - DataTransformLastRunStatusFailure DataTransformLastRunStatus = "Failure" - DataTransformLastRunStatusInProgress DataTransformLastRunStatus = "In_Progress" - DataTransformLastRunStatusNone DataTransformLastRunStatus = "None" - DataTransformLastRunStatusPartialFailure DataTransformLastRunStatus = "PartialFailure" - DataTransformLastRunStatusPartiallyCanceled DataTransformLastRunStatus = "PartiallyCanceled" - DataTransformLastRunStatusPending DataTransformLastRunStatus = "Pending" - DataTransformLastRunStatusSuccess DataTransformLastRunStatus = "Success" + RunStatusCanceled DataTransformRunStatus = "Canceled" + RunStatusFailure DataTransformRunStatus = "Failure" + RunStatusInProgress DataTransformRunStatus = "In_Progress" + RunStatusNone DataTransformRunStatus = "None" + RunStatusPartialFailure DataTransformRunStatus = "PartialFailure" + RunStatusPartiallyCanceled DataTransformRunStatus = "PartiallyCanceled" + RunStatusPending DataTransformRunStatus = "Pending" + RunStatusSuccess DataTransformRunStatus = "Success" ) +// helper +func (s DataTransformRunStatus) isEq(other DataTransformRunStatus) bool { + return strings.EqualFold(string(s), string(other)) +} + +func (s DataTransformRunStatus) IsCanceled() bool { return s.isEq(RunStatusCanceled) } +func (s DataTransformRunStatus) IsFailure() bool { return s.isEq(RunStatusFailure) } +func (s DataTransformRunStatus) IsInProgress() bool { return s.isEq(RunStatusInProgress) } +func (s DataTransformRunStatus) IsNone() bool { return s.isEq(RunStatusNone) } +func (s DataTransformRunStatus) IsPartialFailure() bool { return s.isEq(RunStatusPartialFailure) } +func (s DataTransformRunStatus) IsPartiallyCanceled() bool { return s.isEq(RunStatusPartiallyCanceled) } +func (s DataTransformRunStatus) IsPending() bool { return s.isEq(RunStatusPending) } +func (s DataTransformRunStatus) IsSuccess() bool { return s.isEq(RunStatusSuccess) } + // CreateDataTransformRequest represents a request to create a data transform type CreateDataTransformRequest struct { CreationType DataTransformCreationType `json:"creationType,omitempty"` @@ -350,13 +378,18 @@ type CreateDataTransformRequest struct { Type DataTransformType `json:"type"` } +type DataTransformValidation struct { + Issues []DataCloudError `json:"issues,omitempty"` + OutputDataObjects map[string][]DataTransformOutputDataObject `json:"outputDataObjects,omitempty"` +} + // DataTransformDefinition represents the base definition of a data transform // This only supports BATCH data transform type DataTransformDefinition struct { Type DataTransformDefinitionType `json:"type"` Version string `json:"version"` // This feature is not publicly available yet; be cautious of possible breaking changes in the future. - Manifest DbtDataTransformDefinition `json:"manifest,omitempty"` + Manifest DbtDataTransformDefinition `json:"manifest"` OutputDataObjects []DataTransformOutputDataObject `json:"outputDataObjects,omitempty"` } @@ -367,11 +400,24 @@ type DbtDataTransformDefinition struct { // Information about the objects into which the data transform writes the transformed data. // See `outputDataObjects` from reference: https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=type:Batch+Data+Transform+Output type DataTransformOutputDataObject struct { + Type string `json:"type"` Name string `json:"name"` + Label string `json:"label,omitempty"` + Category string `json:"category,omitempty"` Namespace string `json:"namespace,omitempty"` + + Fields []DataObjectField `json:"fields"` // TODO: There are more fields known to be missing, add if necessary } +type DataObjectField struct { + Name string `json:"name"` + KeyQualifierFieldName string `json:"keyQualifierFieldName,omitempty"` + Label string `json:"label"` + Type string `json:"type"` + IsPrimaryKey bool `json:"isPrimaryKey"` +} + // DataTransformNode represents a node in a data transform type DbtDataTransformNode struct { // The name of the node, must match its associated key in the DbtDataTransformDefinition.Nodes map @@ -381,35 +427,36 @@ type DbtDataTransformNode struct { RelationName string `json:"relation_name,omitempty"` Config DbtDataTransformNodeConfig `json:"config"` CompiledCode string `json:"compiled_code"` - DependsOn map[string]interface{} `json:"depends_on,omitempty"` + DependsOn map[string]any `json:"depends_on,omitempty"` } type DbtDataTransformNodeConfig struct { Materialized string `json:"materialized"` + WriteMode string `json:"writeMode,omitzero"` } // DataTransform represents the response from creating a data transform // see Responses from https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=createDataTransform type DataTransform struct { - ActionUrls DataTransformActionUrls `json:"actionUrls,omitempty"` - CreatedBy DataTransformUser `json:"createdBy"` - CreatedDate string `json:"createdDate"` - CreationType DataTransformCreationType `json:"creationType,omitempty"` - Definition DataTransformDefinition `json:"definition"` - ID string `json:"id"` - Label string `json:"label"` - LastModifiedBy DataTransformUser `json:"lastModifiedBy"` - LastModifiedDate string `json:"lastModifiedDate"` - LastRunStatus DataTransformLastRunStatus `json:"lastRunStatus"` - Name string `json:"name"` - Namespace string `json:"namespace,omitempty"` - Status DataTransformStatus `json:"status"` - Type DataTransformType `json:"type"` - URL string `json:"url"` - DataSpaceName string `json:"dataSpaceName,omitempty"` - Description string `json:"description,omitempty"` - LastRunDate string `json:"lastRunDate,omitempty"` - Version int64 `json:"version,omitempty"` + ActionUrls DataTransformActionUrls `json:"actionUrls"` + CreatedBy DataTransformUser `json:"createdBy"` + CreatedDate string `json:"createdDate"` + CreationType DataTransformCreationType `json:"creationType,omitempty"` + Definition DataTransformDefinition `json:"definition"` + ID string `json:"id"` + Label string `json:"label"` + LastModifiedBy DataTransformUser `json:"lastModifiedBy"` + LastModifiedDate string `json:"lastModifiedDate"` + LastRunStatus DataTransformRunStatus `json:"lastRunStatus"` + Name string `json:"name"` + Namespace string `json:"namespace,omitempty"` + Status DataTransformStatus `json:"status"` + Type DataTransformType `json:"type"` + URL string `json:"url"` + DataSpaceName string `json:"dataSpaceName,omitempty"` + Description string `json:"description,omitempty"` + LastRunDate string `json:"lastRunDate,omitempty"` + Version int64 `json:"version,omitempty"` } type DataTransformList struct { @@ -539,7 +586,7 @@ type DataLakeObjects struct { // DataLakeObject represents the response from creating a Data Lake Object type DataLakeObject struct { - Capabilities map[string]interface{} `json:"capabilities"` + Capabilities map[string]any `json:"capabilities"` Category DataLakeObjectCategory `json:"category"` DataLakeFieldInfoRepresentation []DataLakeFieldOutput `json:"dataLakeFieldInfoRepresentation"` DataSpaceInfo []DataSpaceObject `json:"dataSpaceInfo"` @@ -552,9 +599,9 @@ type DataLakeObject struct { EventDateTimeFieldName string `json:"eventDateTimeFieldName,omitempty"` OrgUnitIdentifierFieldName string `json:"orgUnitIdentifierFieldName,omitempty"` RecordModifiedFieldName string `json:"recordModifiedFieldName,omitempty"` - CreatedBy DataTransformUser `json:"createdBy,omitempty"` + CreatedBy DataTransformUser `json:"createdBy"` CreatedDate string `json:"createdDate,omitempty"` - LastModifiedBy DataTransformUser `json:"lastModifiedBy,omitempty"` + LastModifiedBy DataTransformUser `json:"lastModifiedBy"` LastModifiedDate string `json:"lastModifiedDate,omitempty"` URL string `json:"url,omitempty"` } @@ -585,8 +632,9 @@ type DataSpaceObject struct { // Error response. // reference: https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=type:Data+Cloud+Error type DataCloudError struct { - ErrorCode string `json:"errorCode"` - ErrorMessage string `json:"errorMessage"` + ErrorCode string `json:"errorCode"` + ErrorMessage string `json:"errorMessage"` + ErrorSeverity string `json:"errorSeverity"` } // Data Cloud action response base. @@ -597,36 +645,34 @@ type DataCloudActionResponse struct { } func (s *DataTransform) IsActive() bool { - // Though it's documented that the status value is Camel case - // see `status` from https://developer.salesforce.com/docs/data/connectapi/references/spec?meta=getDataTransform - // it's actually returned as all upper case - return strings.EqualFold(string(s.Status), string(DataTransformStatusActive)) + return s.Status.IsActive() + } func (s *DataTransform) IsError() bool { - return strings.EqualFold(string(s.Status), string(DataTransformStatusError)) + return s.Status.IsError() } func (s *DataTransform) IsLastRunSuccess() bool { - return strings.EqualFold(string(s.LastRunStatus), string(DataTransformLastRunStatusSuccess)) + return strings.EqualFold(string(s.LastRunStatus), string(RunStatusSuccess)) } func (s *DataTransform) IsLastRunFailure() bool { - return strings.EqualFold(string(s.LastRunStatus), string(DataTransformLastRunStatusFailure)) || - strings.EqualFold(string(s.LastRunStatus), string(DataTransformLastRunStatusPartialFailure)) + return strings.EqualFold(string(s.LastRunStatus), string(RunStatusFailure)) || + strings.EqualFold(string(s.LastRunStatus), string(RunStatusPartialFailure)) } func (s *DataTransform) IsLastRunCanceled() bool { - return strings.EqualFold(string(s.LastRunStatus), string(DataTransformLastRunStatusCanceled)) || - strings.EqualFold(string(s.LastRunStatus), string(DataTransformLastRunStatusPartiallyCanceled)) + return strings.EqualFold(string(s.LastRunStatus), string(RunStatusCanceled)) || + strings.EqualFold(string(s.LastRunStatus), string(RunStatusPartiallyCanceled)) } func (s *DataTransform) IsLastRunInProgress() bool { - return strings.EqualFold(string(s.LastRunStatus), string(DataTransformLastRunStatusInProgress)) + return strings.EqualFold(string(s.LastRunStatus), string(RunStatusInProgress)) } func (s *DataTransform) IsLastRunPending() bool { - return strings.EqualFold(string(s.LastRunStatus), string(DataTransformLastRunStatusPending)) + return strings.EqualFold(string(s.LastRunStatus), string(RunStatusPending)) } // DataStreamType represents the type of data stream @@ -765,7 +811,7 @@ type DataStream struct { URL string `json:"url,omitempty"` // Data stream specific fields - AdvancedAttributes map[string]interface{} `json:"advancedAttributes,omitempty"` + AdvancedAttributes map[string]any `json:"advancedAttributes,omitempty"` ConnectorInfo DataStreamConnectorInfo `json:"connectorInfo"` DataAccessMode *DataAccessMode `json:"dataAccessMode,omitempty"` DataLakeObjectInfo DataLakeObject `json:"dataLakeObjectInfo"` diff --git a/go/adbc/driver/salesforce/gosalesforce/api/wrappers.go b/go/adbc/driver/salesforce/gosalesforce/api/wrappers.go index 2d09a0fabd..8a233b26c6 100644 --- a/go/adbc/driver/salesforce/gosalesforce/api/wrappers.go +++ b/go/adbc/driver/salesforce/gosalesforce/api/wrappers.go @@ -41,7 +41,7 @@ func (client *Client) DeleteIfDloExists(ctx context.Context, name string) error exponentialBackOff.InitialInterval = INITIAL_INTERVAL exponentialBackOff.MaxInterval = MAX_INTERVAL - operation := func() (interface{}, error) { + operation := func() (any, error) { _, err := client.GetDataLakeObjectByName(ctx, name) if err != nil { return nil, nil @@ -84,7 +84,7 @@ func (client *Client) DeleteDataTransformIfExists(ctx context.Context, name stri exponentialBackOff.InitialInterval = INITIAL_INTERVAL exponentialBackOff.MaxInterval = MAX_INTERVAL - operation := func() (interface{}, error) { + operation := func() (any, error) { _, err := client.GetDataTransform(ctx, name) if err != nil { // Data Transform doesn't exist, deletion complete or not needed @@ -166,7 +166,7 @@ func (client *Client) CreateDataLakeObjectWithInferredSchema(ctx context.Context exponentialBackOff.InitialInterval = INITIAL_INTERVAL exponentialBackOff.MaxInterval = MAX_INTERVAL - waitForActiveOp := func() (interface{}, error) { + waitForActiveOp := func() (any, error) { currentDLO, err := client.GetDataLakeObject(ctx, dataLakeObject.Name, nil, nil, "") if err != nil { return nil, fmt.Errorf("failed to get DLO status: %w", err) @@ -236,7 +236,7 @@ func (client *Client) TriggerDbtBatchDataTransform(ctx context.Context, targetDl exponentialBackOff := backoff.NewExponentialBackOff() // Waits for the data transform to be active - waitForActiveOp := func() (interface{}, error) { + waitForActiveOp := func() (any, error) { // Eagerly refreshes status, otherwise `client.GetDataTransform` may respond with a stale status refreshStatusResponse, err := client.RefreshDataTransformStatus(ctx, dataTransform.Name) if err != nil { @@ -279,7 +279,7 @@ func (client *Client) TriggerDbtBatchDataTransform(ctx context.Context, targetDl } // Waits for the data transform run to be success - waitForRunOp := func() (interface{}, error) { + waitForRunOp := func() (any, error) { refreshStatusResponse, err := client.RefreshDataTransformStatus(ctx, dataTransform.Name) if !refreshStatusResponse.Success { return nil, fmt.Errorf("failed to refresh the data transform status [%v]", refreshStatusResponse.Errors) @@ -323,6 +323,126 @@ func (client *Client) TriggerDbtBatchDataTransform(ctx context.Context, targetDl return dataTransform, nil } +// Polls the data transform status until it is no longer processing. +// Returns the latest data transform received from the server. +func (client *Client) WaitForDataTransform(ctx context.Context, dt *DataTransform) (*DataTransform, error) { + client.logger.DebugContext(ctx, "WaitForDataTransform") + + op := func() (*DataTransform, error) { + ndt, err := client.GetDataTransform(ctx, dt.Name) + if err != nil { + return nil, backoff.Permanent(err) + } + + if ndt.Status.IsProcessing() { + return ndt, fmt.Errorf("still processing") + } + + return ndt, nil + } + + policy := backoff.NewExponentialBackOff() + policy.InitialInterval = 1 * time.Second + policy.MaxInterval = 15 * time.Second + + return backoff.Retry( + ctx, + op, + backoff.WithBackOff(policy), + backoff.WithMaxElapsedTime(MAX_ELAPSED_TIME), + backoff.WithNotify(func(err error, duration time.Duration) { + client.logger.DebugContext(ctx, "waiting", "err", err, "duration", duration) + }), + ) +} + +// Polls the data transform last run status until it is no longer pending or in-progress. +// Returns the latest data transform received from the server. +func (client *Client) WaitForDataTransformRun(ctx context.Context, dt *DataTransform, runTimeout time.Duration) (*DataTransform, error) { + client.logger.DebugContext(ctx, "WaitForDataTransformRun", "timeout", runTimeout) + op := func() (*DataTransform, error) { + err := client.MustRefreshDataTransformStatus(ctx, dt.Name) + if err != nil { + // return nil, backoff.Permanent(err) + return nil, err + } + + ndt, err := client.GetDataTransform(ctx, dt.Name) + if err != nil { + return nil, backoff.Permanent(err) + } + + switch { + case ndt.LastRunStatus.IsPending(): + return ndt, fmt.Errorf("run pending") + case ndt.LastRunStatus.IsInProgress(): + return ndt, fmt.Errorf("run in progress") + } + + return ndt, nil + } + + policy := backoff.NewExponentialBackOff() + policy.InitialInterval = 30 * time.Second + policy.MaxInterval = 15 * time.Second + + return backoff.Retry( + ctx, + op, + backoff.WithBackOff(policy), + backoff.WithMaxElapsedTime(runTimeout), + ) +} + +func (client *Client) CreateOrUpdateDataTransform(ctx context.Context, req *CreateDataTransformRequest) (*DataTransform, error) { + l := client.logger.With("operation", "CreateOrUpdateDataTransform") + + l.DebugContext(ctx, "checking DT exists") + + // naively check if the data-transform already exists + // TODO: There there are errors that can occur if the data transform exists (i.e. rate limiting, transient server issue, etc) + if _, err := client.GetDataTransform(ctx, req.Name); err != nil { + // TODO: What if existing.Status == DELETING | PROGRESSING + l.DebugContext(ctx, "creating data transform", "name", req.Name) + return client.CreateDataTransform(ctx, req) + } else { + l.DebugContext(ctx, "updating data transform", "name", req.Name) + return client.UpdateDataTransform(ctx, req) + } + +} + +type DataSpaceMember struct { + Name string `json:"memberName"` + Filter *FilterConfig `json:"filter,omitzero"` +} + +func (client *Client) UpsertDataSpaceMembers(ctx context.Context, dataSpace string, members []DataSpaceMember) (*DataCloudActionResponse, error) { + type upsertDataspaceMemberBody struct { + Members struct { + Members []DataSpaceMember `json:"members"` + } `json:"members"` + } + + type upsertDataspaceMemberResp struct { + *DataCloudActionResponse + Members struct { + Members []DataSpaceMember `json:"members"` + } `json:"dataSpaceMembers"` + } + + var reqBody upsertDataspaceMemberBody + reqBody.Members.Members = members + + path := fmt.Sprintf("data-spoaces/%s/members", dataSpace) + resp, err := PutJSON[upsertDataspaceMemberBody, upsertDataspaceMemberResp](client, ctx, path, &reqBody) + if err != nil { + return nil, err + } + + return resp.DataCloudActionResponse, nil +} + // NewClientWithJWT creates a new client using JWT authentication // It expects the private key to be stored in the home directory at `~/.salesforce/JWT/server.key` // It expects the login URL to be stored in the environment variable `SALESFORCE_LOGIN_URL` diff --git a/go/adbc/driver/salesforce/gosalesforce/examples/metadata/main.go b/go/adbc/driver/salesforce/gosalesforce/examples/metadata/main.go index 0c3496a62c..d93ed752da 100644 --- a/go/adbc/driver/salesforce/gosalesforce/examples/metadata/main.go +++ b/go/adbc/driver/salesforce/gosalesforce/examples/metadata/main.go @@ -40,10 +40,7 @@ func demonstrateMetadata(client *api.Client) { // Display a summary of entities if len(metadataResp.Metadata) > 0 { fmt.Printf("\nFirst 10 entities:\n") - maxDisplay := len(metadataResp.Metadata) - if maxDisplay > 10 { - maxDisplay = 10 - } + maxDisplay := min(len(metadataResp.Metadata), 10) for i := 0; i < maxDisplay; i++ { entity := metadataResp.Metadata[i] @@ -84,10 +81,7 @@ func demonstrateMetadata(client *api.Client) { if len(firstEntity.Fields) > 0 { fmt.Printf("\nFields (%d):\n", len(firstEntity.Fields)) - maxFields := len(firstEntity.Fields) - if maxFields > 5 { - maxFields = 5 - } + maxFields := min(len(firstEntity.Fields), 5) for i := 0; i < maxFields; i++ { field := firstEntity.Fields[i] nullable := "NOT NULL" @@ -107,10 +101,7 @@ func demonstrateMetadata(client *api.Client) { if len(firstEntity.Relationships) > 0 { fmt.Printf("\nRelationships (%d):\n", len(firstEntity.Relationships)) - maxRels := len(firstEntity.Relationships) - if maxRels > 3 { - maxRels = 3 - } + maxRels := min(len(firstEntity.Relationships), 3) for i := 0; i < maxRels; i++ { rel := firstEntity.Relationships[i] fmt.Printf(" %s.%s -> %s.%s (%s)\n", diff --git a/go/adbc/driver/salesforce/statement.go b/go/adbc/driver/salesforce/statement.go index 447928b728..34d0ff59d0 100644 --- a/go/adbc/driver/salesforce/statement.go +++ b/go/adbc/driver/salesforce/statement.go @@ -18,8 +18,11 @@ package salesforce import ( + "cmp" "context" "fmt" + "log/slog" + "slices" "time" "github.com/apache/arrow-adbc/go/adbc" @@ -42,6 +45,7 @@ type statement struct { // Create DLO options dloCategory string dloPrimaryKey string + dloWriteMode string // Data Transform options targetDLO string @@ -84,8 +88,12 @@ func (s *statement) executeSQLQuery(ctx context.Context) (array.RecordReader, in } } + logger := s.cnxn.Logger.With("operation", "executeSQLQuery") + // This is supposed to be equivalent to `CREATE OR REPLACE TABLE` if s.dloCategory != "" && s.dloPrimaryKey != "" && s.targetDLO != "" { + logger := logger.With(slog.Group("opt", "dloCategory", s.dloCategory, "dloPrimaryKey", s.dloPrimaryKey, "targetDLO", s.targetDLO, "dataSpace", s.cnxn.dataSpace)) + if s.cnxn.dataSpace == "" { return nil, 0, adbc.Error{ Code: adbc.StatusInvalidState, @@ -93,36 +101,86 @@ func (s *statement) executeSQLQuery(ctx context.Context) (array.RecordReader, in } } - // Delete the existing DLO - err := s.cnxn.client.DeleteIfDloExists(ctx, s.targetDLO) + logger.DebugContext(ctx, "Writing sql to DT...") + // Creates a data transform + req := api.NewBatchDataTransformRequest( + s.targetDLO, + s.targetDLO, + map[string]api.DbtDataTransformNode{ + "node": api.NewDbtDataTransformNode( + "node", + s.targetDLO, + s.query, + "TABLE", + s.dloWriteMode, + nil, + ), + }, + ) + + logger.DebugContext(ctx, "Validating batch DT", "req", req) + + valid, err := s.cnxn.client.ValidateDataTransform(ctx, req) if err != nil { - return nil, 0, adbc.Error{ - Code: adbc.StatusInternal, - Msg: err.Error(), - } + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "failed to validate the data transform for create/update: %v", err) + } + logger.DebugContext(ctx, "Validated", "issues", valid.Issues, "odo", valid.OutputDataObjects) + + odo := slices.Clone(valid.OutputDataObjects[req.Name]) + + for i := range odo[0].Fields { + f := &odo[0].Fields[i] + f.IsPrimaryKey = f.Name == s.dloPrimaryKey + f.Label = cmp.Or(f.Label, f.Name) // default label } - // Creates the DLO - dataLakeObject, err := s.cnxn.client.CreateDataLakeObjectWithInferredSchema(ctx, s.query, s.cnxn.dataSpace, s.targetDLO, s.dloPrimaryKey, api.DataLakeObjectCategory(s.dloCategory)) + odo[0].Category = "Profile" // TODO + odo[0].Label = cmp.Or(odo[0].Label, odo[0].Name) // default label + req.Definition.OutputDataObjects = odo + + logger.DebugContext(ctx, "Creating batch DT", "req", req) + dt, err := s.cnxn.client.CreateOrUpdateDataTransform(ctx, req) if err != nil { - return nil, 0, adbc.Error{ - Code: adbc.StatusInternal, - Msg: err.Error(), - } + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "failed to create/update the data transform: %v", err) } - // Inserts data - _, err = s.cnxn.client.TriggerDbtBatchDataTransform(ctx, dataLakeObject, s.query, true, s.dataTransformTimeout) + logger.DebugContext(ctx, "Created batch DT. Waiting...", "dt", dt) + + dt, err = s.cnxn.client.WaitForDataTransform(ctx, dt) if err != nil { - return nil, 0, adbc.Error{ - Code: adbc.StatusInternal, - Msg: err.Error(), - } + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "failed while wating for data transform: %v", err) + } + if !dt.Status.IsActive() { + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "data transform is not active, current status: %v", dt.Status) + } + + logger.DebugContext(ctx, "Running batch DT.", "dt", dt) + + err = s.cnxn.client.MustRunDataTransform(ctx, dt.Name) + if err != nil { + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "failed to run data transform: %v", err) + } + + logger.DebugContext(ctx, "Run started. Waiting...", "dt", dt) + + dt, err = s.cnxn.client.WaitForDataTransformRun(ctx, dt, s.dataTransformTimeout) + if err != nil { + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "failed while wating for data transform run: %v", err) + } + if !dt.LastRunStatus.IsSuccess() { + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "data transform run was unsuccessful: last run status: %v", dt.LastRunStatus) + } + + logger.DebugContext(ctx, "Run complete. Associating with dataspace.") + + _, err = s.cnxn.client.UpsertDataSpaceMembers(ctx, s.cnxn.dataSpace, []api.DataSpaceMember{{Name: s.targetDLO}}) + if err != nil { + return nil, 0, s.cnxn.ErrorHelper.Errorf(adbc.StatusInternal, "failed to associate with dataspace: %v", err) } // Returns empty - emptySchema := arrow.NewSchema([]arrow.Field{}, nil) - reader, err := array.NewRecordReader(emptySchema, []arrow.Record{}) + emptySchema := arrow.NewSchema([]arrow.Field{}, nil) // TODO + reader, err := array.NewRecordReader(emptySchema, []arrow.RecordBatch{}) if err != nil { err = fmt.Errorf("failed to create empty record reader: %w", err) return nil, 0, adbc.Error{ @@ -167,7 +225,7 @@ func (s *statement) convertSqlQueryResponseToArrow(response *api.SqlQueryRespons if len(response.Data) == 0 { // Return empty reader with schema if available schema := s.buildArrowSchema(response.Metadata) - reader, err := array.NewRecordReader(schema, []arrow.Record{}) + reader, err := array.NewRecordReader(schema, []arrow.RecordBatch{}) return reader, 0, err } @@ -218,6 +276,10 @@ func (s *statement) GetOption(key string) (string, error) { return s.dloCategory, nil case OptionStringDLOPrimaryKey: return s.dloPrimaryKey, nil + case OptionStringDLOWriteMode: + return s.dloWriteMode, nil + case OptionStringDLOMaterialized: // TODO + return "table", nil case OptionsStringTargetDLO: return s.targetDLO, nil } @@ -258,6 +320,10 @@ func (s *statement) SetOption(key, value string) error { s.dloCategory = value case OptionStringDLOPrimaryKey: s.dloPrimaryKey = value + case OptionStringDLOWriteMode: + s.dloWriteMode = value // TODO validate + case OptionStringDLOMaterialized: // TODO + // TODO: noop for now case OptionsStringTargetDLO: s.targetDLO = value default: diff --git a/go/adbc/driver/salesforce/types.go b/go/adbc/driver/salesforce/types.go index 4e1d1752d7..dd067d029f 100644 --- a/go/adbc/driver/salesforce/types.go +++ b/go/adbc/driver/salesforce/types.go @@ -27,7 +27,7 @@ func (s *statement) buildArrowSchema(metadata []api.SqlQueryMetadata) *arrow.Sch } // buildArrowRecords converts the raw data to Arrow records -func (s *statement) buildArrowRecords(schema *arrow.Schema, data [][]interface{}) ([]arrow.Record, error) { +func (s *statement) buildArrowRecords(schema *arrow.Schema, data [][]any) ([]arrow.Record, error) { if len(data) == 0 { return []arrow.Record{}, nil } @@ -77,7 +77,7 @@ func (s *statement) buildArrowRecords(schema *arrow.Schema, data [][]interface{} } // appendValueToBuilder appends a value to the appropriate builder type using the dataType -func appendValueToBuilder(builder array.Builder, value interface{}, dataType arrow.DataType) { +func appendValueToBuilder(builder array.Builder, value any, dataType arrow.DataType) { // Convert value based on the target Arrow data type switch dataType.ID() { case arrow.STRING: @@ -156,7 +156,7 @@ func appendValueToBuilder(builder array.Builder, value interface{}, dataType arr } } -func convertToTimestamp(value interface{}) (arrow.Timestamp, bool) { +func convertToTimestamp(value any) (arrow.Timestamp, bool) { switch v := value.(type) { case string: timestamp, err := arrow.TimestampFromString(v, arrow.Microsecond) @@ -168,7 +168,7 @@ func convertToTimestamp(value interface{}) (arrow.Timestamp, bool) { } // Helper functions for type conversion -func convertToInt64(value interface{}) (int64, bool) { +func convertToInt64(value any) (int64, bool) { switch v := value.(type) { case int64: return v, true @@ -184,7 +184,7 @@ func convertToInt64(value interface{}) (int64, bool) { return 0, false } -func convertToInt32(value interface{}) (int32, bool) { +func convertToInt32(value any) (int32, bool) { switch v := value.(type) { case int32: return v, true @@ -198,7 +198,7 @@ func convertToInt32(value interface{}) (int32, bool) { return 0, false } -func convertToInt16(value interface{}) (int16, bool) { +func convertToInt16(value any) (int16, bool) { switch v := value.(type) { case int16: return v, true @@ -210,7 +210,7 @@ func convertToInt16(value interface{}) (int16, bool) { return 0, false } -func convertToFloat64(value interface{}) (float64, bool) { +func convertToFloat64(value any) (float64, bool) { switch v := value.(type) { case float64: return v, true @@ -220,7 +220,7 @@ func convertToFloat64(value interface{}) (float64, bool) { return 0, false } -func convertToFloat32(value interface{}) (float32, bool) { +func convertToFloat32(value any) (float32, bool) { switch v := value.(type) { case float32: return v, true