feat: Add PreparedStatement and update ExecuteQuery API to use it (#2534)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/java-bigtable/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)
- [ ] Rollback plan is reviewed and LGTMed
- [ ] All new data plane features have a completed end to end testing plan

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://github.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: jackdingilian

google-cloud-bigtable/clirr-ignored-differences.xml

@@ -282,20 +282,6 @@
         <method>*getTimestamp(*)</method>
         <to>java.time.Instant</to>
     </difference>
-    <difference>
-        <!-- BetaApi was updated -->
-        <differenceType>7006</differenceType>
-        <className>com/google/cloud/bigtable/data/v2/models/sql/StructReader</className>
-        <method>*getTimestamp(*)</method>
-        <to>java.time.Instant</to>
-    </difference>
-    <difference>
-        <!-- BetaApi was updated -->
-        <differenceType>7005</differenceType>
-        <className>com/google/cloud/bigtable/data/v2/models/sql/Statement$Builder</className>
-        <method>*setTimestampParam(java.lang.String, org.threeten.bp.Instant)</method>
-        <to>*setTimestampParam(java.lang.String, java.time.Instant)</to>
-    </difference>
     <difference>
         <!-- ChangeStream api is internal, only used by apache/beam-->
         <differenceType>7013</differenceType>
@@ -320,4 +306,80 @@
         <className>com/google/cloud/bigtable/data/v2/stub/metrics/BigtableCloudMonitoringExporter</className>
         <method>*</method>
     </difference>
+    <difference>
+        <!-- BetaApi was updated -->
+        <differenceType>7005</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/BigtableDataClient</className>
+        <method>*executeQuery*</method>
+        <to>*</to>
+    </difference>
+    <difference>
+        <!-- BetaApi was renamed -->
+        <differenceType>8001</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/models/sql/Statement</className>
+        <method>*</method>
+    </difference>
+    <difference>
+        <!-- BetaApi was renamed -->
+        <differenceType>8001</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/models/sql/Statement$Builder</className>
+        <method>*</method>
+    </difference>
+    <difference>
+        <!-- BetaApi was renamed -->
+        <differenceType>8001</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/models/sql/Statement$Builder</className>
+        <method>*</method>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7004</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/internal/SqlRowMergerUtil</className>
+        <method>*</method>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7004</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/sql/ExecuteQueryCallContext</className>
+        <method>*ExecuteQueryCallContext*</method>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7009</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/sql/ExecuteQueryCallContext</className>
+        <method>*ExecuteQueryCallContext*</method>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7005</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/sql/ExecuteQueryCallContext</className>
+        <method>*create*</method>
+        <to>*</to>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7004</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/sql/ExecuteQueryCallable</className>
+        <method>*ExecuteQueryCallable*</method>
+        <to>*</to>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7005</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/sql/ExecuteQueryCallable</className>
+        <method>*call*</method>
+        <to>*</to>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>8001</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/sql/MetadataResolvingCallable</className>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7004</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/sql/SqlRowMerger</className>
+        <method>*</method>
+        <to>*</to>
+    </difference>
 </differences>

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClient.java

@@ -30,6 +30,9 @@
 import com.google.api.gax.rpc.ServerStream;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.api.gax.rpc.UnaryCallable;
+import com.google.cloud.bigtable.data.v2.internal.PrepareQueryRequest;
+import com.google.cloud.bigtable.data.v2.internal.PrepareResponse;
+import com.google.cloud.bigtable.data.v2.internal.PreparedStatementImpl;
 import com.google.cloud.bigtable.data.v2.internal.ResultSetImpl;
 import com.google.cloud.bigtable.data.v2.models.BulkMutation;
 import com.google.cloud.bigtable.data.v2.models.ChangeStreamRecord;
@@ -48,14 +51,17 @@
 import com.google.cloud.bigtable.data.v2.models.SampleRowKeysRequest;
 import com.google.cloud.bigtable.data.v2.models.TableId;
 import com.google.cloud.bigtable.data.v2.models.TargetId;
+import com.google.cloud.bigtable.data.v2.models.sql.BoundStatement;
+import com.google.cloud.bigtable.data.v2.models.sql.PreparedStatement;
 import com.google.cloud.bigtable.data.v2.models.sql.ResultSet;
-import com.google.cloud.bigtable.data.v2.models.sql.Statement;
+import com.google.cloud.bigtable.data.v2.models.sql.SqlType;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStub;
 import com.google.cloud.bigtable.data.v2.stub.sql.SqlServerStream;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.ByteString;
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
@@ -2705,30 +2711,61 @@ public void readChangeStreamAsync(
    * Executes a SQL Query and returns a ResultSet to iterate over the results. The returned
    * ResultSet instance is not threadsafe, it can only be used from single thread.
    *
+   * <p> The {@link BoundStatement} must be built from a {@link PreparedStatement} created using
+   * the same instance and app profile.
+   *
    * <p>Sample code:
    *
    * <pre>{@code
    * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
    *   String query = "SELECT CAST(cf['stringCol'] AS STRING) FROM [TABLE]";
-   *
-   *   try (ResultSet resultSet = bigtableDataClient.executeQuery(Statement.of(query))) {
-   *     while (resultSet.next()) {
-   *        String s = resultSet.getString("stringCol");
-   *        // do something with data
-   *     }
-   *   } catch (RuntimeException e) {
-   *     e.printStackTrace();
+   *   Map<String, SqlType<?>> paramTypes = new HashMap<>();
+   *   PreparedStatement preparedStatement = bigtableDataClient.prepareStatement(query, paramTypes));
+   *   // Ideally one PreparedStatement should be reused across requests
+   *   BoundStatement boundStatement = preparedStatement.bind()
+   *      // set any query params before calling build
+   *      .build();
+   *   try (ResultSet resultSet = bigtableDataClient.executeQuery(boundStatement)) {
+   *       while (resultSet.next()) {
+   *           String s = resultSet.getString("stringCol");
+   *            // do something with data
+   *       }
+   *    } catch (RuntimeException e) {
+   *        e.printStackTrace();
    *   }
    * }</pre>
    *
-   * @see Statement For query options.
+   * @see {@link PreparedStatement} & {@link BoundStatement} for query options.
    */
-  @BetaApi
-  public ResultSet executeQuery(Statement statement) {
-    SqlServerStream stream = stub.createExecuteQueryCallable().call(statement);
+  public ResultSet executeQuery(BoundStatement boundStatement) {
+    boundStatement.assertUsingSameStub(stub);
+    SqlServerStream stream = stub.createExecuteQueryCallable().call(boundStatement);
     return ResultSetImpl.create(stream);
   }
 
+  /**
+   * Prepares a query for execution. If possible this should be called once and reused across
+   * requests. This will amortize the cost of query preparation.
+   *
+   * <p>A parameterized query should contain placeholders in the form of {@literal @} followed by
+   * the parameter name. Parameter names may consist of any combination of letters, numbers, and
+   * underscores.
+   *
+   * <p>Parameters can appear anywhere that a literal value is expected. The same parameter name can
+   * be used more than once, for example: {@code WHERE cf["qualifier1"] = @value OR cf["qualifier2"]
+   * = @value }
+   *
+   * @param query sql query string to prepare
+   * @param paramTypes a Map of the parameter names and the corresponding {@link SqlType} for all
+   *     query parameters in 'query'
+   * @return {@link PreparedStatement} which is used to create {@link BoundStatement}s to execute
+   */
+  public PreparedStatement prepareStatement(String query, Map<String, SqlType<?>> paramTypes) {
+    PrepareQueryRequest request = PrepareQueryRequest.create(query, paramTypes);
+    PrepareResponse response = stub.prepareQueryCallable().call(request);
+    return PreparedStatementImpl.create(response, paramTypes, request, stub);
+  }
+
   /** Close the clients and releases all associated resources. */
   @Override
   public void close() {

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/AbstractProtoStructReader.java

@@ -24,7 +24,6 @@
 import com.google.cloud.bigtable.data.v2.models.sql.StructReader;
 import com.google.common.base.Preconditions;
 import com.google.protobuf.ByteString;
-import com.google.protobuf.Timestamp;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -169,15 +168,15 @@ public boolean getBoolean(String columnName) {
   public Instant getTimestamp(int columnIndex) {
     checkNonNullOfType(columnIndex, SqlType.timestamp(), columnIndex);
     Value value = values().get(columnIndex);
-    return toInstant(value.getTimestampValue());
+    return TimestampUtil.toInstant(value.getTimestampValue());
   }
 
   @Override
   public Instant getTimestamp(String columnName) {
     int columnIndex = getColumnIndex(columnName);
     checkNonNullOfType(columnIndex, SqlType.timestamp(), columnName);
     Value value = values().get(columnIndex);
-    return toInstant(value.getTimestampValue());
+    return TimestampUtil.toInstant(value.getTimestampValue());
   }
 
   @Override
@@ -275,7 +274,7 @@ Object decodeValue(Value value, SqlType<?> type) {
       case BOOL:
         return value.getBoolValue();
       case TIMESTAMP:
-        return toInstant(value.getTimestampValue());
+        return TimestampUtil.toInstant(value.getTimestampValue());
       case DATE:
         return fromProto(value.getDateValue());
       case STRUCT:
@@ -329,10 +328,6 @@ private void checkNonNullOfType(
     }
   }
 
-  private Instant toInstant(Timestamp timestamp) {
-    return Instant.ofEpochSecond(timestamp.getSeconds(), timestamp.getNanos());
-  }
-
   private Date fromProto(com.google.type.Date proto) {
     return Date.fromYearMonthDay(proto.getYear(), proto.getMonth(), proto.getDay());
   }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/PrepareQueryRequest.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2025 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.bigtable.data.v2.internal;
+
+import com.google.api.core.InternalApi;
+import com.google.auto.value.AutoValue;
+import com.google.bigtable.v2.Type;
+import com.google.cloud.bigtable.data.v2.models.sql.SqlType;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Internal representation of PrepareQueryRequest that handles conversion from user-facing types to
+ * proto.
+ *
+ * <p>This is considered an internal implementation detail and should not be used by applications.
+ */
+@InternalApi("For internal use only")
+@AutoValue
+public abstract class PrepareQueryRequest {
+
+  public abstract String query();
+
+  public abstract Map<String, SqlType<?>> paramTypes();
+
+  public static PrepareQueryRequest create(String query, Map<String, SqlType<?>> paramTypes) {
+    return new AutoValue_PrepareQueryRequest(query, paramTypes);
+  }
+
+  public com.google.bigtable.v2.PrepareQueryRequest toProto(RequestContext requestContext) {
+    HashMap<String, Type> protoParamTypes = new HashMap<>(paramTypes().size());
+    for (Map.Entry<String, SqlType<?>> entry : paramTypes().entrySet()) {
+      Type proto = QueryParamUtil.convertToQueryParamProto(entry.getValue());
+      protoParamTypes.put(entry.getKey(), proto);
+    }
+
+    return com.google.bigtable.v2.PrepareQueryRequest.newBuilder()
+        .setInstanceName(
+            NameUtil.formatInstanceName(
+                requestContext.getProjectId(), requestContext.getInstanceId()))
+        .setAppProfileId(requestContext.getAppProfileId())
+        .setQuery(query())
+        .putAllParamTypes(protoParamTypes)
+        .build();
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/PrepareResponse.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2025 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.bigtable.data.v2.internal;
+
+import com.google.api.core.InternalApi;
+import com.google.auto.value.AutoValue;
+import com.google.bigtable.v2.PrepareQueryResponse;
+import com.google.cloud.bigtable.data.v2.models.sql.ResultSetMetadata;
+import com.google.protobuf.ByteString;
+import java.time.Instant;
+
+/**
+ * Wrapper for results of a PrepareQuery call.
+ *
+ * <p>This should only be managed by {@link
+ * com.google.cloud.bigtable.data.v2.models.sql.PreparedStatement}, and never used directly by users
+ *
+ * <p>This is considered an internal implementation detail and should not be used by applications.
+ */
+@InternalApi("For internal use only")
+@AutoValue
+public abstract class PrepareResponse {
+  public abstract ResultSetMetadata resultSetMetadata();
+
+  public abstract ByteString preparedQuery();
+
+  public abstract Instant validUntil();
+
+  public static PrepareResponse fromProto(PrepareQueryResponse proto) {
+    ResultSetMetadata metadata = ProtoResultSetMetadata.fromProto(proto.getMetadata());
+    Instant validUntil = TimestampUtil.toInstant(proto.getValidUntil());
+    return new AutoValue_PrepareResponse(metadata, proto.getPreparedQuery(), validUntil);
+  }
+}