@Test public void readMultiResultSetQueryTest() { PartialResultSet p1 = PartialResultSet.newBuilder().setMetadata( this.resultSetMetadata ).setChunkedValue(false) .addValues(this.a1).build(); PartialResultSet p2 = PartialResultSet.newBuilder().setChunkedValue(false) .addValues(this.a2).build(); Flux<PartialResultSet> inputs = Flux.just(p1, p2); when(this.mockClient.executeStreamingSql( same(this.mockContext), any(), any(), any())).thenReturn(inputs); StepVerifier.create(Flux.from( new SpannerStatement(this.mockClient, this.mockContext, "SELECT", TEST_CONFIG).execute()) .flatMap(r -> Mono.from(r.getRowsUpdated()))) .expectNext(0) .verifyComplete(); }
@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();  }