Statement parameter binding

src/main/java/com/google/cloud/spanner/r2dbc/SpannerConnection.java

@@ -100,7 +100,8 @@ public Publisher<Void> createSavepoint(String s) {
 
   @Override
   public Statement createStatement(String sql) {
-    return new SpannerStatement(this.client, this.session, this.currentTransaction, sql);
+    return new SpannerStatement(this.client, this.session,
+        SpannerTransactionContext.from(this.currentTransaction), sql);
   }
 
   @Override

src/main/java/com/google/cloud/spanner/r2dbc/SpannerStatement.java

@@ -16,13 +16,26 @@
 
 package com.google.cloud.spanner.r2dbc;
 
+import static java.util.Objects.requireNonNull;
+
 import com.google.cloud.spanner.r2dbc.client.Client;
+import com.google.cloud.spanner.r2dbc.codecs.Codec;
+import com.google.cloud.spanner.r2dbc.codecs.Codecs;
+import com.google.cloud.spanner.r2dbc.codecs.DefaultCodecs;
 import com.google.cloud.spanner.r2dbc.result.PartialResultRowExtractor;
+import com.google.protobuf.Struct;
+import com.google.protobuf.Struct.Builder;
 import com.google.spanner.v1.PartialResultSet;
 import com.google.spanner.v1.Session;
-import com.google.spanner.v1.Transaction;
+import com.google.spanner.v1.Type;
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Statement;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
 import javax.annotation.Nullable;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -37,16 +50,23 @@ public class SpannerStatement implements Statement {
 
   private Session session;
 
-  private Transaction transaction;
+  private SpannerTransactionContext transaction;
 
   private String sql;
 
+  private LinkedList<Map<String, Object>> bindings = new LinkedList<>();
+
+  private Codecs codecs = new DefaultCodecs();
+
+  private Map<String, Type> types = Collections.EMPTY_MAP;
+
+  private Map<String, Codec> resolvedCodecs = new HashMap<>();
+
   /**
    * Creates a Spanner statement for a given SQL statement.
    *
    * <p>If no transaction is present, a temporary strongly consistent readonly transaction will be
    * used.
-   *
    * @param client cloud spanner client to use for performing the query operation
    * @param session current cloud spanner session
    * @param transaction current cloud spanner transaction, or empty if no transaction is started
@@ -55,45 +75,88 @@ public class SpannerStatement implements Statement {
   public SpannerStatement(
       Client client,
       Session session,
-      @Nullable Transaction transaction,
+      @Nullable SpannerTransactionContext transaction,
       String sql) {
 
     this.client = client;
     this.session = session;
     this.transaction = transaction;
     this.sql = sql;
+    add();
   }
 
   @Override
   public Statement add() {
-    return null;
+    this.bindings.add(new HashMap<>());
+    return this;
   }
 
   @Override
-  public Statement bind(Object o, Object o1) {
-    return null;
+  public Statement bind(Object identifier, Object value) {
+    requireNonNull(identifier);
+    if (identifier instanceof String) {
+      this.bindings.getLast().put((String)identifier, value);
+      return this;
+    }
+    throw new IllegalArgumentException("Only String identifiers are supported");
   }
 
   @Override
   public Statement bind(int i, Object o) {
-    return null;
+    throw new IllegalArgumentException("Only named parameters are supported");
   }
 
   @Override
-  public Statement bindNull(Object o, Class<?> type) {
-    return null;
+  public Statement bindNull(Object identifier, Class<?> type) {
+    return bind(identifier, null);
   }
 
   @Override
   public Statement bindNull(int i, Class<?> type) {
-    return null;
+    throw new IllegalArgumentException("Only named parameters are supported");
   }
 
   @Override
   public Publisher<? extends Result> execute() {
+    List<Struct> paramsStructs = new ArrayList<>();
+    for (Map<String, Object> bindingsBatch : this.bindings) {
+      Builder paramsStructBuilder = Struct.newBuilder();
+      Map<String, Type> types = this.types.isEmpty() ? new HashMap<>() : null;
+
+      for (Map.Entry<String, Object> binding : bindingsBatch.entrySet()) {
+        String paramName = binding.getKey();
+        Codec codec = this.resolvedCodecs.computeIfAbsent(paramName,
+            name -> this.codecs.getCodec(binding.getValue()));
+        paramsStructBuilder
+            .putFields(paramName, codec.encode(binding.getValue()));
+
+        if (this.types.isEmpty()) {
+          types.put(paramName,
+              Type.newBuilder().setCode(codec.getTypeCode()).build());
+        }
+
+      }
+
+      if (types != null) {
+        this.types = types;
+      }
+
+      paramsStructs.add(paramsStructBuilder.build());
+    }
+
+    Flux<Struct> structFlux = Flux.fromIterable(paramsStructs);
+
+    if (this.sql != null && this.sql.trim().toLowerCase().startsWith("select")) {
+      return structFlux.flatMap(this::runSingleStatement);
+    }
+    return structFlux.concatMapDelayError(this::runSingleStatement);
+  }
+
+  private Mono<? extends Result> runSingleStatement(Struct params) {
     PartialResultRowExtractor partialResultRowExtractor = new PartialResultRowExtractor();
 
-    return this.client.executeStreamingSql(this.session, this.transaction, this.sql)
+    return this.client
+        .executeStreamingSql(this.session, this.transaction, this.sql, params, this.types)
         .switchOnFirst((signal, flux) -> {
           if (signal.hasError()) {
             return Mono.error(signal.getThrowable());

src/main/java/com/google/cloud/spanner/r2dbc/SpannerTransactionContext.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2019 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.r2dbc;
+
+import com.google.spanner.v1.Transaction;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.Nullable;
+
+/**
+ * A class to hold transaction-related data.
+ */
+public class SpannerTransactionContext {
+
+  private Transaction transaction;
+
+  private AtomicLong seqNum = new AtomicLong(0);
+
+  private SpannerTransactionContext(Transaction transaction) {
+    this.transaction = transaction;
+  }
+
+  public Transaction getTransaction() {
+    return this.transaction;
+  }
+
+  public long nextSeqNum() {
+    return this.seqNum.getAndIncrement();
+  }
+
+  /**
+   * Creates the SpannerTransactionContext.
+   * @param transaction spanner transaction
+   * @return spanner transaction context
+   */
+  public static @Nullable SpannerTransactionContext from(Transaction transaction) {
+    if (transaction == null) {
+      return null;
+    }
+    return new SpannerTransactionContext(transaction);
+  }
+}

src/main/java/com/google/cloud/spanner/r2dbc/client/Client.java

@@ -16,10 +16,14 @@
 
 package com.google.cloud.spanner.r2dbc.client;
 
+import com.google.cloud.spanner.r2dbc.SpannerTransactionContext;
+import com.google.protobuf.Struct;
 import com.google.spanner.v1.CommitResponse;
 import com.google.spanner.v1.PartialResultSet;
 import com.google.spanner.v1.Session;
 import com.google.spanner.v1.Transaction;
+import com.google.spanner.v1.Type;
+import java.util.Map;
 import javax.annotation.Nullable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -71,7 +75,13 @@ public interface Client {
    * Execute a streaming query and get partial results.
    */
   Flux<PartialResultSet> executeStreamingSql(
-      Session session, @Nullable Transaction transaction, String sql);
+      Session session, @Nullable SpannerTransactionContext transaction, String sql, Struct params,
+      Map<String, Type> types);
+
+  default Flux<PartialResultSet> executeStreamingSql(
+      Session session, @Nullable SpannerTransactionContext transaction, String sql) {
+    return  executeStreamingSql(session, transaction, sql, null, null);
+  }
 
   /**
    * Release any resources held by the {@link Client}.

src/main/java/com/google/cloud/spanner/r2dbc/client/GrpcClient.java

@@ -17,9 +17,11 @@
 package com.google.cloud.spanner.r2dbc.client;
 
 import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.spanner.r2dbc.SpannerTransactionContext;
 import com.google.cloud.spanner.r2dbc.util.ObservableReactiveUtil;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.Empty;
+import com.google.protobuf.Struct;
 import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.CommitRequest;
 import com.google.spanner.v1.CommitResponse;
@@ -35,10 +37,12 @@
 import com.google.spanner.v1.TransactionOptions;
 import com.google.spanner.v1.TransactionOptions.ReadWrite;
 import com.google.spanner.v1.TransactionSelector;
+import com.google.spanner.v1.Type;
 import io.grpc.CallCredentials;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.auth.MoreCallCredentials;
+import java.util.Map;
 import javax.annotation.Nullable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -155,20 +159,27 @@ public Mono<Void> deleteSession(Session session) {
     });
   }
 
-  // TODO: add information about parameters being added to signature
   @Override
   public Flux<PartialResultSet> executeStreamingSql(
-      Session session, @Nullable Transaction transaction, String sql) {
+      Session session, @Nullable SpannerTransactionContext transactionContext, String sql,
+      Struct params, Map<String, Type> types) {
 
     return Flux.defer(() -> {
       ExecuteSqlRequest.Builder executeSqlRequest =
           ExecuteSqlRequest.newBuilder()
               .setSql(sql)
               .setSession(session.getName());
+      if (params != null) {
+        executeSqlRequest
+            .setParams(params)
+            .putAllParamTypes(types);
+      }
 
-      if (transaction != null) {
+      if (transactionContext != null && transactionContext.getTransaction() != null) {
         executeSqlRequest.setTransaction(
-            TransactionSelector.newBuilder().setId(transaction.getId()).build());
+            TransactionSelector.newBuilder().setId(transactionContext.getTransaction().getId())
+                .build());
+        executeSqlRequest.setSeqno(transactionContext.nextSeqNum());
       }
 
       return ObservableReactiveUtil.streamingCall(

src/main/java/com/google/cloud/spanner/r2dbc/codecs/Codec.java

@@ -18,9 +18,10 @@
 
 import com.google.protobuf.Value;
 import com.google.spanner.v1.Type;
+import com.google.spanner.v1.TypeCode;
 import reactor.util.annotation.Nullable;
 
-interface Codec<T> {
+public interface Codec<T> {
 
   /**
    * Indicates if the codec can decode a value.
@@ -80,4 +81,5 @@ interface Codec<T> {
    */
   Class<?> type();
 
+  TypeCode getTypeCode();
 }

src/main/java/com/google/cloud/spanner/r2dbc/codecs/Codecs.java

@@ -46,4 +46,7 @@ public interface Codecs {
    * @throws NullPointerException if {@code value} is {@code null}
    */
   Value encode(Object value);
+
+  Codec getCodec(Object value);
+
 }

src/main/java/com/google/cloud/spanner/r2dbc/codecs/DefaultCodecs.java

@@ -99,17 +99,22 @@ public <T> T decode(Value value, Type spannerType, Class<? extends T> type) {
   }
 
   @Override
-  public Value encode(Object value) {
-    if (value == null) {
-      return NULL_VALUE;
-    }
+  public Codec getCodec(Object value) {
     for (Codec<?> codec : this.codecs) {
       if (codec.canEncode(value)) {
-        return codec.encode(value);
+        return codec;
       }
     }
-
     throw new IllegalArgumentException(
-        String.format("Cannot encode parameter of type %s", value.getClass().getName()));
+          String.format("Cannot encode parameter of type %s", value.getClass().getName()));
+  }
+
+  @Override
+  public Value encode(Object value) {
+    if (value == null) {
+      return NULL_VALUE;
+    }
+    Codec codec = getCodec(value);
+    return codec.encode(value);
   }
 }

src/main/java/com/google/cloud/spanner/r2dbc/codecs/SpannerCodec.java

@@ -82,6 +82,11 @@ private boolean doCanDecode(Type dataType) {
     return dataType.getCode() == this.typeCode;
   }
 
+  @Override
+  public TypeCode getTypeCode() {
+    return this.typeCode;
+  }
+
   T doDecode(Value value, Type spannerType, Class<? extends T> type) {
     return (T) ValueUtils.decodeValue(spannerType, value);
   }