You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@zeppelin.apache.org by "jongyoul (via GitHub)" <gi...@apache.org> on 2026/03/22 05:03:08 UTC
[PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
jongyoul opened a new pull request, #5191:
URL: https://github.com/apache/zeppelin/pull/5191
### [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client
### What is this PR for?
The current BigQuery interpreter uses the legacy `google-api-services-bigquery` library, which is outdated and lacks modern features like the Storage Read API. This PR updates the interpreter to use the modern `google-cloud-bigquery` Cloud Client Library.
Key improvements:
- Updated core dependencies to `google-cloud-bigquery` (v2.38.0).
- Simplified query execution and result iteration using modern APIs.
- Enhanced authentication flow:
- Supports Application Default Credentials (ADC).
- Added a fallback mechanism to prompt for Service Account JSON via Zeppelin's GUI if ADC is unavailable.
- Improved error handling and job management (cancellation support).
### What type of PR is it?
Improvement, Refactoring
### Todos
* [x] - Update `bigquery/pom.xml` with modern Google Cloud dependencies.
* [x] - Refactor `BigQueryInterpreter.java` to use the new client library.
* [x] - Implement fallback authentication via `z.input` GUI.
* [x] - Update and verify `BigQueryInterpreterTest` against a live GCP environment.
### What is the Jira issue?
* [ZEPPELIN-6408](https://issues.apache.org/jira/browse/ZEPPELIN-6408)
### How should this be tested?
* **Automated Tests**: Updated `BigQueryInterpreterTest` to use the new client. Verified against a live GCP project using public datasets.
* **Manual Verification**:
1. Configure a BigQuery interpreter in Zeppelin.
2. Run a query. It should automatically use ADC from the environment.
3. If ADC is missing, it should display an input box asking for a Service Account JSON. Paste a valid JSON key and run again to verify successful authentication.
### Screenshots (if appropriate)
(Not applicable for this library update, but functional verification confirmed via logs)
### Questions:
* Does the license files need to update? No.
* Is there breaking changes for older versions? No, existing properties are maintained.
* Does this needs documentation? No major documentation changes required, but GUI fallback is a new UX feature.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971064724
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971007533
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
Review Comment:
BigQuery client creation uses `GoogleCredentials.getApplicationDefault()` directly. For ADC backed by a service-account JSON (or some other credential types), `createScopedRequired()` can be true and calls will fail unless you apply OAuth scopes (e.g. `https://www.googleapis.com/auth/cloud-platform` or BigQuery scope) before building the client.
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
Review Comment:
The service-account JSON fallback builds `ServiceAccountCredentials` from the pasted JSON but never scopes it. `ServiceAccountCredentials` typically requires `createScoped(...)` before use; otherwise BigQuery requests can fail with insufficient/invalid scopes. Consider applying required scopes when `credentials.createScopedRequired()` is true.
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
Review Comment:
`currentJobId` is written during execution and read/cleared from `cancel(...)` without any synchronization. Since cancellation can be triggered from another thread, this risks races/visibility issues. Consider making it `volatile` (or using an `AtomicReference<JobId>`) and updating it in a thread-safe way.
```suggestion
private volatile JobId currentJobId = null;
```
##########
bigquery/src/test/java/org/apache/zeppelin/bigquery/BigQueryInterpreterTest.java:
##########
@@ -53,7 +54,7 @@ public String getWrong() {
@BeforeAll
public static void initConstants() {
- InputStream is = ClassLoader.class.getResourceAsStream("/constants.json");
+ InputStream is = BigQueryInterpreterTest.class.getResourceAsStream("/constants.json");
constants = (new Gson()).<Constants> fromJson(new InputStreamReader(is), Constants.class);
Review Comment:
`InputStream` from `getResourceAsStream` isn't closed. Consider using try-with-resources around the stream/reader to avoid leaking file handles (even though it's test code).
```suggestion
try (InputStream is = BigQueryInterpreterTest.class.getResourceAsStream("/constants.json");
InputStreamReader reader = new InputStreamReader(is)) {
constants = (new Gson()).<Constants> fromJson(reader, Constants.class);
}
```
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
+ exceptionOnConnect = null;
+ } catch (IOException ex) {
+ return new InterpreterResult(Code.ERROR, "Failed to parse Service Account JSON: " + ex.getMessage());
}
}
- return new PageIterator(requestTemplate);
- }
-
- //Function to call bigQuery to run SQL and return results to the Interpreter for output
- private InterpreterResult executeSql(String sql) {
- int counter = 0;
- StringBuilder finalmessage = null;
- finalmessage = new StringBuilder("%table ");
- String projId = getProperty(PROJECT_ID);
- long wTime = Long.parseLong(getProperty(WAIT_TIME));
- long maxRows = Long.parseLong(getProperty(MAX_ROWS));
+ long wTime = Long.parseLong(getProperty(WAIT_TIME, "5000"));
+ long maxRows = Long.parseLong(getProperty(MAX_ROWS, "100000"));
String sqlDialect = getProperty(SQL_DIALECT, "").toLowerCase();
String region = getProperty(REGION, null);
- Boolean useLegacySql;
+
+ QueryJobConfiguration.Builder queryConfigBuilder = QueryJobConfiguration.newBuilder(sql);
+
switch (sqlDialect) {
case "standardsql":
- useLegacySql = false;
+ queryConfigBuilder.setUseLegacySql(false);
break;
case "legacysql":
- useLegacySql = true;
+ queryConfigBuilder.setUseLegacySql(true);
break;
default:
- // Enable query prefix like '#standardSQL' if specified
- useLegacySql = null;
+ // Use default (usually Standard SQL if not specified, though library default is legacy for backward compat. Better to set explicitly based on old behavior)
+ queryConfigBuilder.setUseLegacySql(null);
}
- Iterator<GetQueryResultsResponse> pages;
- try {
- pages = run(sql, projId, wTime, maxRows, useLegacySql, region);
- } catch (IOException ex) {
- LOGGER.error(ex.getMessage());
- return new InterpreterResult(Code.ERROR, ex.getMessage());
+
+ QueryJobConfiguration queryConfig = queryConfigBuilder.build();
+
+ String jobIdStr = UUID.randomUUID().toString();
+ if (StringUtils.isNotBlank(region)) {
+ currentJobId = JobId.newBuilder().setLocation(region).setJob(jobIdStr).build();
+ } else {
+ currentJobId = JobId.of(jobIdStr);
}
+
try {
- while (pages.hasNext()) {
- finalmessage.append(printRows(pages.next()));
+ LOGGER.info("Executing query: {}", sql);
+ Job queryJob = bqClient.create(JobInfo.newBuilder(queryConfig).setJobId(currentJobId).build());
+
+ // Wait for the query to complete
+ queryJob = queryJob.waitFor();
+
+ if (queryJob == null) {
+ return new InterpreterResult(Code.ERROR, "Job no longer exists");
+ } else if (queryJob.getStatus().getError() != null) {
+ return new InterpreterResult(Code.ERROR, queryJob.getStatus().getError().toString());
}
- return new InterpreterResult(Code.SUCCESS, finalmessage.toString());
- } catch (NullPointerException ex) {
- return new InterpreterResult(Code.ERROR, ex.getMessage());
- }
- }
- //Function to run the SQL on bigQuery service
- public static Iterator<GetQueryResultsResponse> run(final String queryString,
- final String projId, final long wTime, final long maxRows,
- Boolean useLegacySql, final String region)
- throws IOException {
- try {
- LOGGER.info("Use legacy sql: {}", useLegacySql);
- QueryResponse query;
- query = service
- .jobs()
- .query(
- projId,
- new QueryRequest().setTimeoutMs(wTime)
- .setUseLegacySql(useLegacySql).setQuery(queryString)
- .setMaxResults(maxRows)).execute();
- jobId = query.getJobReference().getJobId();
- projectId = query.getJobReference().getProjectId();
- GetQueryResults getRequest = service.jobs().getQueryResults(
- projectId,
- jobId);
- if (StringUtils.isNotBlank(region)) {
- getRequest = getRequest.setLocation(region);
+ TableResult result = queryJob.getQueryResults();
+
+ StringBuilder msg = new StringBuilder("%table ");
+
+ // Get Schema
+ List<String> schemaNames = new ArrayList<>();
+ for (Field field : result.getSchema().getFields()) {
+ schemaNames.add(field.getName());
+ }
+ msg.append(StringUtils.join(schemaNames, TAB)).append(NEWLINE);
+
+ // Get Data
+ long count = 0;
+ for (FieldValueList row : result.iterateAll()) {
+ if (count >= maxRows) {
+ break;
Review Comment:
Indentation inside the `if (count >= maxRows)` block doesn't match the surrounding 2-space indentation and may violate the repo's checkstyle indentation rules. Please reformat to consistent 2-space offsets.
```suggestion
break;
```
##########
bigquery/README.md:
##########
@@ -1,23 +1,36 @@
# Overview
-BigQuery interpreter for Apache Zeppelin
+BigQuery interpreter for Apache Zeppelin using the modern [google-cloud-bigquery](https://github.com/googleapis/java-bigquery) library.
-# Unit Tests
-BigQuery Unit tests are excluded as these tests depend on the BigQuery external service. This is because BigQuery does not have a local mock at this point.
+# Authentication
+The interpreter supports multiple ways to authenticate with Google Cloud:
-If you like to run these tests manually, please follow the following steps:
-* [Create a new project](https://support.google.com/cloud/answer/6251787?hl=en)
-* [Create a Google Compute Engine instance](https://cloud.google.com/compute/docs/instances/create-start-instance)
-* Copy the project ID that you created and add it to the property "projectId" in `resources/constants.json`
-* Run the command ./mvnw <options> -Dbigquery.text.exclude='' test -pl bigquery -am
+1. **Application Default Credentials (ADC)**:
+ This is the recommended way. If Zeppelin is running on GCE, GKE, or any environment where `gcloud auth application-default login` has been executed, the interpreter will automatically discover the credentials.
-# Connection
-The Interpreter opens a connection with the BigQuery Service using the supplied Google project ID and the compute environment variables.
+2. **Service Account JSON Key (Manual Fallback)**:
+ If ADC is not available, the interpreter will prompt you to input your Service Account JSON key through the Zeppelin GUI.
+ - To get a JSON key:
+ 1. Go to the [GCP Console Service Accounts page](https://console.cloud.google.com/iam-admin/serviceaccounts).
+ 2. Select your project and service account.
+ 3. Click **Keys** -> **Add Key** -> **Create new key**.
+ 4. Select **JSON** and click **Create**.
+ 5. Copy the entire content of the downloaded JSON file and paste it into the Zeppelin input box when prompted.
Review Comment:
README describes GUI pasting of Service Account JSON keys but doesn't warn about secret persistence/exposure. Please add a caution that keys should not be pasted into shared notes and point users to Zeppelin Credentials/ADC as the preferred secure approach.
```suggestion
5. Copy the entire content of the downloaded JSON file and paste it into the Zeppelin input box when prompted. Treat this JSON key as a secret.
- **Security caution:** Do not paste this key into shared notes, notebooks, version control, or any place where it might be stored or visible to others. Prefer using Application Default Credentials (ADC) or Zeppelin's secure credentials mechanisms where possible, and only use this manual JSON key approach as a fallback when more secure options are not available.
```
##########
docs/interpreter/bigquery.md:
##########
@@ -60,70 +60,42 @@ limitations under the License.
</tr>
</table>
+## Authentication
-## BigQuery API
-Zeppelin is built against BigQuery API version v2-rev265-1.21.0 - [API Javadocs](https://developers.google.com/resources/api-libraries/documentation/bigquery/v2/java/latest/)
+The BigQuery interpreter supports two primary ways to authenticate:
-## Enabling the BigQuery Interpreter
+### 1. Application Default Credentials (ADC)
-In a notebook, to enable the **BigQuery** interpreter, click the **Gear** icon and select **bigquery**.
+This is the recommended approach for server environments.
+- **Within GCP**: If Zeppelin is running on Google Compute Engine (GCE) or Google Kubernetes Engine (GKE), it will automatically use the attached service account.
+- **Outside GCP**: Set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable to the path of your service account JSON key file, or run `gcloud auth application-default login` on the server.
-### Provide Application Default Credentials
+### 2. Service Account JSON Key (GUI Fallback)
-Within Google Cloud Platform (e.g. Google App Engine, Google Compute Engine),
-built-in credentials are used by default.
+If no environment-level credentials are found, the interpreter will prompt you to input your **Service Account JSON key** directly in the notebook paragraph using an input form.
-Outside of GCP, follow the Google API authentication instructions for [Zeppelin Google Cloud Storage](https://zeppelin.apache.org/docs/latest/setup/storage/storage.html#notebook-storage-in-google-cloud-storage)
+**How to get a Service Account JSON key:**
+1. Go to the [IAM & Admin > Service Accounts](https://console.cloud.google.com/iam-admin/serviceaccounts) page in the GCP Console.
+2. Select or create a service account with `BigQuery User` and `BigQuery Data Viewer` roles.
+3. Click the **Keys** tab, then **Add Key > Create new key**.
+4. Choose **JSON** format and click **Create**.
+5. When the BigQuery interpreter prompts you in Zeppelin, copy and paste the entire content of this JSON file into the input box.
Review Comment:
Docs instruct users to paste a full Service Account JSON key into a Zeppelin input form. This is sensitive material and may be persisted with the note/paragraph state and exposed to other note viewers. Please add an explicit security warning and recommend safer alternatives (ADC, server-side credential file path, or Zeppelin Credentials).
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
Review Comment:
Prompting users to paste a full Service Account JSON key via `context.getGui().input(...)` risks credential exposure (values can be stored in the note/paragraph state and visible to other note readers). Prefer integrating with Zeppelin's encrypted credentials store (`AuthenticationInfo.getUserCredentials()` / credential injection) or using a server-side credential file path, and avoid persisting the raw JSON in GUI state.
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
Review Comment:
`service = bqClient` caches the user-provided service-account credentials at the interpreter-instance level. In shared interpreter mode this can leak one user's credentials into subsequent queries from other users. Consider caching per-user (keyed by `AuthenticationInfo.getUser()`), or not caching at all and instead using Zeppelin's per-user credential mechanism.
```suggestion
// Do not cache this client in a shared field to avoid leaking user credentials
```
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
+ exceptionOnConnect = null;
+ } catch (IOException ex) {
+ return new InterpreterResult(Code.ERROR, "Failed to parse Service Account JSON: " + ex.getMessage());
}
}
- return new PageIterator(requestTemplate);
- }
-
- //Function to call bigQuery to run SQL and return results to the Interpreter for output
- private InterpreterResult executeSql(String sql) {
- int counter = 0;
- StringBuilder finalmessage = null;
- finalmessage = new StringBuilder("%table ");
- String projId = getProperty(PROJECT_ID);
- long wTime = Long.parseLong(getProperty(WAIT_TIME));
- long maxRows = Long.parseLong(getProperty(MAX_ROWS));
+ long wTime = Long.parseLong(getProperty(WAIT_TIME, "5000"));
+ long maxRows = Long.parseLong(getProperty(MAX_ROWS, "100000"));
String sqlDialect = getProperty(SQL_DIALECT, "").toLowerCase();
String region = getProperty(REGION, null);
- Boolean useLegacySql;
+
+ QueryJobConfiguration.Builder queryConfigBuilder = QueryJobConfiguration.newBuilder(sql);
Review Comment:
`WAIT_TIME` is parsed into `wTime` but it is not applied to the job/query. As a result, queries can block indefinitely on `waitFor()`. Consider mapping this to `QueryJobConfiguration.Builder#setJobTimeoutMs(...)` and/or using a bounded wait strategy that cancels the job when the timeout elapses.
```suggestion
QueryJobConfiguration.Builder queryConfigBuilder = QueryJobConfiguration.newBuilder(sql)
.setJobTimeoutMs(wTime);
```
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -320,18 +281,16 @@ public int getProgress(InterpreterContext context) {
@Override
public void cancel(InterpreterContext context) {
LOGGER.info("Trying to Cancel current query statement.");
-
- if (service != null && jobId != null && projectId != null) {
- try {
- Bigquery.Jobs.Cancel request = service.jobs().cancel(projectId, jobId);
- JobCancelResponse response = request.execute();
- jobId = null;
- LOGGER.info("Query Execution cancelled");
- } catch (IOException ex) {
- LOGGER.error("Could not cancel the SQL execution");
+ if (service != null && currentJobId != null) {
+ boolean cancelled = service.cancel(currentJobId);
+ if (cancelled) {
+ LOGGER.info("Query Execution cancelled");
+ } else {
+ LOGGER.warn("Query Execution cancellation returned false");
}
+ currentJobId = null;
Review Comment:
`service.cancel(currentJobId)` can throw runtime exceptions (e.g., `BigQueryException`) on transport/auth failures. `cancel(...)` should catch/log exceptions so that paragraph cancellation doesn't crash the interpreter thread.
```suggestion
try {
boolean cancelled = service.cancel(currentJobId);
if (cancelled) {
LOGGER.info("Query Execution cancelled");
} else {
LOGGER.warn("Query Execution cancellation returned false");
}
} catch (RuntimeException e) {
LOGGER.warn("Failed to cancel BigQuery job {}", currentJobId, e);
} finally {
currentJobId = null;
}
```
##########
bigquery/pom.xml:
##########
@@ -74,7 +53,7 @@
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
- <version>${guava.version}</version>
+ <version>32.1.3-jre</version>
</dependency>
Review Comment:
The module now appears to have no direct `com.google.common.*` usage, but `guava` is still declared as a direct dependency with a hard-coded version. If it's only needed transitively, consider removing the direct dependency (or managing it via a shared property) to reduce jar size and avoid version drift across modules.
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
+ exceptionOnConnect = null;
+ } catch (IOException ex) {
+ return new InterpreterResult(Code.ERROR, "Failed to parse Service Account JSON: " + ex.getMessage());
}
}
- return new PageIterator(requestTemplate);
- }
-
- //Function to call bigQuery to run SQL and return results to the Interpreter for output
- private InterpreterResult executeSql(String sql) {
- int counter = 0;
- StringBuilder finalmessage = null;
- finalmessage = new StringBuilder("%table ");
- String projId = getProperty(PROJECT_ID);
- long wTime = Long.parseLong(getProperty(WAIT_TIME));
- long maxRows = Long.parseLong(getProperty(MAX_ROWS));
+ long wTime = Long.parseLong(getProperty(WAIT_TIME, "5000"));
+ long maxRows = Long.parseLong(getProperty(MAX_ROWS, "100000"));
String sqlDialect = getProperty(SQL_DIALECT, "").toLowerCase();
String region = getProperty(REGION, null);
- Boolean useLegacySql;
+
+ QueryJobConfiguration.Builder queryConfigBuilder = QueryJobConfiguration.newBuilder(sql);
+
switch (sqlDialect) {
case "standardsql":
- useLegacySql = false;
+ queryConfigBuilder.setUseLegacySql(false);
break;
case "legacysql":
- useLegacySql = true;
+ queryConfigBuilder.setUseLegacySql(true);
break;
default:
- // Enable query prefix like '#standardSQL' if specified
- useLegacySql = null;
+ // Use default (usually Standard SQL if not specified, though library default is legacy for backward compat. Better to set explicitly based on old behavior)
+ queryConfigBuilder.setUseLegacySql(null);
}
- Iterator<GetQueryResultsResponse> pages;
- try {
- pages = run(sql, projId, wTime, maxRows, useLegacySql, region);
- } catch (IOException ex) {
- LOGGER.error(ex.getMessage());
- return new InterpreterResult(Code.ERROR, ex.getMessage());
+
+ QueryJobConfiguration queryConfig = queryConfigBuilder.build();
+
+ String jobIdStr = UUID.randomUUID().toString();
+ if (StringUtils.isNotBlank(region)) {
+ currentJobId = JobId.newBuilder().setLocation(region).setJob(jobIdStr).build();
+ } else {
+ currentJobId = JobId.of(jobIdStr);
}
Review Comment:
Indentation in this block is inconsistent (extra spaces on the `currentJobId = ...` lines). The repo checkstyle enables the `Indentation` module with `basicOffset=2`, so this may trigger checkstyle warnings/failures; please align indentation to 2-space offsets.
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -320,18 +281,16 @@ public int getProgress(InterpreterContext context) {
@Override
public void cancel(InterpreterContext context) {
LOGGER.info("Trying to Cancel current query statement.");
-
- if (service != null && jobId != null && projectId != null) {
- try {
- Bigquery.Jobs.Cancel request = service.jobs().cancel(projectId, jobId);
- JobCancelResponse response = request.execute();
- jobId = null;
- LOGGER.info("Query Execution cancelled");
- } catch (IOException ex) {
- LOGGER.error("Could not cancel the SQL execution");
+ if (service != null && currentJobId != null) {
+ boolean cancelled = service.cancel(currentJobId);
+ if (cancelled) {
+ LOGGER.info("Query Execution cancelled");
+ } else {
+ LOGGER.warn("Query Execution cancellation returned false");
Review Comment:
Indentation inside the `if (cancelled)` / `else` block is inconsistent with the rest of the file and may trigger checkstyle indentation warnings. Please reformat to match the 2-space indentation used elsewhere.
```suggestion
LOGGER.info("Query Execution cancelled");
} else {
LOGGER.warn("Query Execution cancellation returned false");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971067134
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -320,18 +281,16 @@ public int getProgress(InterpreterContext context) {
@Override
public void cancel(InterpreterContext context) {
LOGGER.info("Trying to Cancel current query statement.");
-
- if (service != null && jobId != null && projectId != null) {
- try {
- Bigquery.Jobs.Cancel request = service.jobs().cancel(projectId, jobId);
- JobCancelResponse response = request.execute();
- jobId = null;
- LOGGER.info("Query Execution cancelled");
- } catch (IOException ex) {
- LOGGER.error("Could not cancel the SQL execution");
+ if (service != null && currentJobId != null) {
+ boolean cancelled = service.cancel(currentJobId);
+ if (cancelled) {
+ LOGGER.info("Query Execution cancelled");
+ } else {
+ LOGGER.warn("Query Execution cancellation returned false");
}
+ currentJobId = null;
Review Comment:
Fixed in latest commit. Thank you for the feedback!
##########
docs/interpreter/bigquery.md:
##########
@@ -60,70 +60,42 @@ limitations under the License.
</tr>
</table>
+## Authentication
-## BigQuery API
-Zeppelin is built against BigQuery API version v2-rev265-1.21.0 - [API Javadocs](https://developers.google.com/resources/api-libraries/documentation/bigquery/v2/java/latest/)
+The BigQuery interpreter supports two primary ways to authenticate:
-## Enabling the BigQuery Interpreter
+### 1. Application Default Credentials (ADC)
-In a notebook, to enable the **BigQuery** interpreter, click the **Gear** icon and select **bigquery**.
+This is the recommended approach for server environments.
+- **Within GCP**: If Zeppelin is running on Google Compute Engine (GCE) or Google Kubernetes Engine (GKE), it will automatically use the attached service account.
+- **Outside GCP**: Set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable to the path of your service account JSON key file, or run `gcloud auth application-default login` on the server.
-### Provide Application Default Credentials
+### 2. Service Account JSON Key (GUI Fallback)
-Within Google Cloud Platform (e.g. Google App Engine, Google Compute Engine),
-built-in credentials are used by default.
+If no environment-level credentials are found, the interpreter will prompt you to input your **Service Account JSON key** directly in the notebook paragraph using an input form.
-Outside of GCP, follow the Google API authentication instructions for [Zeppelin Google Cloud Storage](https://zeppelin.apache.org/docs/latest/setup/storage/storage.html#notebook-storage-in-google-cloud-storage)
+**How to get a Service Account JSON key:**
+1. Go to the [IAM & Admin > Service Accounts](https://console.cloud.google.com/iam-admin/serviceaccounts) page in the GCP Console.
+2. Select or create a service account with `BigQuery User` and `BigQuery Data Viewer` roles.
+3. Click the **Keys** tab, then **Add Key > Create new key**.
+4. Choose **JSON** format and click **Create**.
+5. When the BigQuery interpreter prompts you in Zeppelin, copy and paste the entire content of this JSON file into the input box.
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971069036
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
+ exceptionOnConnect = null;
+ } catch (IOException ex) {
+ return new InterpreterResult(Code.ERROR, "Failed to parse Service Account JSON: " + ex.getMessage());
}
}
- return new PageIterator(requestTemplate);
- }
-
- //Function to call bigQuery to run SQL and return results to the Interpreter for output
- private InterpreterResult executeSql(String sql) {
- int counter = 0;
- StringBuilder finalmessage = null;
- finalmessage = new StringBuilder("%table ");
- String projId = getProperty(PROJECT_ID);
- long wTime = Long.parseLong(getProperty(WAIT_TIME));
- long maxRows = Long.parseLong(getProperty(MAX_ROWS));
+ long wTime = Long.parseLong(getProperty(WAIT_TIME, "5000"));
+ long maxRows = Long.parseLong(getProperty(MAX_ROWS, "100000"));
String sqlDialect = getProperty(SQL_DIALECT, "").toLowerCase();
String region = getProperty(REGION, null);
- Boolean useLegacySql;
+
+ QueryJobConfiguration.Builder queryConfigBuilder = QueryJobConfiguration.newBuilder(sql);
+
switch (sqlDialect) {
case "standardsql":
- useLegacySql = false;
+ queryConfigBuilder.setUseLegacySql(false);
break;
case "legacysql":
- useLegacySql = true;
+ queryConfigBuilder.setUseLegacySql(true);
break;
default:
- // Enable query prefix like '#standardSQL' if specified
- useLegacySql = null;
+ // Use default (usually Standard SQL if not specified, though library default is legacy for backward compat. Better to set explicitly based on old behavior)
+ queryConfigBuilder.setUseLegacySql(null);
}
- Iterator<GetQueryResultsResponse> pages;
- try {
- pages = run(sql, projId, wTime, maxRows, useLegacySql, region);
- } catch (IOException ex) {
- LOGGER.error(ex.getMessage());
- return new InterpreterResult(Code.ERROR, ex.getMessage());
+
+ QueryJobConfiguration queryConfig = queryConfigBuilder.build();
+
+ String jobIdStr = UUID.randomUUID().toString();
+ if (StringUtils.isNotBlank(region)) {
+ currentJobId = JobId.newBuilder().setLocation(region).setJob(jobIdStr).build();
+ } else {
+ currentJobId = JobId.of(jobIdStr);
}
Review Comment:
Fixed in latest commit. Thank you for the feedback!
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
+ exceptionOnConnect = null;
+ } catch (IOException ex) {
+ return new InterpreterResult(Code.ERROR, "Failed to parse Service Account JSON: " + ex.getMessage());
}
}
- return new PageIterator(requestTemplate);
- }
-
- //Function to call bigQuery to run SQL and return results to the Interpreter for output
- private InterpreterResult executeSql(String sql) {
- int counter = 0;
- StringBuilder finalmessage = null;
- finalmessage = new StringBuilder("%table ");
- String projId = getProperty(PROJECT_ID);
- long wTime = Long.parseLong(getProperty(WAIT_TIME));
- long maxRows = Long.parseLong(getProperty(MAX_ROWS));
+ long wTime = Long.parseLong(getProperty(WAIT_TIME, "5000"));
+ long maxRows = Long.parseLong(getProperty(MAX_ROWS, "100000"));
String sqlDialect = getProperty(SQL_DIALECT, "").toLowerCase();
String region = getProperty(REGION, null);
- Boolean useLegacySql;
+
+ QueryJobConfiguration.Builder queryConfigBuilder = QueryJobConfiguration.newBuilder(sql);
+
switch (sqlDialect) {
case "standardsql":
- useLegacySql = false;
+ queryConfigBuilder.setUseLegacySql(false);
break;
case "legacysql":
- useLegacySql = true;
+ queryConfigBuilder.setUseLegacySql(true);
break;
default:
- // Enable query prefix like '#standardSQL' if specified
- useLegacySql = null;
+ // Use default (usually Standard SQL if not specified, though library default is legacy for backward compat. Better to set explicitly based on old behavior)
+ queryConfigBuilder.setUseLegacySql(null);
}
- Iterator<GetQueryResultsResponse> pages;
- try {
- pages = run(sql, projId, wTime, maxRows, useLegacySql, region);
- } catch (IOException ex) {
- LOGGER.error(ex.getMessage());
- return new InterpreterResult(Code.ERROR, ex.getMessage());
+
+ QueryJobConfiguration queryConfig = queryConfigBuilder.build();
+
+ String jobIdStr = UUID.randomUUID().toString();
+ if (StringUtils.isNotBlank(region)) {
+ currentJobId = JobId.newBuilder().setLocation(region).setJob(jobIdStr).build();
+ } else {
+ currentJobId = JobId.of(jobIdStr);
}
+
try {
- while (pages.hasNext()) {
- finalmessage.append(printRows(pages.next()));
+ LOGGER.info("Executing query: {}", sql);
+ Job queryJob = bqClient.create(JobInfo.newBuilder(queryConfig).setJobId(currentJobId).build());
+
+ // Wait for the query to complete
+ queryJob = queryJob.waitFor();
+
+ if (queryJob == null) {
+ return new InterpreterResult(Code.ERROR, "Job no longer exists");
+ } else if (queryJob.getStatus().getError() != null) {
+ return new InterpreterResult(Code.ERROR, queryJob.getStatus().getError().toString());
}
- return new InterpreterResult(Code.SUCCESS, finalmessage.toString());
- } catch (NullPointerException ex) {
- return new InterpreterResult(Code.ERROR, ex.getMessage());
- }
- }
- //Function to run the SQL on bigQuery service
- public static Iterator<GetQueryResultsResponse> run(final String queryString,
- final String projId, final long wTime, final long maxRows,
- Boolean useLegacySql, final String region)
- throws IOException {
- try {
- LOGGER.info("Use legacy sql: {}", useLegacySql);
- QueryResponse query;
- query = service
- .jobs()
- .query(
- projId,
- new QueryRequest().setTimeoutMs(wTime)
- .setUseLegacySql(useLegacySql).setQuery(queryString)
- .setMaxResults(maxRows)).execute();
- jobId = query.getJobReference().getJobId();
- projectId = query.getJobReference().getProjectId();
- GetQueryResults getRequest = service.jobs().getQueryResults(
- projectId,
- jobId);
- if (StringUtils.isNotBlank(region)) {
- getRequest = getRequest.setLocation(region);
+ TableResult result = queryJob.getQueryResults();
+
+ StringBuilder msg = new StringBuilder("%table ");
+
+ // Get Schema
+ List<String> schemaNames = new ArrayList<>();
+ for (Field field : result.getSchema().getFields()) {
+ schemaNames.add(field.getName());
+ }
+ msg.append(StringUtils.join(schemaNames, TAB)).append(NEWLINE);
+
+ // Get Data
+ long count = 0;
+ for (FieldValueList row : result.iterateAll()) {
+ if (count >= maxRows) {
+ break;
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971066039
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971065231
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
Review Comment:
Fixed in latest commit. Thank you for the feedback!
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971067900
##########
bigquery/README.md:
##########
@@ -1,23 +1,36 @@
# Overview
-BigQuery interpreter for Apache Zeppelin
+BigQuery interpreter for Apache Zeppelin using the modern [google-cloud-bigquery](https://github.com/googleapis/java-bigquery) library.
-# Unit Tests
-BigQuery Unit tests are excluded as these tests depend on the BigQuery external service. This is because BigQuery does not have a local mock at this point.
+# Authentication
+The interpreter supports multiple ways to authenticate with Google Cloud:
-If you like to run these tests manually, please follow the following steps:
-* [Create a new project](https://support.google.com/cloud/answer/6251787?hl=en)
-* [Create a Google Compute Engine instance](https://cloud.google.com/compute/docs/instances/create-start-instance)
-* Copy the project ID that you created and add it to the property "projectId" in `resources/constants.json`
-* Run the command ./mvnw <options> -Dbigquery.text.exclude='' test -pl bigquery -am
+1. **Application Default Credentials (ADC)**:
+ This is the recommended way. If Zeppelin is running on GCE, GKE, or any environment where `gcloud auth application-default login` has been executed, the interpreter will automatically discover the credentials.
-# Connection
-The Interpreter opens a connection with the BigQuery Service using the supplied Google project ID and the compute environment variables.
+2. **Service Account JSON Key (Manual Fallback)**:
+ If ADC is not available, the interpreter will prompt you to input your Service Account JSON key through the Zeppelin GUI.
+ - To get a JSON key:
+ 1. Go to the [GCP Console Service Accounts page](https://console.cloud.google.com/iam-admin/serviceaccounts).
+ 2. Select your project and service account.
+ 3. Click **Keys** -> **Add Key** -> **Create new key**.
+ 4. Select **JSON** and click **Create**.
+ 5. Copy the entire content of the downloaded JSON file and paste it into the Zeppelin input box when prompted.
Review Comment:
Fixed in latest commit. Thank you for the feedback!
##########
bigquery/pom.xml:
##########
@@ -74,7 +53,7 @@
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
- <version>${guava.version}</version>
+ <version>32.1.3-jre</version>
</dependency>
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971066464
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
private Exception exceptionOnConnect;
- private static final Function<CharSequence, String> sequenceToStringTransformer =
- new Function<CharSequence, String>() {
- public String apply(CharSequence seq) {
- return seq.toString();
- }
- };
+ private static final List<InterpreterCompletion> NO_COMPLETION = new ArrayList<>();
public BigQueryInterpreter(Properties property) {
super(property);
}
- //Function to return valid BigQuery Service
@Override
public void open() {
- if (service == null) {
- synchronized (serviceLock) {
- if (service == null) {
- try {
- service = createAuthorizedClient();
- exceptionOnConnect = null;
- LOGGER.info("Opened BigQuery SQL Connection");
- } catch (IOException e) {
- LOGGER.error("Cannot open connection", e);
- exceptionOnConnect = e;
- close();
- }
- }
+ LOGGER.info("Opening BigQuery SQL Connection...");
+ // Service initialization is lazy and depends on InterpreterContext in interpret()
+ // However, if we can init with ADC, we do it here.
+ try {
+ if (service == null) {
+ service = createDefaultClient();
+ exceptionOnConnect = null;
+ LOGGER.info("Opened BigQuery SQL Connection with ADC");
}
+ } catch (Exception e) {
+ LOGGER.warn("Cannot open connection with Application Default Credentials. Will try user credentials on interpret.", e);
+ exceptionOnConnect = e;
}
}
- //Function that Creates an authorized client to Google Bigquery.
- private static Bigquery createAuthorizedClient() throws IOException {
- HttpTransport transport = new NetHttpTransport();
- JsonFactory jsonFactory = new JacksonFactory();
- GoogleCredential credential = GoogleCredential.getApplicationDefault(transport, jsonFactory);
-
- if (credential.createScopedRequired()) {
- Collection<String> bigqueryScopes = BigqueryScopes.all();
- credential = credential.createScoped(bigqueryScopes);
+ private BigQuery createDefaultClient() throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(GoogleCredentials.getApplicationDefault());
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
-
- return new Bigquery.Builder(transport, jsonFactory, credential)
- .setApplicationName("Zeppelin/1.0 (GPN:Apache Zeppelin;)").build();
+
+ return builder.build().getService();
}
- //Function that generates and returns the schema and the rows as string
- public static String printRows(final GetQueryResultsResponse response) {
- StringBuilder msg = new StringBuilder();
- try {
- List<String> schemNames = new ArrayList<String>();
- for (TableFieldSchema schem: response.getSchema().getFields()) {
- schemNames.add(schem.getName());
- }
- msg.append(Joiner.on(TAB).join(schemNames));
- msg.append(NEWLINE);
- for (TableRow row : response.getRows()) {
- List<String> fieldValues = new ArrayList<String>();
- for (TableCell field : row.getF()) {
- fieldValues.add(field.getV().toString());
- }
- msg.append(Joiner.on(TAB).join(fieldValues));
- msg.append(NEWLINE);
- }
- return msg.toString();
- } catch (NullPointerException ex) {
- throw new NullPointerException("SQL Execution returned an error!");
+ private BigQuery getClientForUser(InterpreterContext context) throws IOException {
+ AuthenticationInfo authInfo = context.getAuthenticationInfo();
+
+ // Check if user has provided credentials via Zeppelin Credentials manager
+ if (authInfo != null && authInfo.getTicket() != null) {
+ // Typically we'd use something from credential manager, but let's assume JSON might be passed or we need z.input
+ // String userKey = authInfo.getTicket();
+ }
+
+ if (service != null) {
+ return service;
}
- }
- //Function to poll a job for completion. Future use
- public static Job pollJob(final Bigquery.Jobs.Get request, final long interval)
- throws IOException, InterruptedException {
- Job job = request.execute();
- while (!job.getStatus().getState().equals("DONE")) {
- System.out.println("Job is "
- + job.getStatus().getState()
- + " waiting " + interval + " milliseconds...");
- Thread.sleep(interval);
- job = request.execute();
+ if (exceptionOnConnect != null) {
+ throw new IOException("Failed to initialize BigQuery client with ADC", exceptionOnConnect);
}
- return job;
+
+ return createDefaultClient();
}
- //Function to page through the results of an arbitrary bigQuery request
- public static <T extends GenericJson> Iterator<T> getPages(
- final BigqueryRequest<T> requestTemplate) {
- class PageIterator implements Iterator<T> {
- private BigqueryRequest<T> request;
- private boolean hasNext = true;
- PageIterator(final BigqueryRequest<T> requestTemplate) {
- this.request = requestTemplate;
- }
- public boolean hasNext() {
- return hasNext;
+ private InterpreterResult executeSql(String sql, InterpreterContext context) {
+ BigQuery bqClient;
+ try {
+ bqClient = getClientForUser(context);
+ } catch (IOException e) {
+ // Fallback: Prompt user to input Service Account JSON via z.input
+ LOGGER.error("Authentication failed. Requesting service account JSON via GUI", e);
+ String saJson = (String) context.getGui().input("GCP Service Account JSON", "");
+ if (StringUtils.isBlank(saJson)) {
+ return new InterpreterResult(Code.ERROR, "%html ⚠️ <b>Authentication Required</b><br/>" +
+ "Could not find Application Default Credentials. Please input your Service Account JSON key in the form below and run again.");
}
- public T next() {
- if (!hasNext) {
- throw new NoSuchElementException();
- }
- try {
- T response = request.execute();
- if (response.containsKey("pageToken")) {
- request = request.set("pageToken", response.get("pageToken"));
- } else {
- hasNext = false;
- }
- return response;
- } catch (IOException e) {
- return null;
+ try {
+ GoogleCredentials credentials = ServiceAccountCredentials.fromStream(
+ new ByteArrayInputStream(saJson.getBytes(StandardCharsets.UTF_8)));
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder()
+ .setCredentials(credentials);
+
+ String projId = getProperty(PROJECT_ID);
+ if (StringUtils.isNotBlank(projId)) {
+ builder.setProjectId(projId);
}
- }
-
- public void remove() {
- this.next();
+
+ bqClient = builder.build().getService();
+ service = bqClient; // Cache it for this interpreter instance
+ exceptionOnConnect = null;
+ } catch (IOException ex) {
+ return new InterpreterResult(Code.ERROR, "Failed to parse Service Account JSON: " + ex.getMessage());
}
}
- return new PageIterator(requestTemplate);
- }
-
- //Function to call bigQuery to run SQL and return results to the Interpreter for output
- private InterpreterResult executeSql(String sql) {
- int counter = 0;
- StringBuilder finalmessage = null;
- finalmessage = new StringBuilder("%table ");
- String projId = getProperty(PROJECT_ID);
- long wTime = Long.parseLong(getProperty(WAIT_TIME));
- long maxRows = Long.parseLong(getProperty(MAX_ROWS));
+ long wTime = Long.parseLong(getProperty(WAIT_TIME, "5000"));
+ long maxRows = Long.parseLong(getProperty(MAX_ROWS, "100000"));
String sqlDialect = getProperty(SQL_DIALECT, "").toLowerCase();
String region = getProperty(REGION, null);
- Boolean useLegacySql;
+
+ QueryJobConfiguration.Builder queryConfigBuilder = QueryJobConfiguration.newBuilder(sql);
Review Comment:
Fixed in latest commit. Thank you for the feedback!
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -83,222 +76,190 @@ public class BigQueryInterpreter extends Interpreter {
private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryInterpreter.class);
private static final char NEWLINE = '\n';
private static final char TAB = '\t';
- private static Bigquery service = null;
- //Mutex created to create the singleton in thread-safe fashion.
- private static Object serviceLock = new Object();
+
+ private BigQuery service = null;
static final String PROJECT_ID = "zeppelin.bigquery.project_id";
static final String WAIT_TIME = "zeppelin.bigquery.wait_time";
static final String MAX_ROWS = "zeppelin.bigquery.max_no_of_rows";
static final String SQL_DIALECT = "zeppelin.bigquery.sql_dialect";
static final String REGION = "zeppelin.bigquery.region";
- private static String jobId = null;
- private static String projectId = null;
-
- private static final List NO_COMPLETION = new ArrayList<>();
+ private JobId currentJobId = null;
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971069850
##########
bigquery/src/main/java/org/apache/zeppelin/bigquery/BigQueryInterpreter.java:
##########
@@ -320,18 +281,16 @@ public int getProgress(InterpreterContext context) {
@Override
public void cancel(InterpreterContext context) {
LOGGER.info("Trying to Cancel current query statement.");
-
- if (service != null && jobId != null && projectId != null) {
- try {
- Bigquery.Jobs.Cancel request = service.jobs().cancel(projectId, jobId);
- JobCancelResponse response = request.execute();
- jobId = null;
- LOGGER.info("Query Execution cancelled");
- } catch (IOException ex) {
- LOGGER.error("Could not cancel the SQL execution");
+ if (service != null && currentJobId != null) {
+ boolean cancelled = service.cancel(currentJobId);
+ if (cancelled) {
+ LOGGER.info("Query Execution cancelled");
+ } else {
+ LOGGER.warn("Query Execution cancellation returned false");
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6408] Modernize BigQuery Interpreter with google-cloud-bigquery client [zeppelin]
Posted by "jongyoul (via GitHub)" <gi...@apache.org>.
jongyoul commented on code in PR #5191:
URL: https://github.com/apache/zeppelin/pull/5191#discussion_r2971068585
##########
bigquery/src/test/java/org/apache/zeppelin/bigquery/BigQueryInterpreterTest.java:
##########
@@ -53,7 +54,7 @@ public String getWrong() {
@BeforeAll
public static void initConstants() {
- InputStream is = ClassLoader.class.getResourceAsStream("/constants.json");
+ InputStream is = BigQueryInterpreterTest.class.getResourceAsStream("/constants.json");
constants = (new Gson()).<Constants> fromJson(new InputStreamReader(is), Constants.class);
Review Comment:
Fixed in latest commit. Thank you for the feedback!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org