Skip to content

Commit 8f6dcee

Browse files
BoqianShiBoqianShi
committed
Add more logging for cloud run debugging
1 parent 4ce3667 commit 8f6dcee

File tree

2 files changed

+162
-101
lines changed

2 files changed

+162
-101
lines changed

‎spark-bigquery-connector-common/src/test/java/com/google/cloud/spark/bigquery/integration/CatalogIntegrationTestBase.java‎

Lines changed: 103 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -15,16 +15,9 @@
1515
*/
1616
package com.google.cloud.spark.bigquery.integration;
1717

18-
import static com.google.common.truth.Truth.assertThat;
19-
20-
import com.google.cloud.bigquery.BigQuery;
21-
import com.google.cloud.bigquery.Dataset;
22-
import com.google.cloud.bigquery.DatasetId;
23-
import com.google.cloud.bigquery.QueryJobConfiguration;
24-
import com.google.cloud.bigquery.Table;
25-
import com.google.cloud.bigquery.TableId;
2618
import java.util.List;
2719
import java.util.stream.Collectors;
20+
2821
import org.apache.spark.sql.Row;
2922
import org.apache.spark.sql.RowFactory;
3023
import org.apache.spark.sql.SparkSession;
@@ -36,6 +29,14 @@
3629
import org.junit.Ignore;
3730
import org.junit.Test;
3831

32+
import com.google.cloud.bigquery.BigQuery;
33+
import com.google.cloud.bigquery.Dataset;
34+
import com.google.cloud.bigquery.DatasetId;
35+
import com.google.cloud.bigquery.QueryJobConfiguration;
36+
import com.google.cloud.bigquery.Table;
37+
import com.google.cloud.bigquery.TableId;
38+
import static com.google.common.truth.Truth.assertThat;
39+
3940
public class CatalogIntegrationTestBase {
4041

4142
public static final String DEFAULT_NAMESPACE = "default";
@@ -45,7 +46,6 @@ public class CatalogIntegrationTestBase {
4546

4647
protected static SparkSession spark;
4748
private String testTable;
48-
// 2. Initialize the SparkSession ONCE before all tests
4949
@BeforeClass
5050
public static void setupSparkSession() {
5151
spark =
@@ -60,8 +60,6 @@ public static void setupSparkSession() {
6060
.getOrCreate();
6161
}
6262

63-
// 4. Stop the SparkSession ONCE after all tests are done
64-
// This fixes the local IllegalStateException (race condition)
6563
@AfterClass
6664
public static void teardownSparkSession() {
6765
if (spark != null) {
@@ -253,66 +251,113 @@ public void testDropDatabase() {
253251

254252
@Test
255253
public void testCatalogInitializationWithProject() {
256-
spark
257-
.conf()
258-
.set("spark.sql.catalog.public_catalog", "com.google.cloud.spark.bigquery.BigQueryCatalog");
259-
spark.conf().set("spark.sql.catalog.public_catalog.project", "bigquery-public-data");
260-
261-
List<Row> rows = spark.sql("SHOW DATABASES IN public_catalog").collectAsList();
262-
List<String> databaseNames =
263-
rows.stream().map(row -> row.getString(0)).collect(Collectors.toList());
264-
assertThat(databaseNames).contains("samples");
265-
266-
List<Row> data =
267-
spark.sql("SELECT * FROM public_catalog.samples.shakespeare LIMIT 10").collectAsList();
268-
assertThat(data).hasSize(10);
254+
try {
255+
spark
256+
.conf()
257+
.set("spark.sql.catalog.public_catalog", "com.google.cloud.spark.bigquery.BigQueryCatalog");
258+
spark.conf().set("spark.sql.catalog.public_catalog.project", "bigquery-public-data");
259+
260+
// Add a small delay to ensure catalog is fully initialized
261+
Thread.sleep(1000);
262+
263+
List<Row> rows = spark.sql("SHOW DATABASES IN public_catalog").collectAsList();
264+
List<String> databaseNames =
265+
rows.stream().map(row -> row.getString(0)).collect(Collectors.toList());
266+
assertThat(databaseNames).contains("samples");
267+
268+
List<Row> data =
269+
spark.sql("SELECT * FROM public_catalog.samples.shakespeare LIMIT 10").collectAsList();
270+
assertThat(data).hasSize(10);
271+
} catch (Exception e) {
272+
// Log the full stack trace to help debug cloud build failures
273+
e.printStackTrace();
274+
throw new RuntimeException("Test failed with detailed error", e);
275+
} finally {
276+
// Clean up catalog configuration to avoid interference with other tests
277+
try {
278+
spark.conf().unset("spark.sql.catalog.public_catalog");
279+
spark.conf().unset("spark.sql.catalog.public_catalog.project");
280+
} catch (Exception ignored) {
281+
}
282+
}
269283
}
270284

271285
@Test
272286
public void testCreateCatalogWithLocation() throws Exception {
273287
String database = String.format("create_db_with_location_%s", System.nanoTime());
274288
DatasetId datasetId = DatasetId.of(database);
275-
spark
276-
.conf()
277-
.set(
278-
"spark.sql.catalog.test_location_catalog",
279-
"com.google.cloud.spark.bigquery.BigQueryCatalog");
280-
spark.conf().set("spark.sql.catalog.test_location_catalog.bigquery_location", "EU");
281-
spark.sql("CREATE DATABASE test_location_catalog." + database);
282-
Dataset dataset = bigquery.getDataset(datasetId);
283-
assertThat(dataset).isNotNull();
284-
assertThat(dataset.getLocation()).isEqualTo("EU");
285-
bigquery.delete(datasetId, BigQuery.DatasetDeleteOption.deleteContents());
289+
try {
290+
spark
291+
.conf()
292+
.set(
293+
"spark.sql.catalog.test_location_catalog",
294+
"com.google.cloud.spark.bigquery.BigQueryCatalog");
295+
spark.conf().set("spark.sql.catalog.test_location_catalog.bigquery_location", "EU");
296+
297+
// Add delay for catalog initialization
298+
Thread.sleep(1000);
299+
300+
spark.sql("CREATE DATABASE test_location_catalog." + database);
301+
Dataset dataset = bigquery.getDataset(datasetId);
302+
assertThat(dataset).isNotNull();
303+
assertThat(dataset.getLocation()).isEqualTo("EU");
304+
} finally {
305+
bigquery.delete(datasetId, BigQuery.DatasetDeleteOption.deleteContents());
306+
// Clean up catalog configuration
307+
try {
308+
spark.conf().unset("spark.sql.catalog.test_location_catalog");
309+
spark.conf().unset("spark.sql.catalog.test_location_catalog.bigquery_location");
310+
} catch (Exception ignored) {
311+
}
312+
}
286313
}
287314

288315
@Test
289316
public void testCreateTableAsSelectWithProjectAndLocation() {
290317
String database = String.format("ctas_db_with_location_%s", System.nanoTime());
291318
String newTable = "ctas_table_from_public";
292319
DatasetId datasetId = DatasetId.of(database);
293-
spark
294-
.conf()
295-
.set("spark.sql.catalog.public_catalog", "com.google.cloud.spark.bigquery.BigQueryCatalog");
296-
spark.conf().set("spark.sql.catalog.public_catalog.projectId", "bigquery-public-data");
297-
spark
298-
.conf()
299-
.set(
300-
"spark.sql.catalog.test_catalog_as_select",
301-
"com.google.cloud.spark.bigquery.BigQueryCatalog");
302-
spark.conf().set("spark.sql.catalog.test_catalog_as_select.bigquery_location", "EU");
303-
spark.sql("CREATE DATABASE test_catalog_as_select." + database);
304-
spark.sql(
305-
"CREATE TABLE test_catalog_as_select."
306-
+ database
307-
+ "."
308-
+ newTable
309-
+ " AS SELECT * FROM public_catalog.samples.shakespeare LIMIT 10");
310-
Dataset dataset = bigquery.getDataset(datasetId);
311-
assertThat(dataset).isNotNull();
312-
assertThat(dataset.getLocation()).isEqualTo("EU");
313-
Table table = bigquery.getTable(TableId.of(datasetId.getDataset(), newTable));
314-
assertThat(table).isNotNull();
315-
bigquery.delete(datasetId, BigQuery.DatasetDeleteOption.deleteContents());
320+
try {
321+
spark
322+
.conf()
323+
.set("spark.sql.catalog.public_catalog", "com.google.cloud.spark.bigquery.BigQueryCatalog");
324+
spark.conf().set("spark.sql.catalog.public_catalog.projectId", "bigquery-public-data");
325+
spark
326+
.conf()
327+
.set(
328+
"spark.sql.catalog.test_catalog_as_select",
329+
"com.google.cloud.spark.bigquery.BigQueryCatalog");
330+
spark.conf().set("spark.sql.catalog.test_catalog_as_select.bigquery_location", "EU");
331+
332+
// Add delay for catalog initialization
333+
Thread.sleep(1000);
334+
335+
spark.sql("CREATE DATABASE test_catalog_as_select." + database);
336+
spark.sql(
337+
"CREATE TABLE test_catalog_as_select."
338+
+ database
339+
+ "."
340+
+ newTable
341+
+ " AS SELECT * FROM public_catalog.samples.shakespeare LIMIT 10");
342+
Dataset dataset = bigquery.getDataset(datasetId);
343+
assertThat(dataset).isNotNull();
344+
assertThat(dataset.getLocation()).isEqualTo("EU");
345+
Table table = bigquery.getTable(TableId.of(datasetId.getDataset(), newTable));
346+
assertThat(table).isNotNull();
347+
} catch (Exception e) {
348+
e.printStackTrace();
349+
throw new RuntimeException("Test failed with detailed error", e);
350+
} finally {
351+
bigquery.delete(datasetId, BigQuery.DatasetDeleteOption.deleteContents());
352+
// Clean up catalog configurations
353+
try {
354+
spark.conf().unset("spark.sql.catalog.public_catalog");
355+
spark.conf().unset("spark.sql.catalog.public_catalog.projectId");
356+
spark.conf().unset("spark.sql.catalog.test_catalog_as_select");
357+
spark.conf().unset("spark.sql.catalog.test_catalog_as_select.bigquery_location");
358+
} catch (Exception ignored) {
359+
}
360+
}
316361
}
317362

318363
private static SparkSession createSparkSession() {

‎spark-bigquery-dsv2/spark-3.5-bigquery-lib/src/main/java/com/google/cloud/spark/bigquery/BigQueryCatalog.java‎

Lines changed: 59 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -15,35 +15,14 @@
1515
*/
1616
package com.google.cloud.spark.bigquery;
1717

18-
import com.google.api.client.googleapis.json.GoogleJsonResponseException;
19-
import com.google.cloud.bigquery.BigQueryException;
20-
import com.google.cloud.bigquery.Dataset;
21-
import com.google.cloud.bigquery.DatasetId;
22-
import com.google.cloud.bigquery.DatasetInfo;
23-
import com.google.cloud.bigquery.Schema;
24-
import com.google.cloud.bigquery.TableDefinition;
25-
import com.google.cloud.bigquery.TableId;
26-
import com.google.cloud.bigquery.TableInfo;
27-
import com.google.cloud.bigquery.connector.common.BigQueryClient;
28-
import com.google.cloud.bigquery.connector.common.BigQueryConnectorException;
29-
import com.google.cloud.spark.bigquery.v2.BigQueryIdentifier;
30-
import com.google.cloud.spark.bigquery.v2.Spark35BigQueryTable;
31-
import com.google.cloud.spark.bigquery.v2.Spark3Util;
32-
import com.google.common.base.Preconditions;
33-
import com.google.common.cache.Cache;
34-
import com.google.common.cache.CacheBuilder;
35-
import com.google.common.cache.CacheLoader;
36-
import com.google.common.cache.LoadingCache;
37-
import com.google.common.collect.ImmutableMap;
38-
import com.google.common.collect.Streams;
39-
import com.google.inject.Injector;
4018
import java.util.Arrays;
4119
import java.util.Map;
4220
import java.util.Optional;
4321
import java.util.ServiceLoader;
4422
import java.util.concurrent.ExecutionException;
4523
import java.util.concurrent.TimeUnit;
4624
import java.util.stream.StreamSupport;
25+
4726
import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException;
4827
import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
4928
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
@@ -63,6 +42,29 @@
6342
import org.slf4j.Logger;
6443
import org.slf4j.LoggerFactory;
6544

45+
import com.google.api.client.googleapis.json.GoogleJsonResponseException;
46+
import com.google.cloud.bigquery.BigQueryException;
47+
import com.google.cloud.bigquery.Dataset;
48+
import com.google.cloud.bigquery.DatasetId;
49+
import com.google.cloud.bigquery.DatasetInfo;
50+
import com.google.cloud.bigquery.Schema;
51+
import com.google.cloud.bigquery.TableDefinition;
52+
import com.google.cloud.bigquery.TableId;
53+
import com.google.cloud.bigquery.TableInfo;
54+
import com.google.cloud.bigquery.connector.common.BigQueryClient;
55+
import com.google.cloud.bigquery.connector.common.BigQueryConnectorException;
56+
import com.google.cloud.spark.bigquery.v2.BigQueryIdentifier;
57+
import com.google.cloud.spark.bigquery.v2.Spark35BigQueryTable;
58+
import com.google.cloud.spark.bigquery.v2.Spark3Util;
59+
import com.google.common.base.Preconditions;
60+
import com.google.common.cache.Cache;
61+
import com.google.common.cache.CacheBuilder;
62+
import com.google.common.cache.CacheLoader;
63+
import com.google.common.cache.LoadingCache;
64+
import com.google.common.collect.ImmutableMap;
65+
import com.google.common.collect.Streams;
66+
import com.google.inject.Injector;
67+
6668
public class BigQueryCatalog implements TableCatalog, SupportsNamespaces {
6769

6870
private static final Logger logger = LoggerFactory.getLogger(BigQueryCatalog.class);
@@ -81,23 +83,29 @@ public class BigQueryCatalog implements TableCatalog, SupportsNamespaces {
8183

8284
@Override
8385
public void initialize(String name, CaseInsensitiveStringMap caseInsensitiveStringMap) {
84-
logger.info("Initializing BigQuery table catalog [{}])", name);
85-
Injector injector =
86-
new InjectorBuilder()
87-
.withOptions(caseInsensitiveStringMap.asCaseSensitiveMap())
88-
.withTableIsMandatory(false)
89-
.build();
90-
tableProvider =
91-
StreamSupport.stream(ServiceLoader.load(DataSourceRegister.class).spliterator(), false)
92-
.filter(candidate -> candidate.shortName().equals("bigquery"))
93-
.map(candidate -> (TableProvider) candidate)
94-
.findFirst()
95-
.orElseThrow(
96-
() -> new IllegalStateException("Could not find a BigQuery TableProvider"));
97-
bigQueryClient = injector.getInstance(BigQueryClient.class);
98-
schemaConverters =
99-
SchemaConverters.from(
100-
SchemaConvertersConfiguration.from(injector.getInstance(SparkBigQueryConfig.class)));
86+
logger.info("Initializing BigQuery table catalog [{}] with options: {}", name, caseInsensitiveStringMap);
87+
try {
88+
Injector injector =
89+
new InjectorBuilder()
90+
.withOptions(caseInsensitiveStringMap.asCaseSensitiveMap())
91+
.withTableIsMandatory(false)
92+
.build();
93+
tableProvider =
94+
StreamSupport.stream(ServiceLoader.load(DataSourceRegister.class).spliterator(), false)
95+
.filter(candidate -> candidate.shortName().equals("bigquery"))
96+
.map(candidate -> (TableProvider) candidate)
97+
.findFirst()
98+
.orElseThrow(
99+
() -> new IllegalStateException("Could not find a BigQuery TableProvider"));
100+
bigQueryClient = injector.getInstance(BigQueryClient.class);
101+
schemaConverters =
102+
SchemaConverters.from(
103+
SchemaConvertersConfiguration.from(injector.getInstance(SparkBigQueryConfig.class)));
104+
logger.info("BigQuery table catalog [{}] initialized successfully", name);
105+
} catch (Exception e) {
106+
logger.error("Failed to initialize BigQuery catalog [{}]", name, e);
107+
throw new BigQueryConnectorException("Failed to initialize BigQuery catalog: " + name, e);
108+
}
101109
}
102110

103111
@Override
@@ -289,10 +297,18 @@ static TableId toTableId(Identifier identifier) {
289297

290298
@Override
291299
public String[][] listNamespaces() throws NoSuchNamespaceException {
292-
return Streams.stream(bigQueryClient.listDatasets())
293-
.map(Dataset::getDatasetId)
294-
.map(this::toNamespace)
295-
.toArray(String[][]::new);
300+
if (bigQueryClient == null) {
301+
throw new IllegalStateException("BigQuery catalog not properly initialized");
302+
}
303+
try {
304+
return Streams.stream(bigQueryClient.listDatasets())
305+
.map(Dataset::getDatasetId)
306+
.map(this::toNamespace)
307+
.toArray(String[][]::new);
308+
} catch (Exception e) {
309+
logger.error("Error listing namespaces", e);
310+
throw new BigQueryConnectorException("Failed to list namespaces", e);
311+
}
296312
}
297313

298314
private String[] toNamespace(DatasetId datasetId) {

0 commit comments

Comments
 (0)