perf(bigquery): [T1-8] avoid redundant per-row FieldValueList re-wrapping in TableResult

java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/FieldValueList.java

@@ -23,6 +23,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Objects;
 
 /**
  * Google BigQuery Table Field Values class, which represents a row in returned query result (table
@@ -77,6 +78,10 @@ public boolean hasSchema() {
     return schema != null;
   }
 
+  FieldList getSchema() {
+    return schema;
+  }
+
   /** Returns the total number of field values in the row. */
   @Override
   public int size() {
@@ -108,6 +113,12 @@ public static FieldValueList of(List<FieldValue> row, Field... schema) {
   }
 
   FieldValueList withSchema(FieldList schema) {
+    // Rows produced by BigQueryImpl.transformTableData and ArrowDeserializer already carry the
+    // table or query FieldList. Returning this instance directly when the schema matches avoids
+    // allocating a duplicate FieldValueList wrapper and copying the row list on every traversal.
+    if (this.schema == schema || Objects.equals(this.schema, schema)) {
+      return this;
+    }
     return new FieldValueList(this.row, schema);
   }
 

java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/TableResult.java

@@ -26,6 +26,8 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Iterators;
 import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
 import java.util.Objects;
 import org.jspecify.annotations.Nullable;
 
@@ -211,19 +213,48 @@ public Iterable<FieldValueList> iterateAll() {
 
   @Override
   public Iterable<FieldValueList> getValues() {
-    return addSchema(getPageNoSchema().getValues());
+    Iterable<FieldValueList> values = getPageNoSchema().getValues();
+    Schema schema = getSchema();
+    if (schema == null) {
+      return values;
+    }
+    FieldList fields = schema.getFields();
+    // When the underlying page values are already materialized with the same FieldList (as in
+    // BigQueryImpl.transformTableData and ArrowDeserializer), return the underlying Collection
+    // directly to avoid lazy Iterables.transform wrapping and preserve Collection fast-paths.
+    if (values instanceof Collection) {
+      Collection<FieldValueList> collection = (Collection<FieldValueList>) values;
+      if (collection.isEmpty()) {
+        return values;
+      }
+      FieldValueList first =
+          (values instanceof List)
+              ? ((List<FieldValueList>) values).get(0)
+              : collection.iterator().next();
+      if (first != null
+          && (first.getSchema() == fields || Objects.equals(first.getSchema(), fields))) {
+        return values;
+      }
+    }
+    return addSchema(values, fields);
   }
 
   private Iterable<FieldValueList> addSchema(Iterable<FieldValueList> iter) {
-    if (getSchema() == null) {
+    Schema schema = getSchema();
+    if (schema == null) {
       return iter;
     }
+    return addSchema(iter, schema.getFields());
+  }
+
+  private static Iterable<FieldValueList> addSchema(
+      Iterable<FieldValueList> iter, final FieldList fields) {
     return Iterables.transform(
         iter,
         new Function<FieldValueList, FieldValueList>() {
           @Override
           public FieldValueList apply(FieldValueList list) {
-            return list.withSchema(getSchema().getFields());
+            return list.withSchema(fields);
           }
         });
   }

java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/FieldValueListTest.java

@@ -19,6 +19,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import com.google.api.client.util.Data;
@@ -225,4 +226,17 @@ void testGetNonExistentField() {
         assertThrows(IllegalArgumentException.class, () -> fieldValues.get("nonexistent"));
     assertNotNull(exception);
   }
+
+  @Test
+  void testWithSchemaShortCircuitsWhenSchemaMatches() {
+    assertSame(fieldValues, fieldValues.withSchema(schema));
+    FieldList equivalentSchema = FieldList.of(schema);
+    assertSame(fieldValues, fieldValues.withSchema(equivalentSchema));
+
+    FieldValueList noSchema = FieldValueList.fromPb(fieldValuesPb, null);
+    assertSame(noSchema, noSchema.withSchema(null));
+    FieldValueList attached = noSchema.withSchema(schema);
+    assertEquals(fieldValues, attached);
+    assertSame(attached, attached.withSchema(schema));
+  }
 }

java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/TableResultTest.java

@@ -260,4 +260,23 @@ void testEqualsAndHashCode() {
     assertThat(result1.toString()).contains("cacheHit=true");
     assertThat(result1.toString()).contains("sessionId=" + SESSION_ID_1);
   }
+
+  @Test
+  void testGetValuesPreservesCollectionAndRowIdentityWhenSchemaAlreadyAttached() {
+    FieldValueList row0 = newFieldValueList("0").withSchema(SCHEMA.getFields());
+    FieldValueList row1 = newFieldValueList("1").withSchema(SCHEMA.getFields());
+    ImmutableList<FieldValueList> pageRows = ImmutableList.of(row0, row1);
+    Page<FieldValueList> pageWithSchema = new PageImpl<>(null, null, pageRows);
+
+    TableResult result =
+        TableResult.newBuilder()
+            .setSchema(SCHEMA)
+            .setTotalRows(2L)
+            .setPageNoSchema(pageWithSchema)
+            .setRowsInPage(2L)
+            .build();
+
+    assertThat(result.getValues()).isSameInstanceAs(pageRows);
+    assertThat(result.iterateAll().iterator().next()).isSameInstanceAs(row0);
+  }
 }