@Test public void noopMapOnUpdateQueriesWhenNoRowsAffected() { String sql = "delete from Books where true"; ResultSet resultSet = ResultSet.newBuilder() .setMetadata(ResultSetMetadata.getDefaultInstance()) .setStats(ResultSetStats.getDefaultInstance()) .build(); List<Statement> statementList = Collections.singletonList( ExecuteBatchDmlRequest.Statement.newBuilder() .setSql(sql) .setParams(Struct.getDefaultInstance()) .build()); when(this.mockClient.executeBatchDml(this.mockContext, statementList)) .thenReturn(Flux.just(resultSet)); SpannerStatement statement = new SpannerStatement(this.mockClient, this.mockContext, sql, TEST_CONFIG); Flux<SpannerResult> result = (Flux<SpannerResult>) statement.execute(); StepVerifier.create(result.flatMap( spannerResult -> spannerResult.map((row, rowMetadata) -> (String) row.get(0)) .collectList())) .expectNext(Collections.emptyList()) .verifyComplete(); StepVerifier.create(result.flatMap(results -> Mono.from(results.getRowsUpdated()))) .expectNext(0) .verifyComplete(); verify(this.mockClient, times(1)) .executeBatchDml(this.mockContext, statementList); }
@Override public Publisher<? extends Result> execute() { if (this.statementType == StatementType.DDL) { return this.client .executeDdl( this.config.getFullyQualifiedDatabaseName(), Collections.singletonList(this.sql), this.config.getDdlOperationTimeout(), this.config.getDdlOperationPollInterval()) .map(operation -> new SpannerResult(Flux.empty(), Mono.just(0))); } else if (this.statementType == StatementType.DML && !this.ctx.isTransactionPartitionedDml()) { List<ExecuteBatchDmlRequest.Statement> dmlStatements = this.statementBindings.getBindings().stream() .map(struct -> ExecuteBatchDmlRequest.Statement.newBuilder() .setSql(this.sql) .setParams(struct) .putAllParamTypes(this.statementBindings.getTypes()) .build()) .collect(Collectors.toList()); return this.client.executeBatchDml(this.ctx, dmlStatements) .map(partialResultSet -> Math.toIntExact(partialResultSet.getStats().getRowCountExact())) .map(rowCount -> new SpannerResult(Flux.empty(), Mono.just(rowCount))); } Flux<Struct> structFlux = Flux.fromIterable(this.statementBindings.getBindings()); return structFlux.flatMap(this::runStreamingSql); }
SpannerStatement implements Statement { @Override public Publisher<? extends Result> execute() { if (this.statementType == StatementType.DDL) { return this.client .executeDdl( this.config.getFullyQualifiedDatabaseName(), Collections.singletonList(this.sql), this.config.getDdlOperationTimeout(), this.config.getDdlOperationPollInterval()) .map(operation -> new SpannerResult(Flux.empty(), Mono.just(0))); } else if (this.statementType == StatementType.DML && !this.ctx.isTransactionPartitionedDml()) { List<ExecuteBatchDmlRequest.Statement> dmlStatements = this.statementBindings.getBindings().stream() .map(struct -> ExecuteBatchDmlRequest.Statement.newBuilder() .setSql(this.sql) .setParams(struct) .putAllParamTypes(this.statementBindings.getTypes()) .build()) .collect(Collectors.toList()); return this.client.executeBatchDml(this.ctx, dmlStatements) .map(partialResultSet -> Math.toIntExact(partialResultSet.getStats().getRowCountExact())) .map(rowCount -> new SpannerResult(Flux.empty(), Mono.just(rowCount))); } Flux<Struct> structFlux = Flux.fromIterable(this.statementBindings.getBindings()); return structFlux.flatMap(this::runStreamingSql); } }
SpannerStatement implements Statement { @Override public Publisher<? extends Result> execute() { if (this.statementType == StatementType.DDL) { return this.client .executeDdl( this.config.getFullyQualifiedDatabaseName(), Collections.singletonList(this.sql), this.config.getDdlOperationTimeout(), this.config.getDdlOperationPollInterval()) .map(operation -> new SpannerResult(Flux.empty(), Mono.just(0))); } else if (this.statementType == StatementType.DML && !this.ctx.isTransactionPartitionedDml()) { List<ExecuteBatchDmlRequest.Statement> dmlStatements = this.statementBindings.getBindings().stream() .map(struct -> ExecuteBatchDmlRequest.Statement.newBuilder() .setSql(this.sql) .setParams(struct) .putAllParamTypes(this.statementBindings.getTypes()) .build()) .collect(Collectors.toList()); return this.client.executeBatchDml(this.ctx, dmlStatements) .map(partialResultSet -> Math.toIntExact(partialResultSet.getStats().getRowCountExact())) .map(rowCount -> new SpannerResult(Flux.empty(), Mono.just(rowCount))); } Flux<Struct> structFlux = Flux.fromIterable(this.statementBindings.getBindings()); return structFlux.flatMap(this::runStreamingSql); } SpannerStatement(
      Client client,
      StatementExecutionContext ctx,
      String sql,
      SpannerConnectionConfiguration config); }
SpannerStatement implements Statement { @Override public Publisher<? extends Result> execute() { if (this.statementType == StatementType.DDL) { return this.client .executeDdl( this.config.getFullyQualifiedDatabaseName(), Collections.singletonList(this.sql), this.config.getDdlOperationTimeout(), this.config.getDdlOperationPollInterval()) .map(operation -> new SpannerResult(Flux.empty(), Mono.just(0))); } else if (this.statementType == StatementType.DML && !this.ctx.isTransactionPartitionedDml()) { List<ExecuteBatchDmlRequest.Statement> dmlStatements = this.statementBindings.getBindings().stream() .map(struct -> ExecuteBatchDmlRequest.Statement.newBuilder() .setSql(this.sql) .setParams(struct) .putAllParamTypes(this.statementBindings.getTypes()) .build()) .collect(Collectors.toList()); return this.client.executeBatchDml(this.ctx, dmlStatements) .map(partialResultSet -> Math.toIntExact(partialResultSet.getStats().getRowCountExact())) .map(rowCount -> new SpannerResult(Flux.empty(), Mono.just(rowCount))); } Flux<Struct> structFlux = Flux.fromIterable(this.statementBindings.getBindings()); return structFlux.flatMap(this::runStreamingSql); } SpannerStatement(
      Client client,
      StatementExecutionContext ctx,
      String sql,
      SpannerConnectionConfiguration config); @Override Statement add(); @Override Statement bind(String identifier, Object value); @Override Statement bind(int i, Object o); @Override Statement bindNull(String identifier, Class<?> type); @Override Statement bindNull(int i, Class<?> type); @Override Publisher<? extends Result> execute(); }
SpannerStatement implements Statement { @Override public Publisher<? extends Result> execute() { if (this.statementType == StatementType.DDL) { return this.client .executeDdl( this.config.getFullyQualifiedDatabaseName(), Collections.singletonList(this.sql), this.config.getDdlOperationTimeout(), this.config.getDdlOperationPollInterval()) .map(operation -> new SpannerResult(Flux.empty(), Mono.just(0))); } else if (this.statementType == StatementType.DML && !this.ctx.isTransactionPartitionedDml()) { List<ExecuteBatchDmlRequest.Statement> dmlStatements = this.statementBindings.getBindings().stream() .map(struct -> ExecuteBatchDmlRequest.Statement.newBuilder() .setSql(this.sql) .setParams(struct) .putAllParamTypes(this.statementBindings.getTypes()) .build()) .collect(Collectors.toList()); return this.client.executeBatchDml(this.ctx, dmlStatements) .map(partialResultSet -> Math.toIntExact(partialResultSet.getStats().getRowCountExact())) .map(rowCount -> new SpannerResult(Flux.empty(), Mono.just(rowCount))); } Flux<Struct> structFlux = Flux.fromIterable(this.statementBindings.getBindings()); return structFlux.flatMap(this::runStreamingSql); } SpannerStatement(
      Client client,
      StatementExecutionContext ctx,
      String sql,
      SpannerConnectionConfiguration config); @Override Statement add(); @Override Statement bind(String identifier, Object value); @Override Statement bind(int i, Object o); @Override Statement bindNull(String identifier, Class<?> type); @Override Statement bindNull(int i, Class<?> type); @Override Publisher<? extends Result> execute();  }