feat: upgrade to R2DBC SPI [do not merge]

cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/v2/SpannerClientLibraryResult.java

@@ -31,18 +31,18 @@ class SpannerClientLibraryResult implements Result {
 
   private final Flux<SpannerClientLibraryRow> resultRows;
 
-  private final int numRowsUpdated;
+  private final long numRowsUpdated;
 
   private RowMetadata rowMetadata;
 
   public SpannerClientLibraryResult(
-      Flux<SpannerClientLibraryRow> resultRows, int numRowsUpdated) {
+      Flux<SpannerClientLibraryRow> resultRows, long numRowsUpdated) {
     this.resultRows = Assert.requireNonNull(resultRows, "A non-null flux of rows is required.");
     this.numRowsUpdated = numRowsUpdated;
   }
 
   @Override
-  public Publisher<Integer> getRowsUpdated() {
+  public Publisher<Long> getRowsUpdated() {
     return Mono.just(this.numRowsUpdated);
   }
 

cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/v2/SpannerClientLibraryRowMetadata.java

@@ -81,7 +81,6 @@ public List<? extends ColumnMetadata> getColumnMetadatas() {
     return this.columnMetadatas;
   }
 
-  @Override
   public Collection<String> getColumnNames() {
     return this.columnNames;
   }

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/it/ClientLibraryBasedIntegrationTest.java

@@ -144,7 +144,7 @@ void testMetadata() {
                 .bind("price", new BigDecimal("123.99"))
                 .execute())
             .flatMapMany(rs -> rs.getRowsUpdated())
-    ).expectNext(1).verifyComplete();
+    ).expectNext(1L).verifyComplete();
 
     StepVerifier.create(
         Mono.from(conn.createStatement("SELECT AUTHOR, PRICE FROM BOOKS LIMIT 1").execute())
@@ -195,7 +195,7 @@ void testDmlInsert() {
                 )
                 .execute())
             .flatMapMany(rs -> rs.getRowsUpdated())
-    ).expectNext(1).verifyComplete();
+    ).expectNext(1L).verifyComplete();
 
     StepVerifier.create(
         Mono.from(conn.createStatement("SELECT count(*) FROM BOOKS").execute())
@@ -244,7 +244,7 @@ void testJsonFieldInsert() {
                         .bind("extra", JsonWrapper.of("{\"b\":9,\"a\":true}"))
                         .execute())
                 .flatMapMany(rs -> rs.getRowsUpdated()))
-        .expectNext(1)
+        .expectNext(1L)
         .verifyComplete();
 
     StepVerifier.create(
@@ -467,7 +467,7 @@ void selectMultipleBoundParameterSetsNoTransaction() {
                 ).flatMap(r -> r.getRowsUpdated())
             ))
 
-    ).expectNext(1, 1, 1).verifyComplete();
+    ).expectNext(1L, 1L, 1L).verifyComplete();
 
     StepVerifier.create(
         Mono.from(connectionFactory.create()).flatMapMany(
@@ -511,7 +511,7 @@ void selectMultipleBoundParameterSetsInTransaction() {
                 ).flatMap(r -> r.getRowsUpdated())
             ))
 
-    ).expectNext(1, 1, 1).verifyComplete();
+    ).expectNext(1L, 1L, 1L).verifyComplete();
 
     StepVerifier.create(
         Mono.from(connectionFactory.create()).flatMapMany(
@@ -553,7 +553,7 @@ void insertMultipleBoundParameterSetsNoTransaction() {
                     .execute()
             )
                 .flatMap(rs -> rs.getRowsUpdated()))
-    ).expectNext(1, 1, 1).as("Row insert count matches").verifyComplete();
+    ).expectNext(1L, 1L, 1L).as("Row insert count matches").verifyComplete();
 
     StepVerifier.create(
         Mono.from(connectionFactory.create())
@@ -767,7 +767,7 @@ void testBatchDml() {
               .add("UPDATE BOOKS SET CATEGORY=17 WHERE CATEGORY=29")
               .execute()
           ).flatMap(r -> r.getRowsUpdated())
-    ).expectNext(1, 1, 2)
+    ).expectNext(1L, 1L, 2L)
         .verifyComplete();
 
     verifyIds(uuid1, uuid2);

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/it/ClientLibraryDdlIntegrationTest.java

@@ -79,7 +79,7 @@ void ddlCreateAndDrop() {
             .execute())
             .log("Table" + tableName + " created")
             .flatMap(res -> Mono.from(res.getRowsUpdated()))
-    ).expectNext(0).as("DDL execution returns zero affected rows")
+    ).expectNext(0L).as("DDL execution returns zero affected rows")
         .verifyComplete();
 
     StepVerifier.create(

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/it/SpannerDmlReactiveStreamVerification.java

@@ -36,7 +36,7 @@
 import reactor.core.publisher.Mono;
 
 class SpannerDmlReactiveStreamVerification extends
-    PublisherVerification<Integer> {
+    PublisherVerification<Long> {
 
   private static ConnectionFactory connectionFactory;
 
@@ -71,7 +71,7 @@ public SpannerDmlReactiveStreamVerification() {
   }
 
   @Override
-  public Publisher<Integer> createPublisher(long l) {
+  public Publisher<Long> createPublisher(long l) {
     return Mono.from(connectionFactory.create())
         .flatMapMany(conn ->
             Flux.from(conn.createStatement(
@@ -82,7 +82,7 @@ public Publisher<Integer> createPublisher(long l) {
   }
 
   @Override
-  public Publisher<Integer> createFailedPublisher() {
+  public Publisher<Long> createFailedPublisher() {
     return Mono.from(connectionFactory.create())
         .flatMapMany(conn -> conn.createStatement("UPDATE BOOKS SET bad syntax ").execute())
         .flatMap(rs -> rs.getRowsUpdated());

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/it/SpannerQueryUtil.java

@@ -43,10 +43,10 @@ static <T> List<T> executeReadQuery(
   /**
    * Executes a DML query and returns the rows updated.
    */
-  static int executeDmlQuery(Connection connection, String sql) {
+  static long executeDmlQuery(Connection connection, String sql) {
 
     Mono.from(connection.beginTransaction()).block();
-    int rowsUpdated = Mono.from(connection.createStatement(sql).execute())
+    long rowsUpdated = Mono.from(connection.createStatement(sql).execute())
         .flatMap(result -> Mono.from(result.getRowsUpdated()))
         .block();
     Mono.from(connection.commitTransaction()).block();

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/AbstractSpannerClientLibraryStatementTest.java

@@ -61,7 +61,7 @@ void noParametersSendsSingleStatement() {
     FakeStatement statement = new FakeStatement(this.mockAdapter, query);
 
     StepVerifier.create(Flux.from(statement.execute()).flatMap(r -> r.getRowsUpdated()))
-        .expectNext(19)
+        .expectNext(19L)
         .verifyComplete();
 
     ArgumentCaptor<Statement> capturedStatement = ArgumentCaptor.forClass(Statement.class);
@@ -82,7 +82,7 @@ void singleSetOfParametersWithNoAddSendsOneStatement() {
     statement.bind("three", "333");
 
     StepVerifier.create(Flux.from(statement.execute()).flatMap(r -> r.getRowsUpdated()))
-        .expectNext(19)
+        .expectNext(19L)
         .verifyComplete();
 
     ArgumentCaptor<Statement> capturedStatement = ArgumentCaptor.forClass(Statement.class);
@@ -108,7 +108,7 @@ void singleSetOfParametersWithAddTriggersBatchWithOneStatement() {
     statement.add();
 
     StepVerifier.create(Flux.from(statement.execute()).flatMap(r -> r.getRowsUpdated()))
-        .expectNext(7, 11, 13)
+        .expectNext(7L, 11L, 13L)
         .verifyComplete();
 
     ArgumentCaptor<List<Statement>> params = ArgumentCaptor.forClass(List.class);
@@ -139,7 +139,7 @@ void twoParameterSetsWithNoTrailingAddSendsTwoStatements() {
     statement.bind("three", "B333");
 
     StepVerifier.create(Flux.from(statement.execute()).flatMap(r -> r.getRowsUpdated()))
-        .expectNext(7, 11, 13)
+        .expectNext(7L, 11L, 13L)
         .verifyComplete();
 
     ArgumentCaptor<List<Statement>> params = ArgumentCaptor.forClass(List.class);
@@ -175,7 +175,7 @@ void twoParameterSetsWithTrailingAddSendsTwoStatements() {
     statement.add();
 
     StepVerifier.create(Flux.from(statement.execute()).flatMap(r -> r.getRowsUpdated()))
-        .expectNext(7, 11, 13)
+        .expectNext(7L, 11L, 13L)
         .verifyComplete();
 
     ArgumentCaptor<List<Statement>> params = ArgumentCaptor.forClass(List.class);

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/DatabaseClientReactiveAdapterTest.java

@@ -283,7 +283,7 @@ void runBatchDml() {
       this.adapter.runBatchDml(
           ImmutableList.of(Statement.of("UPDATE something"), Statement.of("UPDATE something else")))
         .flatMap(result -> result.getRowsUpdated())
-    ).expectNext(42, 81)
+    ).expectNext(42L, 81L)
         .verifyComplete();
   }
 
@@ -299,7 +299,7 @@ void runBatchDml_rowCountAboveMaxIntIsTruncated() {
         this.adapter.runBatchDml(
             ImmutableList.of(Statement.of("UPDATE something")))
             .flatMap(result -> result.getRowsUpdated())
-    ).expectNext(Integer.MAX_VALUE)
+    ).expectNext(Long.valueOf(Integer.MAX_VALUE))
         .verifyComplete();
   }
 }

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/SpannerBatchTest.java

@@ -78,7 +78,7 @@ void batchPassesCorrectQueriesToAdapter() {
             batch.add("UPDATE tbl SET col1=val1")
                 .add("UPDATE tbl SET col2=val2").execute()
         ).flatMap(r -> r.getRowsUpdated())
-    ).expectNext(35, 47)
+    ).expectNext(35L, 47L)
         .verifyComplete();
 
     ArgumentCaptor<List<Statement>> argCaptor = ArgumentCaptor.forClass(List.class);

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/SpannerClientLibraryConnectionTest.java

@@ -71,7 +71,7 @@ void batchUsesCorrectAdapter() {
         Flux.from(
             batch.add("UPDATE tbl SET col1=val1").execute()
         ).flatMap(r -> r.getRowsUpdated())
-    ).expectNext(35)
+    ).expectNext(35L)
     .verifyComplete();
 
     ArgumentCaptor<List<Statement>> argCaptor = ArgumentCaptor.forClass(List.class);

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/SpannerClientLibraryDdlStatementTest.java

@@ -59,7 +59,7 @@ void executeDdlAffectsZeroRows() {
 
     StepVerifier.create(
         statement.execute().flatMap((Result r) -> Mono.from(r.getRowsUpdated()))
-    ).expectNext(0)
+    ).expectNext(0L)
         .verifyComplete();
   }
 

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/SpannerClientLibraryDmlStatementTest.java

@@ -48,7 +48,7 @@ void executeSingleNoRowsUpdated() {
     StepVerifier.create(
             Flux.from(dmlStatement.execute()).flatMap(result -> result.getRowsUpdated())
     )
-        .expectNext(0)
+        .expectNext(0L)
         .verifyComplete();
   }
 
@@ -68,7 +68,7 @@ void executeMultiple() {
 
     StepVerifier.create(
             Flux.from(dmlStatement.execute()).flatMap(result -> result.getRowsUpdated()))
-        .expectNext(2, 5)
+        .expectNext(2L, 5L)
         .verifyComplete();
   }
 }

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/SpannerClientLibraryResultTest.java

@@ -41,7 +41,7 @@ void nullRowsNotAllowed() {
   void getRowsUpdatedReturnsCorrectNumber() {
     SpannerClientLibraryResult result = new SpannerClientLibraryResult(Flux.empty(), 42);
     StepVerifier.create(result.getRowsUpdated())
-        .expectNext(42)
+        .expectNext(42L)
         .verifyComplete();
   }
 

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/SpannerClientLibraryStatementTest.java

@@ -56,7 +56,7 @@ void executeSingleNoRowsUpdated() {
 
     StepVerifier.create(
         Flux.from(statement.execute()).flatMap(result -> result.getRowsUpdated()))
-        .expectNext(0)
+        .expectNext(0L)
         .verifyComplete();
 
     StepVerifier.create(

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/SpannerResultTest.java

@@ -50,7 +50,7 @@ public void setup() {
   void getRowsUpdatedTest() {
     StepVerifier.create(
         ((Mono) new SpannerClientLibraryResult(this.resultSet, 2).getRowsUpdated()))
-        .expectNext(2)
+        .expectNext(2L)
         .verifyComplete();
   }
 

pom.xml

@@ -86,7 +86,7 @@
 
     <google-cloud-bom.version>26.16.0</google-cloud-bom.version>
 
-    <r2dbc.version>0.9.0.RELEASE</r2dbc.version>
+    <r2dbc.version>1.0.0.RELEASE</r2dbc.version>
     <reactor.version>2022.0.7</reactor.version>
 
     <mockito.version>4.9.0</mockito.version>