1// Copyright 2018 Google LLC
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15package spanner
16
17import (
18	"context"
19
20	"cloud.google.com/go/internal/trace"
21	"github.com/googleapis/gax-go/v2"
22	sppb "google.golang.org/genproto/googleapis/spanner/v1"
23	"google.golang.org/grpc/codes"
24)
25
26// PartitionedUpdate executes a DML statement in parallel across the database,
27// using separate, internal transactions that commit independently. The DML
28// statement must be fully partitionable: it must be expressible as the union
29// of many statements each of which accesses only a single row of the table. The
30// statement should also be idempotent, because it may be applied more than once.
31//
32// PartitionedUpdate returns an estimated count of the number of rows affected.
33// The actual number of affected rows may be greater than the estimate.
34func (c *Client) PartitionedUpdate(ctx context.Context, statement Statement) (count int64, err error) {
35	return c.partitionedUpdate(ctx, statement, c.qo)
36}
37
38// PartitionedUpdateWithOptions executes a DML statement in parallel across the database,
39// using separate, internal transactions that commit independently. The sql
40// query execution will be optimized based on the given query options.
41func (c *Client) PartitionedUpdateWithOptions(ctx context.Context, statement Statement, opts QueryOptions) (count int64, err error) {
42	return c.partitionedUpdate(ctx, statement, c.qo.merge(opts))
43}
44
45func (c *Client) partitionedUpdate(ctx context.Context, statement Statement, options QueryOptions) (count int64, err error) {
46	ctx = trace.StartSpan(ctx, "cloud.google.com/go/spanner.PartitionedUpdate")
47	defer func() { trace.EndSpan(ctx, err) }()
48	if err := checkNestedTxn(ctx); err != nil {
49		return 0, err
50	}
51
52	sh, err := c.idleSessions.take(ctx)
53	if err != nil {
54		return 0, ToSpannerError(err)
55	}
56	if sh != nil {
57		defer sh.recycle()
58	}
59
60	// Create the parameters and the SQL request, but without a transaction.
61	// The transaction reference will be added by the executePdml method.
62	params, paramTypes, err := statement.convertParams()
63	if err != nil {
64		return 0, ToSpannerError(err)
65	}
66	req := &sppb.ExecuteSqlRequest{
67		Session:        sh.getID(),
68		Sql:            statement.SQL,
69		Params:         params,
70		ParamTypes:     paramTypes,
71		QueryOptions:   options.Options,
72		RequestOptions: createRequestOptions(options.Priority, options.RequestTag, ""),
73	}
74
75	// Make a retryer for Aborted and certain Internal errors.
76	retryer := onCodes(DefaultRetryBackoff, codes.Aborted, codes.Internal)
77	// Execute the PDML and retry if the transaction is aborted.
78	executePdmlWithRetry := func(ctx context.Context) (int64, error) {
79		for {
80			count, err := executePdml(ctx, sh, req)
81			if err == nil {
82				return count, nil
83			}
84			delay, shouldRetry := retryer.Retry(err)
85			if !shouldRetry {
86				return 0, err
87			}
88			if err := gax.Sleep(ctx, delay); err != nil {
89				return 0, err
90			}
91		}
92	}
93	return executePdmlWithRetry(ctx)
94}
95
96// executePdml executes the following steps:
97// 1. Begin a PDML transaction
98// 2. Add the ID of the PDML transaction to the SQL request.
99// 3. Execute the update statement on the PDML transaction
100//
101// Note that PDML transactions cannot be committed or rolled back.
102func executePdml(ctx context.Context, sh *sessionHandle, req *sppb.ExecuteSqlRequest) (count int64, err error) {
103	// Begin transaction.
104	res, err := sh.getClient().BeginTransaction(contextWithOutgoingMetadata(ctx, sh.getMetadata()), &sppb.BeginTransactionRequest{
105		Session: sh.getID(),
106		Options: &sppb.TransactionOptions{
107			Mode: &sppb.TransactionOptions_PartitionedDml_{PartitionedDml: &sppb.TransactionOptions_PartitionedDml{}},
108		},
109	})
110	if err != nil {
111		return 0, ToSpannerError(err)
112	}
113	// Add a reference to the PDML transaction on the ExecuteSql request.
114	req.Transaction = &sppb.TransactionSelector{
115		Selector: &sppb.TransactionSelector_Id{Id: res.Id},
116	}
117	resultSet, err := sh.getClient().ExecuteSql(contextWithOutgoingMetadata(ctx, sh.getMetadata()), req)
118	if err != nil {
119		return 0, err
120	}
121	if resultSet.Stats == nil {
122		return 0, spannerErrorf(codes.InvalidArgument, "query passed to Update: %q", req.Sql)
123	}
124	return extractRowCount(resultSet.Stats)
125}
126