From 5149217c007cc9d9e3168e0bc99b502ce9fb918c Mon Sep 17 00:00:00 2001 From: "C.J. Collier" Date: Fri, 25 Jul 2025 15:10:30 -0700 Subject: [PATCH] feat(connector): Add warning log for MV read permission fallback When reading a Materialized View without 'bigquery.tables.getData' permission, the connector silently falls back to re-executing the view's query. This is opaque to the user and causes unexpected costs and poor performance, which are difficult to diagnose. This change introduces a try-catch block to detect a permission-denied error (HTTP 403) when attempting a direct read of a Materialized View. If this specific error is caught, a detailed WARN message is logged, explaining the cause of the fallback and instructing the user how to resolve it by granting the 'roles/bigquery.dataViewer' role. A new integration test, MaterializedViewReadIT, has been added to verify this behavior by impersonating a service account with insufficient permissions and asserting that the warning is logged. This significantly improves the supportability and user experience of the connector by making the fallback behavior transparent and empowering users to self-resolve the underlying permissions issue. Related to: https://issuetracker.google.com/296281345 Related to: https://github.com/dataform-co/dataform/issues/1640 Resolves #1401 --- .gitignore | 5 + bigquery-connector-common/pom.xml | 11 ++ .../connector/common/ReadSessionCreator.java | 44 +++++- .../spark-3.5-bigquery/pom.xml | 7 + .../integration/MaterializedViewReadIT.java | 139 ++++++++++++++++++ 5 files changed, 205 insertions(+), 1 deletion(-) create mode 100644 spark-bigquery-dsv2/spark-3.5-bigquery/src/test/java/com/google/cloud/spark/bigquery/integration/MaterializedViewReadIT.java diff --git a/.gitignore b/.gitignore index 9e3fd25aca..bb4ba549bf 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,8 @@ AdAdHocITSuite.scala # toolchains file toolchains.xml + +# Emacs +*~ +\#*\# +.\#* \ No newline at end of file diff --git a/bigquery-connector-common/pom.xml b/bigquery-connector-common/pom.xml index 7a3ac31bc5..5369f8dd60 100644 --- a/bigquery-connector-common/pom.xml +++ b/bigquery-connector-common/pom.xml @@ -153,6 +153,17 @@ + + org.apache.maven.plugins + maven-jar-plugin + + + + test-jar + + + + diff --git a/bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/ReadSessionCreator.java b/bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/ReadSessionCreator.java index 4265b27a42..8f5034e51d 100644 --- a/bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/ReadSessionCreator.java +++ b/bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/ReadSessionCreator.java @@ -18,6 +18,7 @@ import static com.google.cloud.bigquery.connector.common.BigQueryErrorCode.UNSUPPORTED; import static java.lang.String.format; +import com.google.cloud.bigquery.BigQueryException; import com.google.cloud.bigquery.TableDefinition; import com.google.cloud.bigquery.TableId; import com.google.cloud.bigquery.TableInfo; @@ -47,6 +48,7 @@ public class ReadSessionCreator { public static final int DEFAULT_MAX_PARALLELISM = 20_000; public static final int MINIMAL_PARALLELISM = 1; public static final int DEFAULT_MIN_PARALLELISM_FACTOR = 3; + private static final String ACCESS_DENIED_REASON = "accessDenied"; private static final Logger log = LoggerFactory.getLogger(ReadSessionCreator.class); private static boolean initialized = false; @@ -92,8 +94,48 @@ public ReadSessionResponse create( TableId table, ImmutableList selectedFields, Optional filter) { Instant sessionPrepStartTime = Instant.now(); TableInfo tableDetails = bigQueryClient.getTable(table); + + if (tableDetails.getDefinition().getType() == TableDefinition.Type.MATERIALIZED_VIEW) { + try { + // Attempt to read the Materialized View directly. This will fail if the required + // permissions are not present. + return createReadSession( + tableDetails, tableDetails, selectedFields, filter, sessionPrepStartTime); + } catch (BigQueryException e) { + if (!isPermissionDeniedError(e)) { + // Not a permission error, so re-throw it. + throw e; + } + // This is a permission error. Log a warning and fall back to materializing the view. + log.warn( + "Failed to initiate a direct read from Materialized View '{}' due to a permission" + + " error. The service account likely lacks 'bigquery.tables.getData'" + + " permission. Falling back to re-executing the view's underlying query. This" + + " will incur additional BigQuery costs and impact performance. For optimal" + + " performance, grant the 'roles/bigquery.dataViewer' role to the principal at" + + " the dataset or table level.", + tableDetails.getTableId().toString()); + // Execution will now fall through to the getActualTable() call below. + } + } + TableInfo actualTable = getActualTable(tableDetails, selectedFields, filter); + return createReadSession( + actualTable, tableDetails, selectedFields, filter, sessionPrepStartTime); + } + + private boolean isPermissionDeniedError(BigQueryException e) { + return e.getCode() == java.net.HttpURLConnection.HTTP_FORBIDDEN + && e.getError() != null + && ACCESS_DENIED_REASON.equals(e.getError().getReason()); + } + private ReadSessionResponse createReadSession( + TableInfo actualTable, + TableInfo tableDetails, + ImmutableList selectedFields, + Optional filter, + Instant sessionPrepStartTime) { BigQueryReadClient bigQueryReadClient = bigQueryReadClientFactory.getBigQueryReadClient(); log.info( "|creation a read session for table {}, parameters: " @@ -203,7 +245,7 @@ public ReadSessionResponse create( if (config.isReadSessionCachingEnabled() && getReadSessionCache().asMap().containsKey(createReadSessionRequest)) { ReadSession readSession = getReadSessionCache().asMap().get(createReadSessionRequest); - log.info("Reusing read session: {}, for table: {}", readSession.getName(), table); + log.info("Reusing read session: {}, for table: {}", readSession.getName(), actualTable); return new ReadSessionResponse(readSession, actualTable); } ReadSession readSession = bigQueryReadClient.createReadSession(createReadSessionRequest); diff --git a/spark-bigquery-dsv2/spark-3.5-bigquery/pom.xml b/spark-bigquery-dsv2/spark-3.5-bigquery/pom.xml index 6e2305769f..54a4cd7528 100644 --- a/spark-bigquery-dsv2/spark-3.5-bigquery/pom.xml +++ b/spark-bigquery-dsv2/spark-3.5-bigquery/pom.xml @@ -38,5 +38,12 @@ ${spark.version} test + + com.google.cloud.spark + bigquery-connector-common + ${project.version} + test-jar + test + diff --git a/spark-bigquery-dsv2/spark-3.5-bigquery/src/test/java/com/google/cloud/spark/bigquery/integration/MaterializedViewReadIT.java b/spark-bigquery-dsv2/spark-3.5-bigquery/src/test/java/com/google/cloud/spark/bigquery/integration/MaterializedViewReadIT.java new file mode 100644 index 0000000000..fefae91134 --- /dev/null +++ b/spark-bigquery-dsv2/spark-3.5-bigquery/src/test/java/com/google/cloud/spark/bigquery/integration/MaterializedViewReadIT.java @@ -0,0 +1,139 @@ +/* + * Copyright 2025 Google LLC and Contributors + * + * 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 + * + * 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 com.google.cloud.spark.bigquery.integration; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.cloud.bigquery.JobInfo; +import com.google.cloud.bigquery.QueryJobConfiguration; +import com.google.cloud.bigquery.TableId; +import com.google.cloud.spark.bigquery.ReadSessionCreator; +import com.google.cloud.spark.bigquery.connector.common.integration.IntegrationTestUtils; +import com.google.cloud.spark.bigquery.connector.common.integration.SparkBigQueryIntegrationTestBase; +import java.io.StringWriter; +import java.util.Arrays; +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.apache.log4j.SimpleLayout; +import org.apache.log4j.WriterAppender; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class MaterializedViewReadIT extends SparkBigQueryIntegrationTestBase { + + private final Logger logger = Logger.getLogger(ReadSessionCreator.class); + private WriterAppender appender; + private StringWriter stringWriter; + private String testDataset; + private String mvName; + private TableId sourceTableId; + private String testServiceAccount; + + @Before + public void setUp() throws InterruptedException { + // Set up a custom log appender to capture logs + stringWriter = new StringWriter(); + appender = new WriterAppender(new SimpleLayout(), stringWriter); + appender.setThreshold(Level.WARN); + logger.addAppender(appender); + + // Create a dedicated service account for this test + String projectId = System.getenv("GOOGLE_CLOUD_PROJECT"); + assertThat(projectId).isNotNull(); + testServiceAccount = + String.format( + "mv-test-%s@%s.iam.gserviceaccount.com", + UUID.randomUUID().toString().substring(0, 8), projectId); + IntegrationTestUtils.createServiceAccount(testServiceAccount); + + // Create a temporary dataset and grant the new SA the BQ User role + // This provides permissions to run jobs (for the fallback) but not to read table data directly + testDataset = "mv_read_it_" + System.nanoTime(); + IntegrationTestUtils.createDataset(testDataset); + IntegrationTestUtils.grantServiceAccountRole( + testServiceAccount, "roles/bigquery.user", testDataset); + IntegrationTestUtils.grantServiceAccountRole( + testServiceAccount, "roles/bigquery.metadataViewer", testDataset); + + // Create a source table + sourceTableId = TableId.of(testDataset, "source_table_" + System.nanoTime()); + IntegrationTestUtils.createTable(sourceTableId, "name:string, value:integer", "name,value"); + + // Create a Materialized View + mvName = "test_mv_" + System.nanoTime(); + String createMvSql = + String.format( + "CREATE MATERIALIZED VIEW `%s.%s` AS SELECT name, value FROM `%s.%s`", + testDataset, mvName, testDataset, sourceTableId.getTable()); + QueryJobConfiguration createMvJob = QueryJobConfiguration.newBuilder(createMvSql).build(); + bigquery.create(JobInfo.of(createMvJob)).waitFor(); + } + + @After + public void tearDown() { + logger.removeAppender(appender); + try { + if (testDataset != null) { + IntegrationTestUtils.deleteDatasetAndTables(testDataset); + } + } finally { + if (testServiceAccount != null) { + IntegrationTestUtils.deleteServiceAccount(testServiceAccount); + } + } + } + + @Test + public void testReadMaterializedView_lackingPermission_logsWarningAndFallsBack() { + // This test confirms that when reading a Materialized View with a service account + // that lacks `bigquery.tables.getData` permission, the connector: + // 1. Logs a specific WARN message indicating the permission issue and the fallback. + // 2. Successfully reads the data by falling back to materializing the view's query. + + // Arrange: Use the dedicated service account with insufficient permissions for a direct read. + String mvToRead = testDataset + "." + mvName; + + // Act: Read the materialized view, impersonating the test service account + Dataset df = + spark + .read() + .format("bigquery") + .option("viewsEnabled", "true") // Required to read any kind of view + .option("impersonationServiceAccount", testServiceAccount) + .load(mvToRead); + + List result = df.collectAsList(); + + // Assert + // 1. Assert that the read was successful via fallback + assertThat(result).hasSize(2); + List names = result.stream().map(row -> row.getString(0)).collect(Collectors.toList()); + assertThat(names).containsExactlyElementsIn(Arrays.asList("name1", "name2")); + + // 2. Assert that the specific warning was logged + String logOutput = stringWriter.toString(); + assertThat(logOutput).contains("Failed to initiate a direct read from Materialized View"); + assertThat(logOutput) + .contains("The service account likely lacks 'bigquery.tables.getData' permission"); + assertThat(logOutput).contains("Falling back to re-executing the view's underlying query"); + } +}