Batch DML support #107
Batch DML support #107
Conversation
…-r2dbc into batch-dml # Conflicts: # src/main/java/com/google/cloud/spanner/r2dbc/SpannerStatement.java # src/test/java/com/google/cloud/spanner/r2dbc/it/SpannerIT.java
dmitry-s
requested review from
elefeint,
meltsufin,
dzou and
ChengyuanZhao
and removed request for
elefeint and
meltsufin
| PartialResultRowExtractor partialResultRowExtractor = new PartialResultRowExtractor(); | ||
|
|
||
| Flux<PartialResultSet> resultSetFlux = | ||
| this.client.executeStreamingSql( | ||
| this.session, this.transaction, this.sql, params, this.statementBindings.getTypes()); | ||
|
|
||
| if (statementType == StatementType.SELECT) { | ||
| if (this.statementType == StatementType.SELECT) { |
There was a problem hiding this comment.
Do we need this check anymore? DML does not go through this method anymore, and I don't imagine DDL will either.
There was a problem hiding this comment.
+1 and should probably rename to runSelectStatement
| } | ||
|
|
||
| return ObservableReactiveUtil | ||
| .streamingCall(obs -> this.spanner.executeBatchDml(request.build(), obs)); |
There was a problem hiding this comment.
It's not really a streaming call; use unaryCall() here.
| @@ -268,12 +292,13 @@ public void testEmptySelect() { | |||
| /** | |||
| * Executes a DML query and returns the rows updated. | |||
| */ | |||
| private int executeDmlQuery(String sql) { | |||
| private List<Integer> executeDmlQuery(String sql) { | |||
There was a problem hiding this comment.
Should this be kept as before, returning integer? B/c String sql can only correspond to one statement.
|
|
||
| if (statementType == StatementType.SELECT) { | ||
| return structFlux.flatMap(struct -> runSingleStatement(struct, statementType)); | ||
| if (this.statementType == StatementType.DML) { |
There was a problem hiding this comment.
I wonder if you should just have a class hierarchy of statements -- SpannerDmlStatement, SpannerDdlStatement, SpannerReadStatement. WDYT?
| } | ||
| // DML statements have to be executed sequentially because they need seqNo to be in order | ||
| return structFlux.concatMapDelayError(struct -> runSingleStatement(struct, statementType)); | ||
| Flux<Struct> structFlux = Flux.fromIterable(this.statementBindings.getBindings()); |
There was a problem hiding this comment.
Else if SELECT? Are you missing the DDL case?
The whole thing cloud be a switch statement, with the default throwing an exception.
Alternatively, see my suggestion about the class hierarchy.
| PartialResultRowExtractor partialResultRowExtractor = new PartialResultRowExtractor(); | ||
|
|
||
| Flux<PartialResultSet> resultSetFlux = | ||
| this.client.executeStreamingSql( | ||
| this.session, this.transaction, this.sql, params, this.statementBindings.getTypes()); | ||
|
|
||
| if (statementType == StatementType.SELECT) { | ||
| if (this.statementType == StatementType.SELECT) { |
There was a problem hiding this comment.
+1 and should probably rename to runSelectStatement
| Connection connection = Mono.from(connectionFactory.create()).block(); | ||
|
|
||
| Mono.from(connection.beginTransaction()).block(); | ||
| int rowsUpdated = Mono.from(connection.createStatement(sql).execute()) | ||
| List<Integer> rowsUpdated = Flux.from(connection.createStatement(sql).execute()) |
There was a problem hiding this comment.
s/rowsUpdated/rowsUpdatedPerStatement
fixes #92