Skip to content

Commit d662a45

Browse files
olavloiteolavloite
authored
feat(spanner): add option for LastStatement in transaction (#11638)
Adds an option to indicate that a statement is the last statement in a transaction. This allows Spanner to skip some validations during the execution of the statement, and instead rely on the validations during the Commit. This option will also be used by the database/sql driver for statements that are executed directly on a connection instead of in a transaction. This change also fixes a number of test cases so that the test time is reduced from 2 minutes to approx 5 seconds.
1 parent 9e508d0 commit d662a45

5 files changed

Lines changed: 245 additions & 8 deletions

File tree

‎spanner/client_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4311,7 +4311,7 @@ func TestClient_WithGRPCConnectionPool(t *testing.T) {
43114311
if useGRPCgcp {
43124312
_, client, teardown = setupMockedTestServerWithConfigAndGCPMultiendpointPool(
43134313
t,
4314-
ClientConfig{},
4314+
ClientConfig{DisableNativeMetrics: true},
43154315
[]option.ClientOption{option.WithGRPCConnectionPool(configuredConnPool)},
43164316
&grpc_gcp.ChannelPoolConfig{
43174317
MinSize: uint32(gcpPoolNumChannels),
@@ -4321,7 +4321,7 @@ func TestClient_WithGRPCConnectionPool(t *testing.T) {
43214321
} else {
43224322
_, client, teardown = setupMockedTestServerWithConfigAndClientOptions(
43234323
t,
4324-
ClientConfig{},
4324+
ClientConfig{DisableNativeMetrics: true},
43254325
[]option.ClientOption{option.WithGRPCConnectionPool(configuredConnPool)},
43264326
)
43274327
}
@@ -4346,7 +4346,7 @@ func TestClient_WithGRPCConnectionPoolAndNumChannels(t *testing.T) {
43464346
if useGRPCgcp {
43474347
_, client, teardown = setupMockedTestServerWithConfigAndGCPMultiendpointPool(
43484348
t,
4349-
ClientConfig{NumChannels: configuredNumChannels},
4349+
ClientConfig{NumChannels: configuredNumChannels, DisableNativeMetrics: true},
43504350
[]option.ClientOption{option.WithGRPCConnectionPool(configuredConnPool)},
43514351
&grpc_gcp.ChannelPoolConfig{
43524352
MaxSize: uint32(gcpPoolNumChannels),
@@ -4356,7 +4356,7 @@ func TestClient_WithGRPCConnectionPoolAndNumChannels(t *testing.T) {
43564356
} else {
43574357
_, client, teardown = setupMockedTestServerWithConfigAndClientOptions(
43584358
t,
4359-
ClientConfig{NumChannels: configuredNumChannels},
4359+
ClientConfig{NumChannels: configuredNumChannels, DisableNativeMetrics: true},
43604360
[]option.ClientOption{option.WithGRPCConnectionPool(configuredConnPool)},
43614361
)
43624362
}
@@ -4381,7 +4381,7 @@ func TestClient_WithGRPCConnectionPoolAndNumChannels_Misconfigured(t *testing.T)
43814381
defer serverTeardown()
43824382
opts = append(opts, option.WithGRPCConnectionPool(configuredConnPool))
43834383

4384-
config := ClientConfig{NumChannels: configuredNumChannels}
4384+
config := ClientConfig{NumChannels: configuredNumChannels, DisableNativeMetrics: true}
43854385
_, err := makeClientWithConfig(context.Background(), "projects/p/instances/i/databases/d", config, server.ServerAddress, opts...)
43864386
if useGRPCgcp {
43874387
// GCPMultiEndpoint channel pool config is preceeding default pool config.

‎spanner/internal/testutil/inmem_spanner_server.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -681,48 +681,57 @@ func (s *inMemSpannerServer) getStatementResult(sql string) (*StatementResult, e
681681

682682
func (s *inMemSpannerServer) simulateExecutionTime(method string, req interface{}) (interface{}, error) {
683683
s.mu.Lock()
684-
defer s.mu.Unlock()
685684

686685
// Check if the server is stopped
687686
if s.stopped {
687+
s.mu.Unlock()
688688
return nil, gstatus.Error(codes.Unavailable, "server has been stopped")
689689
}
690690

691691
// Send the request to the receivedRequests channel
692692
s.receivedRequests <- req
693+
s.mu.Unlock()
694+
s.ready()
695+
s.mu.Lock()
693696

694697
// Check for a simulated error
695698
if s.err != nil {
696699
err := s.err
697700
s.err = nil
701+
s.mu.Unlock()
698702
return nil, err
699703
}
700704

701705
// Check for a simulated execution time
702706
executionTime, ok := s.executionTimes[method]
707+
s.mu.Unlock()
703708
if ok {
704709
var randTime int64
705710
if executionTime.RandomExecutionTime > 0 {
706711
randTime = rand.Int63n(int64(executionTime.RandomExecutionTime))
707712
}
708713
totalExecutionTime := time.Duration(int64(executionTime.MinimumExecutionTime) + randTime)
709714
<-time.After(totalExecutionTime)
715+
s.mu.Lock()
710716

711717
// Check for errors in the execution time
712718
if len(executionTime.Errors) > 0 {
713719
err := executionTime.Errors[0]
714720
if !executionTime.KeepError {
715721
executionTime.Errors = executionTime.Errors[1:]
716722
}
723+
s.mu.Unlock()
717724
return nil, err
718725
}
719726

720727
// Check for responses in the execution time
721728
if len(executionTime.Responses) > 0 {
722729
response := executionTime.Responses[0]
723730
executionTime.Responses = executionTime.Responses[1:]
731+
s.mu.Unlock()
724732
return response, nil
725733
}
734+
s.mu.Unlock()
726735
}
727736

728737
return nil, nil

‎spanner/request_id_header_test.go‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,7 @@ func TestRequestIDHeader_sentOnEveryClientCall(t *testing.T) {
133133
WriteSessions: 0.2,
134134
incStep: 2,
135135
},
136+
DisableNativeMetrics: true,
136137
}
137138
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
138139
t.Cleanup(tearDown)
@@ -469,6 +470,7 @@ func TestRequestIDHeader_onRetriesWithFailedTransactionCommit(t *testing.T) {
469470
WriteSessions: 0.2,
470471
incStep: 2,
471472
},
473+
DisableNativeMetrics: true,
472474
}
473475
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
474476
t.Cleanup(tearDown)
@@ -531,6 +533,7 @@ func TestRequestIDHeader_retriesOnSessionNotFound(t *testing.T) {
531533
WriteSessions: 0.2,
532534
incStep: 2,
533535
},
536+
DisableNativeMetrics: true,
534537
}
535538
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
536539
t.Cleanup(tearDown)
@@ -613,6 +616,7 @@ func TestRequestIDHeader_BatchDMLWithMultipleDML(t *testing.T) {
613616
WriteSessions: 0.2,
614617
incStep: 2,
615618
},
619+
DisableNativeMetrics: true,
616620
}
617621

618622
ctx := context.Background()
@@ -697,6 +701,7 @@ func TestRequestIDHeader_clientBatchWrite(t *testing.T) {
697701
WriteSessions: 0.2,
698702
incStep: 2,
699703
},
704+
DisableNativeMetrics: true,
700705
}
701706

702707
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -759,6 +764,7 @@ func TestRequestIDHeader_ClientBatchWriteWithSessionNotFound(t *testing.T) {
759764
WriteSessions: 0.2,
760765
incStep: 2,
761766
},
767+
DisableNativeMetrics: true,
762768
}
763769

764770
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -829,6 +835,7 @@ func TestRequestIDHeader_ClientBatchWriteWithError(t *testing.T) {
829835
WriteSessions: 0.2,
830836
incStep: 2,
831837
},
838+
DisableNativeMetrics: true,
832839
}
833840

834841
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -906,6 +913,7 @@ func testRequestIDHeaderPartitionQuery(t *testing.T, mustErrorOnPartitionQuery b
906913
WriteSessions: 0.2,
907914
incStep: 2,
908915
},
916+
DisableNativeMetrics: true,
909917
}
910918

911919
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1077,6 +1085,7 @@ func TestRequestIDHeader_ReadWriteTransactionUpdate(t *testing.T) {
10771085
WriteSessions: 0.2,
10781086
incStep: 2,
10791087
},
1088+
DisableNativeMetrics: true,
10801089
}
10811090

10821091
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1164,6 +1173,7 @@ func TestRequestIDHeader_ReadWriteTransactionBatchUpdateWithOptions(t *testing.T
11641173
WriteSessions: 0.2,
11651174
incStep: 2,
11661175
},
1176+
DisableNativeMetrics: true,
11671177
}
11681178

11691179
_, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1227,6 +1237,7 @@ func TestRequestIDHeader_multipleParallelCallsWithConventionalCustomerCalls(t *t
12271237
WriteSessions: 0.2,
12281238
incStep: 2,
12291239
},
1240+
DisableNativeMetrics: true,
12301241
}
12311242

12321243
// We created exactly 1 client.
@@ -1358,6 +1369,7 @@ func TestRequestIDHeader_RetryOnAbortAndValidate(t *testing.T) {
13581369
WriteSessions: 0.2,
13591370
incStep: 2,
13601371
},
1372+
DisableNativeMetrics: true,
13611373
}
13621374

13631375
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1446,6 +1458,7 @@ func TestRequestIDHeader_BatchCreateSessions_Unavailable(t *testing.T) {
14461458
WriteSessions: 0.2,
14471459
incStep: 2,
14481460
},
1461+
DisableNativeMetrics: true,
14491462
}
14501463

14511464
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1530,6 +1543,7 @@ func TestRequestIDHeader_SingleUseReadOnly_ExecuteStreamingSql_Unavailable(t *te
15301543
WriteSessions: 0.2,
15311544
incStep: 2,
15321545
},
1546+
DisableNativeMetrics: true,
15331547
}
15341548

15351549
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1613,6 +1627,7 @@ func TestRequestIDHeader_SingleUseReadOnly_ExecuteStreamingSql_InvalidArgument(t
16131627
WriteSessions: 0.2,
16141628
incStep: 2,
16151629
},
1630+
DisableNativeMetrics: true,
16161631
}
16171632

16181633
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1657,6 +1672,7 @@ func TestRequestIDHeader_SingleUseReadOnly_ExecuteStreamingSql_ContextDeadlineEx
16571672
WriteSessions: 0.2,
16581673
incStep: 2,
16591674
},
1675+
DisableNativeMetrics: true,
16601676
}
16611677

16621678
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1702,6 +1718,7 @@ func TestRequestIDHeader_Commit_ContextDeadlineExceeded(t *testing.T) {
17021718
WriteSessions: 0.2,
17031719
incStep: 2,
17041720
},
1721+
DisableNativeMetrics: true,
17051722
}
17061723

17071724
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1744,7 +1761,8 @@ func TestRequestIDHeader_VerifyChannelNumber(t *testing.T) {
17441761
MaxOpened: 400,
17451762
incStep: 25,
17461763
},
1747-
NumChannels: 4,
1764+
NumChannels: 4,
1765+
DisableNativeMetrics: true,
17481766
}
17491767

17501768
_, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
@@ -1861,6 +1879,7 @@ func TestRequestIDHeader_SingleUseReadOnly_ExecuteStreamingSql_UnavailableDuring
18611879
WriteSessions: 0.2,
18621880
incStep: 2,
18631881
},
1882+
DisableNativeMetrics: true,
18641883
}
18651884
server, sc, tearDown := setupMockedTestServerWithConfigAndClientOptions(t, clientConfig, clientOpts)
18661885
t.Cleanup(tearDown)

‎spanner/transaction.go‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -484,8 +484,19 @@ type QueryOptions struct {
484484

485485
// Controls whether to exclude recording modifications in current partitioned update operation
486486
// from the allowed tracking change streams(with DDL option allow_txn_exclusion=true). Setting
487-
// this value for any sql/dml requests other than partitioned udpate will receive an error.
487+
// this value for any sql/dml requests other than partitioned update will receive an error.
488488
ExcludeTxnFromChangeStreams bool
489+
490+
// LastStatement indicates whether this statement is the last statement in this transaction.
491+
// If set to true, this option marks the end of the transaction. The transaction should be
492+
// committed or rolled back after this statement executes, and attempts to execute any other requests
493+
// against this transaction (including reads and queries) will be rejected. Mixing mutations with
494+
// statements that are marked as the last statement is not allowed.
495+
//
496+
// For DML statements, setting this option may cause some error reporting to be deferred until
497+
// commit time (e.g. validation of unique constraints). Given this, successful execution of a DML
498+
// statement should not be assumed until the transaction commits.
499+
LastStatement bool
489500
}
490501

491502
// merge combines two QueryOptions that the input parameter will have higher
@@ -499,6 +510,7 @@ func (qo QueryOptions) merge(opts QueryOptions) QueryOptions {
499510
DataBoostEnabled: qo.DataBoostEnabled,
500511
DirectedReadOptions: qo.DirectedReadOptions,
501512
ExcludeTxnFromChangeStreams: qo.ExcludeTxnFromChangeStreams || opts.ExcludeTxnFromChangeStreams,
513+
LastStatement: qo.LastStatement || opts.LastStatement,
502514
}
503515
if opts.Mode != nil {
504516
merged.Mode = opts.Mode
@@ -683,6 +695,7 @@ func (t *txReadOnly) prepareExecuteSQL(ctx context.Context, stmt Statement, opti
683695
RequestOptions: createRequestOptions(options.Priority, options.RequestTag, t.txOpts.TransactionTag),
684696
DataBoostEnabled: options.DataBoostEnabled,
685697
DirectedReadOptions: options.DirectedReadOptions,
698+
LastStatement: options.LastStatement,
686699
}
687700
return req, sh, nil
688701
}
@@ -1347,6 +1360,7 @@ func (t *ReadWriteTransaction) batchUpdateWithOptions(ctx context.Context, stmts
13471360
Statements: sppbStmts,
13481361
Seqno: atomic.AddInt64(&t.sequenceNumber, 1),
13491362
RequestOptions: createRequestOptions(opts.Priority, opts.RequestTag, t.txOpts.TransactionTag),
1363+
LastStatements: opts.LastStatement,
13501364
}, gax.WithGRPCOptions(grpc.Header(&md)))
13511365

13521366
if getGFELatencyMetricsFlag() && md != nil && t.ct != nil {

0 commit comments

Comments
 (0)