feat(salesforce): incremental support

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)

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"
 

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{

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)

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),
 	}
 }