diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowEqualityTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowEqualityTest.java
new file mode 100644
index 000000000000..5b8a095a3d09
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowEqualityTest.java
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://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 org.apache.beam.sdk.io.gcp.bigquery;
+
+import static org.apache.beam.sdk.io.gcp.bigquery.TableRowMatchers.isTableRowEqualTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
+
+import com.google.api.services.bigquery.model.TableRow;
+import java.util.Arrays;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for the {@link TableRowMatchers} class. */
+@RunWith(JUnit4.class)
+public class BigQueryTableRowEqualityTest {
+
+  @Test
+  public void testIdenticalRows() {
+    TableRow row1 = new TableRow().set("count", 1).set("name", "Alice");
+    TableRow row2 = new TableRow().set("count", 1).set("name", "Alice");
+    assertThat(row1, isTableRowEqualTo(row2));
+  }
+
+  @Test
+  public void testEmptyRows() {
+    TableRow row1 = new TableRow();
+    TableRow row2 = new TableRow();
+    assertThat(row1, isTableRowEqualTo(row2));
+  }
+
+  @Test
+  public void testIntegerVsString() {
+    TableRow rowWithInteger = new TableRow().set("count", 1);
+    TableRow rowWithString = new TableRow().set("count", "1");
+    assertThat(rowWithInteger, not(isTableRowEqualTo(rowWithString)));
+  }
+
+  @Test
+  public void testDoubleVsInteger() {
+    TableRow rowWithDouble = new TableRow().set("value", 1.0);
+    TableRow rowWithInteger = new TableRow().set("value", 1);
+    assertThat(rowWithDouble, not(isTableRowEqualTo(rowWithInteger)));
+  }
+
+  @Test
+  public void testBooleanVsString() {
+    TableRow rowWithBoolean = new TableRow().set("active", true);
+    TableRow rowWithString = new TableRow().set("active", "true");
+    assertThat(rowWithBoolean, not(isTableRowEqualTo(rowWithString)));
+  }
+
+  @Test
+  public void testBothFieldsNull() {
+    TableRow row1 = new TableRow().set("name", null);
+    TableRow row2 = new TableRow().set("name", null);
+    assertThat(row1, isTableRowEqualTo(row2));
+  }
+
+  @Test
+  public void testNullVsNonNull() {
+    TableRow rowWithNull = new TableRow().set("name", null);
+    TableRow rowWithValue = new TableRow().set("name", "Alice");
+    assertThat(rowWithNull, not(isTableRowEqualTo(rowWithValue)));
+  }
+
+  @Test
+  public void testEmptyStringVsNull() {
+    TableRow rowWithEmptyString = new TableRow().set("name", "");
+    TableRow rowWithNull = new TableRow().set("name", null);
+    assertThat(rowWithEmptyString, not(isTableRowEqualTo(rowWithNull)));
+  }
+
+  @Test
+  public void testWhitespaceDifference() {
+    TableRow rowWithoutSpace = new TableRow().set("name", "Alice");
+    TableRow rowWithLeadingSpace = new TableRow().set("name", " Alice");
+    assertThat(rowWithoutSpace, not(isTableRowEqualTo(rowWithLeadingSpace)));
+  }
+
+  @Test
+  public void testDifferentFieldCount() {
+    TableRow rowWithTwoFields = new TableRow().set("a", 1).set("b", 2);
+    TableRow rowWithOneField = new TableRow().set("a", 1);
+    assertThat(rowWithTwoFields, not(isTableRowEqualTo(rowWithOneField)));
+  }
+
+  @Test
+  public void testMissingField() {
+    TableRow rowWithFieldB = new TableRow().set("a", 1).set("b", 2);
+    TableRow rowWithFieldC = new TableRow().set("a", 1).set("c", 2);
+    assertThat(rowWithFieldB, not(isTableRowEqualTo(rowWithFieldC)));
+  }
+
+  @Test
+  public void testDifferentInsertionOrder() {
+    TableRow row1 = new TableRow().set("a", 1).set("b", 2);
+    TableRow row2 = new TableRow().set("b", 2).set("a", 1);
+    assertThat(row1, isTableRowEqualTo(row2));
+  }
+
+  @Test
+  public void testIdenticalNestedRows() {
+    TableRow innerRow1 = new TableRow().set("id", 42);
+    TableRow innerRow2 = new TableRow().set("id", 42);
+    TableRow outerRow1 = new TableRow().set("nested", innerRow1);
+    TableRow outerRow2 = new TableRow().set("nested", innerRow2);
+    assertThat(outerRow1, isTableRowEqualTo(outerRow2));
+  }
+
+  @Test
+  public void testNestedRowsWithTypeMismatch() {
+    TableRow innerRowWithInteger = new TableRow().set("id", 42);
+    TableRow innerRowWithString = new TableRow().set("id", "42");
+    TableRow outerRow1 = new TableRow().set("nested", innerRowWithInteger);
+    TableRow outerRow2 = new TableRow().set("nested", innerRowWithString);
+    assertThat(outerRow1, not(isTableRowEqualTo(outerRow2)));
+  }
+
+  @Test
+  public void testDeeplyNestedRowsWithTypeMismatch() {
+    TableRow level3WithInteger = new TableRow().set("val", 1);
+    TableRow level3WithString = new TableRow().set("val", "1");
+    TableRow level2Row1 = new TableRow().set("l2", level3WithInteger);
+    TableRow level2Row2 = new TableRow().set("l2", level3WithString);
+    TableRow level1Row1 = new TableRow().set("l1", level2Row1);
+    TableRow level1Row2 = new TableRow().set("l1", level2Row2);
+    assertThat(level1Row1, not(isTableRowEqualTo(level1Row2)));
+  }
+
+  @Test
+  public void testIdenticalListFields() {
+    TableRow row1 = new TableRow().set("tags", Arrays.asList("a", "b"));
+    TableRow row2 = new TableRow().set("tags", Arrays.asList("a", "b"));
+    assertThat(row1, isTableRowEqualTo(row2));
+  }
+
+  @Test
+  public void testListFieldsWithDifferentOrder() {
+    TableRow row1 = new TableRow().set("tags", Arrays.asList("a", "b"));
+    TableRow row2 = new TableRow().set("tags", Arrays.asList("b", "a"));
+    assertThat(row1, not(isTableRowEqualTo(row2)));
+  }
+
+  @Test
+  public void testZeroIntegerVsZeroDouble() {
+    TableRow rowWithZeroInteger = new TableRow().set("value", 0);
+    TableRow rowWithZeroDouble = new TableRow().set("value", 0.0);
+    assertThat(rowWithZeroInteger, not(isTableRowEqualTo(rowWithZeroDouble)));
+  }
+
+  @Test
+  public void testNegativeNumbers() {
+    TableRow row1 = new TableRow().set("temp", -10);
+    TableRow row2 = new TableRow().set("temp", -10);
+    assertThat(row1, isTableRowEqualTo(row2));
+  }
+
+  @Test
+  public void testLongVsInteger() {
+    TableRow rowWithLong = new TableRow().set("count", 1L);
+    TableRow rowWithInteger = new TableRow().set("count", 1);
+    assertThat(rowWithLong, not(isTableRowEqualTo(rowWithInteger)));
+  }
+
+  @Test
+  public void testTrueVsFalse() {
+    TableRow rowWithTrue = new TableRow().set("active", true);
+    TableRow rowWithFalse = new TableRow().set("active", false);
+    assertThat(rowWithTrue, not(isTableRowEqualTo(rowWithFalse)));
+  }
+
+  @Test
+  public void testLargeIntegerVsLong() {
+    TableRow rowWithInteger = new TableRow().set("big", Integer.MAX_VALUE);
+    TableRow rowWithLong = new TableRow().set("big", (long) Integer.MAX_VALUE);
+    assertThat(rowWithInteger, not(isTableRowEqualTo(rowWithLong)));
+  }
+
+  @Test
+  public void testMultipleFieldsWithOneTypeMismatch() {
+    TableRow row1 = new TableRow().set("id", 1).set("name", "Alice").set("score", 99);
+    TableRow row2 = new TableRow().set("id", 1).set("name", "Alice").set("score", "99");
+    assertThat(row1, not(isTableRowEqualTo(row2)));
+  }
+}
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowMatchers.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowMatchers.java
new file mode 100644
index 000000000000..988b1bd0842c
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowMatchers.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://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 org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.google.api.services.bigquery.model.TableRow;
+import java.util.stream.Collectors;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeMatcher;
+
+class TableRowMatchers {
+
+  static Matcher<TableRow> isTableRowEqualTo(TableRow expected) {
+    return new TypeSafeMatcher<TableRow>() {
+
+      @Override
+      protected boolean matchesSafely(TableRow actual) {
+        return rowsMatch(expected, actual);
+      }
+
+      @Override
+      public void describeTo(Description description) {
+        description.appendText("TableRow (strict) ");
+        description.appendText(formatValue(expected, 0));
+      }
+
+      @Override
+      protected void describeMismatchSafely(TableRow actual, Description mismatch) {
+        describeRowMismatch(expected, actual, mismatch);
+      }
+    };
+  }
+
+  private static boolean rowsMatch(TableRow expected, TableRow actual) {
+    if (expected == null && actual == null) {
+      return true;
+    }
+
+    if (expected == null || actual == null) {
+      return false;
+    }
+
+    if (actual.size() != expected.size()) {
+      return false;
+    }
+
+    for (String key : expected.keySet()) {
+      if (!actual.containsKey(key)) {
+        return false;
+      }
+
+      Object expectedVal = expected.get(key);
+      Object actualVal = actual.get(key);
+
+      if (expectedVal == null && actualVal == null) {
+        continue;
+      }
+      if (expectedVal == null || actualVal == null) {
+        return false;
+      }
+
+      // recursively compare nested TableRows
+      if (expectedVal instanceof TableRow && actualVal instanceof TableRow) {
+        if (!rowsMatch((TableRow) expectedVal, (TableRow) actualVal)) {
+          return false;
+        }
+        continue;
+      }
+
+      if (!expectedVal.getClass().equals(actualVal.getClass())) {
+        return false;
+      }
+
+      if (!expectedVal.equals(actualVal)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static void describeRowMismatch(
+      TableRow expected, TableRow actual, Description mismatch) {
+
+    // size mismatch
+    if (actual.size() != expected.size()) {
+      mismatch.appendText(
+          String.format(
+              "had %d field(s) %s but expected %d field(s) %s",
+              actual.size(), actual.keySet(), expected.size(), expected.keySet()));
+      return;
+    }
+
+    // missing field
+    for (String key : expected.keySet()) {
+      if (!actual.containsKey(key)) {
+        mismatch.appendText(String.format("missing field '%s'", key));
+        return;
+      }
+    }
+
+    // value/type mismatch
+    for (String key : expected.keySet()) {
+      Object expectedVal = expected.get(key);
+      Object actualVal = actual.get(key);
+
+      if (expectedVal == null && actualVal == null) {
+        continue;
+      }
+      ;
+
+      if (expectedVal == null) {
+        mismatch.appendText(
+            String.format(
+                "field '%s': expected null but was %s(%s)",
+                key, actualVal.getClass().getSimpleName(), formatValue(actualVal, 0)));
+        return;
+      }
+
+      if (actualVal == null) {
+        mismatch.appendText(
+            String.format(
+                "field '%s': expected %s(%s) but was null",
+                key, expectedVal.getClass().getSimpleName(), formatValue(expectedVal, 0)));
+        return;
+      }
+
+      // recurse into nested TableRows
+      if (expectedVal instanceof TableRow && actualVal instanceof TableRow) {
+        if (!rowsMatch((TableRow) expectedVal, (TableRow) actualVal)) {
+          mismatch.appendText(String.format("field '%s': ", key));
+          describeRowMismatch((TableRow) expectedVal, (TableRow) actualVal, mismatch);
+          return;
+        }
+        continue;
+      }
+
+      // type mismatch
+      if (!expectedVal.getClass().equals(actualVal.getClass())) {
+        mismatch.appendText(
+            String.format(
+                "field '%s': expected %s(%s) but was %s(%s)",
+                key,
+                expectedVal.getClass().getSimpleName(),
+                formatValue(expectedVal, 0),
+                actualVal.getClass().getSimpleName(),
+                formatValue(actualVal, 0)));
+        return;
+      }
+
+      // value mismatch
+      if (!expectedVal.equals(actualVal)) {
+        mismatch.appendText(
+            String.format(
+                "field '%s': expected value (%s) but was (%s)",
+                key, formatValue(expectedVal, 0), formatValue(actualVal, 0)));
+        return;
+      }
+    }
+  }
+
+  // recursive formatter
+  private static String formatValue(Object val, int depth) {
+    if (val == null) {
+      return "null";
+    }
+
+    // safety net against infinite recursion
+    if (depth > 10) {
+      return "...";
+    }
+    if (val instanceof TableRow) {
+      TableRow row = (TableRow) val;
+      String fields =
+          row.keySet().stream()
+              .map(
+                  k -> {
+                    Object v = row.get(k);
+                    String typeName =
+                        v == null
+                            ? "null"
+                            : v instanceof TableRow ? "TableRow" : v.getClass().getSimpleName();
+                    return String.format("%s(%s)=%s", k, typeName, formatValue(v, depth + 1));
+                  })
+              .collect(Collectors.joining(", "));
+      return "TableRow{" + fields + "}";
+    }
+    return String.valueOf(val);
+  }
+}