diff --git a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java
index f99ba0bbe5dd..7264b3349772 100644
--- a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java
+++ b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java
@@ -1011,7 +1011,7 @@ public void testFindMatchingBigQueryObjects_Routines_ListWithPattern() throws Ex
 
     Page<Routine> page = mock(Page.class);
     when(page.iterateAll()).thenReturn(Arrays.asList(proc1, func1, proc2, otherProc));
-    when(bigqueryClient.listRoutines(eq(datasetId), any(BigQuery.RoutineListOption.class)))
+    when(bigqueryClient.listRoutines(eq(datasetId), any(BigQuery.RoutineListOption[].class)))
         .thenReturn(page);
 
     Pattern regex = dbMetadata.compileSqlLikePattern(pattern);
@@ -1030,7 +1030,7 @@ public void testFindMatchingBigQueryObjects_Routines_ListWithPattern() throws Ex
             dbMetadata.LOG);
 
     verify(bigqueryClient, times(1))
-        .listRoutines(eq(datasetId), any(BigQuery.RoutineListOption.class));
+        .listRoutines(eq(datasetId), any(BigQuery.RoutineListOption[].class));
     verify(bigqueryClient, never()).getRoutine(any(RoutineId.class));
 
     assertNotNull(results);
@@ -1055,7 +1055,7 @@ public void testFindMatchingBigQueryObjects_Routines_ListNoPattern() throws Exce
 
     Page<Routine> page = mock(Page.class);
     when(page.iterateAll()).thenReturn(Arrays.asList(proc1, func1));
-    when(bigqueryClient.listRoutines(eq(datasetId), any(BigQuery.RoutineListOption.class)))
+    when(bigqueryClient.listRoutines(eq(datasetId), any(BigQuery.RoutineListOption[].class)))
         .thenReturn(page);
 
     Pattern regex = dbMetadata.compileSqlLikePattern(pattern);
@@ -1072,7 +1072,7 @@ public void testFindMatchingBigQueryObjects_Routines_ListNoPattern() throws Exce
             dbMetadata.LOG);
 
     verify(bigqueryClient, times(1))
-        .listRoutines(eq(datasetId), any(BigQuery.RoutineListOption.class));
+        .listRoutines(eq(datasetId), any(BigQuery.RoutineListOption[].class));
 
     assertNotNull(results);
     List<Routine> resultList = new ArrayList<>(results);
@@ -1108,7 +1108,7 @@ public void testFindMatchingBigQueryObjects_Routines_GetSpecific() throws Except
 
     verify(bigqueryClient, times(1)).getRoutine(eq(routineId));
     verify(bigqueryClient, never())
-        .listRoutines(any(DatasetId.class), any(BigQuery.RoutineListOption.class));
+        .listRoutines(any(DatasetId.class), any(BigQuery.RoutineListOption[].class));
 
     assertNotNull(results);
     List<Routine> resultList = new ArrayList<>(results);
@@ -2931,11 +2931,14 @@ public void testGetSchemas_NoArgs_DelegatesCorrectly() throws SQLException {
     ResultSet mockResultSet = mock(ResultSet.class);
     doReturn(mockResultSet).when(spiedDbMetadata).getSchemas(null, null);
 
-    ResultSet rs = spiedDbMetadata.getSchemas();
+    try (ResultSet rs = spiedDbMetadata.getSchemas()) {
 
-    assertSame(
-        mockResultSet, rs, "The returned ResultSet should be the one from the two-argument method");
-    verify(spiedDbMetadata, times(1)).getSchemas(null, null);
+      assertSame(
+          mockResultSet,
+          rs,
+          "The returned ResultSet should be the one from the two-argument method");
+      verify(spiedDbMetadata, times(1)).getSchemas(null, null);
+    }
   }
 
   // Non-Resultset DatabaseMetadata tests
@@ -3262,18 +3265,19 @@ public void testGetCatalogs_WithProjectDiscovery() throws SQLException {
         .thenReturn(Arrays.asList("discovered-1", "discovered-2"));
     when(bigQueryConnection.getAdditionalProjects()).thenReturn("additional-1,additional-2");
 
-    ResultSet rs = dbMetadata.getCatalogs();
-    assertNotNull(rs);
+    try (ResultSet rs = dbMetadata.getCatalogs()) {
+      assertNotNull(rs);
 
-    List<String> catalogs = new ArrayList<>();
-    while (rs.next()) {
-      catalogs.add(rs.getString("TABLE_CAT"));
-    }
+      List<String> catalogs = new ArrayList<>();
+      while (rs.next()) {
+        catalogs.add(rs.getString("TABLE_CAT"));
+      }
 
-    assertThat(catalogs)
-        .containsExactly(
-            "additional-1", "additional-2", "discovered-1", "discovered-2", "primary-project")
-        .inOrder();
+      assertThat(catalogs)
+          .containsExactly(
+              "additional-1", "additional-2", "discovered-1", "discovered-2", "primary-project")
+          .inOrder();
+    }
   }
 
   @Test
@@ -3284,17 +3288,18 @@ public void testGetCatalogs_WithoutProjectDiscovery() throws SQLException {
         .thenReturn(Arrays.asList("discovered-1", "discovered-2"));
     when(bigQueryConnection.getAdditionalProjects()).thenReturn("additional-1,additional-2");
 
-    ResultSet rs = dbMetadata.getCatalogs();
-    assertNotNull(rs);
+    try (ResultSet rs = dbMetadata.getCatalogs()) {
+      assertNotNull(rs);
 
-    List<String> catalogs = new ArrayList<>();
-    while (rs.next()) {
-      catalogs.add(rs.getString("TABLE_CAT"));
-    }
+      List<String> catalogs = new ArrayList<>();
+      while (rs.next()) {
+        catalogs.add(rs.getString("TABLE_CAT"));
+      }
 
-    assertThat(catalogs)
-        .containsExactly("additional-1", "additional-2", "primary-project")
-        .inOrder();
+      assertThat(catalogs)
+          .containsExactly("additional-1", "additional-2", "primary-project")
+          .inOrder();
+    }
   }
 
   @Test
@@ -3307,37 +3312,39 @@ public void testGetSchemas_WithProjectDiscovery() throws SQLException {
     Page<Dataset> pagePrimary = mock(Page.class);
     Dataset dsPrimary = mockBigQueryDataset("primary-project", "dataset_p");
     when(pagePrimary.iterateAll()).thenReturn(Collections.singletonList(dsPrimary));
-    when(bigqueryClient.listDatasets(eq("primary-project"), any(BigQuery.DatasetListOption.class)))
+    when(bigqueryClient.listDatasets(
+            eq("primary-project"), any(BigQuery.DatasetListOption[].class)))
         .thenReturn(pagePrimary);
 
     Page<Dataset> pageAdditional = mock(Page.class);
     Dataset dsAdditional = mockBigQueryDataset("additional-1", "dataset_a");
     when(pageAdditional.iterateAll()).thenReturn(Collections.singletonList(dsAdditional));
-    when(bigqueryClient.listDatasets(eq("additional-1"), any(BigQuery.DatasetListOption.class)))
+    when(bigqueryClient.listDatasets(eq("additional-1"), any(BigQuery.DatasetListOption[].class)))
         .thenReturn(pageAdditional);
 
     Page<Dataset> pageDiscovered = mock(Page.class);
     Dataset dsDiscovered = mockBigQueryDataset("discovered-1", "dataset_d");
     when(pageDiscovered.iterateAll()).thenReturn(Collections.singletonList(dsDiscovered));
-    when(bigqueryClient.listDatasets(eq("discovered-1"), any(BigQuery.DatasetListOption.class)))
+    when(bigqueryClient.listDatasets(eq("discovered-1"), any(BigQuery.DatasetListOption[].class)))
         .thenReturn(pageDiscovered);
 
-    ResultSet rs = dbMetadata.getSchemas(null, null);
-    assertNotNull(rs);
+    try (ResultSet rs = dbMetadata.getSchemas(null, null)) {
+      assertNotNull(rs);
 
-    List<String> schemas = new ArrayList<>();
-    List<String> catalogs = new ArrayList<>();
-    while (rs.next()) {
-      schemas.add(rs.getString("TABLE_SCHEM"));
-      catalogs.add(rs.getString("TABLE_CATALOG"));
-    }
+      List<String> schemas = new ArrayList<>();
+      List<String> catalogs = new ArrayList<>();
+      while (rs.next()) {
+        schemas.add(rs.getString("TABLE_SCHEM"));
+        catalogs.add(rs.getString("TABLE_CATALOG"));
+      }
 
-    // Results are sorted by catalog (TABLE_CATALOG) then schema (TABLE_SCHEM)
-    // alphabetical catalog: "additional-1", "discovered-1", "primary-project"
-    assertThat(catalogs)
-        .containsExactly("additional-1", "discovered-1", "primary-project")
-        .inOrder();
-    assertThat(schemas).containsExactly("dataset_a", "dataset_d", "dataset_p").inOrder();
+      // Results are sorted by catalog (TABLE_CATALOG) then schema (TABLE_SCHEM)
+      // alphabetical catalog: "additional-1", "discovered-1", "primary-project"
+      assertThat(catalogs)
+          .containsExactly("additional-1", "discovered-1", "primary-project")
+          .inOrder();
+      assertThat(schemas).containsExactly("dataset_a", "dataset_d", "dataset_p").inOrder();
+    }
   }
 
   @Test
@@ -3350,31 +3357,294 @@ public void testGetSchemas_WithoutProjectDiscovery() throws SQLException {
     Page<Dataset> pagePrimary = mock(Page.class);
     Dataset dsPrimary = mockBigQueryDataset("primary-project", "dataset_p");
     when(pagePrimary.iterateAll()).thenReturn(Collections.singletonList(dsPrimary));
-    when(bigqueryClient.listDatasets(eq("primary-project"), any(BigQuery.DatasetListOption.class)))
+    when(bigqueryClient.listDatasets(
+            eq("primary-project"), any(BigQuery.DatasetListOption[].class)))
         .thenReturn(pagePrimary);
 
     Page<Dataset> pageAdditional = mock(Page.class);
     Dataset dsAdditional = mockBigQueryDataset("additional-1", "dataset_a");
     when(pageAdditional.iterateAll()).thenReturn(Collections.singletonList(dsAdditional));
-    when(bigqueryClient.listDatasets(eq("additional-1"), any(BigQuery.DatasetListOption.class)))
+    when(bigqueryClient.listDatasets(eq("additional-1"), any(BigQuery.DatasetListOption[].class)))
         .thenReturn(pageAdditional);
 
-    ResultSet rs = dbMetadata.getSchemas(null, null);
-    assertNotNull(rs);
+    try (ResultSet rs = dbMetadata.getSchemas(null, null)) {
+      assertNotNull(rs);
+
+      List<String> schemas = new ArrayList<>();
+      List<String> catalogs = new ArrayList<>();
+      while (rs.next()) {
+        schemas.add(rs.getString("TABLE_SCHEM"));
+        catalogs.add(rs.getString("TABLE_CATALOG"));
+      }
+
+      // Results are sorted by catalog (TABLE_CATALOG) then schema (TABLE_SCHEM)
+      // alphabetical catalog: "additional-1", "primary-project" (discovered-1 is ignored)
+      assertThat(catalogs).containsExactly("additional-1", "primary-project").inOrder();
+      assertThat(schemas).containsExactly("dataset_a", "dataset_p").inOrder();
+
+      verify(bigqueryClient, never())
+          .listDatasets(eq("discovered-1"), any(BigQuery.DatasetListOption[].class));
+    }
+  }
+
+  private void mockDatasetIteration(DatasetId datasetId) {
+    Page<Dataset> pagePrimary = mock(Page.class);
+    Dataset dsPrimary = mock(Dataset.class);
+    when(dsPrimary.getDatasetId()).thenReturn(datasetId);
+    when(pagePrimary.iterateAll()).thenReturn(Collections.singletonList(dsPrimary));
+    when(bigqueryClient.listDatasets(
+            eq(datasetId.getProject()), any(BigQuery.DatasetListOption[].class)))
+        .thenReturn(pagePrimary);
+  }
+
+  private Table mockTableWithConstraints(TableId tableId, TableConstraints constraints) {
+    Table mockTable = mock(Table.class);
+    when(mockTable.getTableId()).thenReturn(tableId);
+    TableDefinition mockDef = mock(TableDefinition.class);
+    when(mockDef.getType()).thenReturn(TableDefinition.Type.TABLE);
+    when(mockTable.getDefinition()).thenReturn(mockDef);
+    if (constraints != null) {
+      when(mockTable.getTableConstraints()).thenReturn(constraints);
+    }
+    when(bigqueryClient.getTable(eq(tableId))).thenReturn(mockTable);
+    return mockTable;
+  }
 
-    List<String> schemas = new ArrayList<>();
-    List<String> catalogs = new ArrayList<>();
-    while (rs.next()) {
-      schemas.add(rs.getString("TABLE_SCHEM"));
-      catalogs.add(rs.getString("TABLE_CATALOG"));
+  private void mockTableIteration(DatasetId datasetId, Table... tables) {
+    Page<Table> pageTables = mock(Page.class);
+    when(pageTables.iterateAll()).thenReturn(Arrays.asList(tables));
+    when(bigqueryClient.listTables(eq(datasetId), any(BigQuery.TableListOption[].class)))
+        .thenReturn(pageTables);
+  }
+
+  @Test
+  public void testGetPrimaryKeys_hasKey() throws SQLException {
+    DatasetId datasetId = DatasetId.of("test-project", "dataset_p");
+    TableId tableId = TableId.of("test-project", "dataset_p", "table_p");
+
+    TableConstraints mockConstraints = mock(TableConstraints.class);
+    PrimaryKey mockPk = mock(PrimaryKey.class);
+    when(mockPk.getColumns()).thenReturn(Arrays.asList("id_col1", "id_col2"));
+    when(mockConstraints.getPrimaryKey()).thenReturn(mockPk);
+
+    mockTableWithConstraints(tableId, mockConstraints);
+
+    try (ResultSet rs = dbMetadata.getPrimaryKeys("test-project", "dataset_p", "table_p")) {
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("TABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("TABLE_SCHEM"));
+      assertEquals("table_p", rs.getString("TABLE_NAME"));
+      assertEquals("id_col1", rs.getString("COLUMN_NAME"));
+      assertEquals(1, rs.getInt("KEY_SEQ"));
+      assertNull(rs.getString("PK_NAME"));
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("TABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("TABLE_SCHEM"));
+      assertEquals("table_p", rs.getString("TABLE_NAME"));
+      assertEquals("id_col2", rs.getString("COLUMN_NAME"));
+      assertEquals(2, rs.getInt("KEY_SEQ"));
+      assertNull(rs.getString("PK_NAME"));
+      assertFalse(rs.next());
     }
+  }
 
-    // Results are sorted by catalog (TABLE_CATALOG) then schema (TABLE_SCHEM)
-    // alphabetical catalog: "additional-1", "primary-project" (discovered-1 is ignored)
-    assertThat(catalogs).containsExactly("additional-1", "primary-project").inOrder();
-    assertThat(schemas).containsExactly("dataset_a", "dataset_p").inOrder();
+  @Test
+  public void testGetPrimaryKeys_noKeys() throws SQLException {
+    DatasetId datasetId = DatasetId.of("test-project", "dataset_p");
+    TableId tableId = TableId.of("test-project", "dataset_p", "table_p");
 
-    verify(bigqueryClient, never())
-        .listDatasets(eq("discovered-1"), any(BigQuery.DatasetListOption.class));
+    mockTableWithConstraints(tableId, null);
+
+    try (ResultSet rs = dbMetadata.getPrimaryKeys("test-project", "dataset_p", "table_p")) {
+      assertFalse(rs.next());
+    }
+  }
+
+  @Test
+  public void testGetImportedKeys_hasKeys() throws SQLException {
+    DatasetId datasetId = DatasetId.of("test-project", "dataset_p");
+    TableId tableId = TableId.of("test-project", "dataset_p", "table_p");
+
+    TableConstraints mockConstraints = mock(TableConstraints.class);
+    ForeignKey mockFk = mock(ForeignKey.class);
+    when(mockFk.getName()).thenReturn("fk_name");
+    when(mockFk.getReferencedTable())
+        .thenReturn(TableId.of("test-project", "dataset_p", "ref_table"));
+
+    ColumnReference mockRef1 = mock(ColumnReference.class);
+    when(mockRef1.getReferencingColumn()).thenReturn("fk_col1");
+    when(mockRef1.getReferencedColumn()).thenReturn("pk_col1");
+    ColumnReference mockRef2 = mock(ColumnReference.class);
+    when(mockRef2.getReferencingColumn()).thenReturn("fk_col2");
+    when(mockRef2.getReferencedColumn()).thenReturn("pk_col2");
+    when(mockFk.getColumnReferences()).thenReturn(Arrays.asList(mockRef1, mockRef2));
+
+    when(mockConstraints.getForeignKeys()).thenReturn(Collections.singletonList(mockFk));
+    mockTableWithConstraints(tableId, mockConstraints);
+
+    try (ResultSet rs = dbMetadata.getImportedKeys("test-project", "dataset_p", "table_p")) {
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("PKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("PKTABLE_SCHEM"));
+      assertEquals("ref_table", rs.getString("PKTABLE_NAME"));
+      assertEquals("pk_col1", rs.getString("PKCOLUMN_NAME"));
+      assertEquals("test-project", rs.getString("FKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("FKTABLE_SCHEM"));
+      assertEquals("table_p", rs.getString("FKTABLE_NAME"));
+      assertEquals("fk_col1", rs.getString("FKCOLUMN_NAME"));
+      assertEquals(1, rs.getInt("KEY_SEQ"));
+      assertEquals("fk_name", rs.getString("FK_NAME"));
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("PKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("PKTABLE_SCHEM"));
+      assertEquals("ref_table", rs.getString("PKTABLE_NAME"));
+      assertEquals("pk_col2", rs.getString("PKCOLUMN_NAME"));
+      assertEquals("test-project", rs.getString("FKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("FKTABLE_SCHEM"));
+      assertEquals("table_p", rs.getString("FKTABLE_NAME"));
+      assertEquals("fk_col2", rs.getString("FKCOLUMN_NAME"));
+      assertEquals(2, rs.getInt("KEY_SEQ"));
+      assertEquals("fk_name", rs.getString("FK_NAME"));
+      assertFalse(rs.next());
+    }
+  }
+
+  @Test
+  public void testGetImportedKeys_noKeys() throws SQLException {
+    DatasetId datasetId = DatasetId.of("test-project", "dataset_p");
+    TableId tableId = TableId.of("test-project", "dataset_p", "table_p");
+
+    mockTableWithConstraints(tableId, null);
+
+    try (ResultSet rs = dbMetadata.getImportedKeys("test-project", "dataset_p", "table_p")) {
+      assertFalse(rs.next());
+    }
+  }
+
+  @Test
+  public void testGetExportedKeys_hasKeys() throws SQLException {
+    DatasetId datasetId = DatasetId.of("test-project", "dataset_p");
+    TableId tableId = TableId.of("test-project", "dataset_p", "table_p");
+    TableId refTableId = TableId.of("test-project", "dataset_p", "ref_table");
+
+    TableConstraints mockConstraints = mock(TableConstraints.class);
+    ForeignKey mockFk = mock(ForeignKey.class);
+    when(mockFk.getName()).thenReturn("fk_name");
+    when(mockFk.getReferencedTable()).thenReturn(refTableId);
+    ColumnReference mockRef1 = mock(ColumnReference.class);
+    when(mockRef1.getReferencingColumn()).thenReturn("fk_col1");
+    when(mockRef1.getReferencedColumn()).thenReturn("pk_col1");
+    ColumnReference mockRef2 = mock(ColumnReference.class);
+    when(mockRef2.getReferencingColumn()).thenReturn("fk_col2");
+    when(mockRef2.getReferencedColumn()).thenReturn("pk_col2");
+    when(mockFk.getColumnReferences()).thenReturn(Arrays.asList(mockRef1, mockRef2));
+    when(mockConstraints.getForeignKeys()).thenReturn(Collections.singletonList(mockFk));
+
+    Table mockTableP = mockTableWithConstraints(tableId, mockConstraints);
+    mockDatasetIteration(datasetId);
+    mockTableIteration(datasetId, mockTableP);
+
+    try (ResultSet rs = dbMetadata.getExportedKeys("test-project", "dataset_p", "ref_table")) {
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("PKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("PKTABLE_SCHEM"));
+      assertEquals("ref_table", rs.getString("PKTABLE_NAME"));
+      assertEquals("pk_col1", rs.getString("PKCOLUMN_NAME"));
+      assertEquals("test-project", rs.getString("FKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("FKTABLE_SCHEM"));
+      assertEquals("table_p", rs.getString("FKTABLE_NAME"));
+      assertEquals("fk_col1", rs.getString("FKCOLUMN_NAME"));
+      assertEquals(1, rs.getInt("KEY_SEQ"));
+      assertEquals("fk_name", rs.getString("FK_NAME"));
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("PKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("PKTABLE_SCHEM"));
+      assertEquals("ref_table", rs.getString("PKTABLE_NAME"));
+      assertEquals("pk_col2", rs.getString("PKCOLUMN_NAME"));
+      assertEquals("test-project", rs.getString("FKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("FKTABLE_SCHEM"));
+      assertEquals("table_p", rs.getString("FKTABLE_NAME"));
+      assertEquals("fk_col2", rs.getString("FKCOLUMN_NAME"));
+      assertEquals(2, rs.getInt("KEY_SEQ"));
+      assertEquals("fk_name", rs.getString("FK_NAME"));
+      assertFalse(rs.next());
+    }
+  }
+
+  @Test
+  public void testGetExportedKeys_noKeys() throws SQLException {
+    DatasetId datasetId = DatasetId.of("test-project", "dataset_p");
+    TableId tableId = TableId.of("test-project", "dataset_p", "table_p");
+
+    Table mockTableP = mockTableWithConstraints(tableId, null);
+    mockDatasetIteration(datasetId);
+    mockTableIteration(datasetId, mockTableP);
+
+    try (ResultSet rs = dbMetadata.getExportedKeys("test-project", "dataset_p", "ref_table")) {
+      assertFalse(rs.next());
+    }
+  }
+
+  @Test
+  public void testGetCrossReference_hasKeys() throws SQLException {
+    TableId fkTableId = TableId.of("test-project", "dataset_p", "fk_table");
+    TableId pkTableId = TableId.of("test-project", "dataset_p", "pk_table");
+
+    TableConstraints mockConstraints = mock(TableConstraints.class);
+    ForeignKey mockFk = mock(ForeignKey.class);
+    when(mockFk.getName()).thenReturn("fk_name");
+    when(mockFk.getReferencedTable()).thenReturn(pkTableId);
+
+    ColumnReference mockRef1 = mock(ColumnReference.class);
+    when(mockRef1.getReferencingColumn()).thenReturn("fk_col1");
+    when(mockRef1.getReferencedColumn()).thenReturn("pk_col1");
+    ColumnReference mockRef2 = mock(ColumnReference.class);
+    when(mockRef2.getReferencingColumn()).thenReturn("fk_col2");
+    when(mockRef2.getReferencedColumn()).thenReturn("pk_col2");
+    when(mockFk.getColumnReferences()).thenReturn(Arrays.asList(mockRef1, mockRef2));
+    when(mockConstraints.getForeignKeys()).thenReturn(Collections.singletonList(mockFk));
+
+    mockTableWithConstraints(fkTableId, mockConstraints);
+
+    try (ResultSet rs =
+        dbMetadata.getCrossReference(
+            "test-project", "dataset_p", "pk_table", "test-project", "dataset_p", "fk_table")) {
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("PKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("PKTABLE_SCHEM"));
+      assertEquals("pk_table", rs.getString("PKTABLE_NAME"));
+      assertEquals("pk_col1", rs.getString("PKCOLUMN_NAME"));
+      assertEquals("test-project", rs.getString("FKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("FKTABLE_SCHEM"));
+      assertEquals("fk_table", rs.getString("FKTABLE_NAME"));
+      assertEquals("fk_col1", rs.getString("FKCOLUMN_NAME"));
+      assertEquals(1, rs.getInt("KEY_SEQ"));
+      assertEquals("fk_name", rs.getString("FK_NAME"));
+      assertTrue(rs.next());
+      assertEquals("test-project", rs.getString("PKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("PKTABLE_SCHEM"));
+      assertEquals("pk_table", rs.getString("PKTABLE_NAME"));
+      assertEquals("pk_col2", rs.getString("PKCOLUMN_NAME"));
+      assertEquals("test-project", rs.getString("FKTABLE_CAT"));
+      assertEquals("dataset_p", rs.getString("FKTABLE_SCHEM"));
+      assertEquals("fk_table", rs.getString("FKTABLE_NAME"));
+      assertEquals("fk_col2", rs.getString("FKCOLUMN_NAME"));
+      assertEquals(2, rs.getInt("KEY_SEQ"));
+      assertEquals("fk_name", rs.getString("FK_NAME"));
+      assertFalse(rs.next());
+    }
+  }
+
+  @Test
+  public void testGetCrossReference_noKeys() throws SQLException {
+    TableId fkTableId = TableId.of("test-project", "dataset_p", "fk_table");
+
+    mockTableWithConstraints(fkTableId, null);
+
+    try (ResultSet rs =
+        dbMetadata.getCrossReference(
+            "test-project", "dataset_p", "pk_table", "test-project", "dataset_p", "fk_table")) {
+      assertFalse(rs.next());
+    }
   }
 }