You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by "kdeschle (via GitHub)" <gi...@apache.org> on 2024/04/06 11:06:48 UTC
[PR] Multi-stream JDBC changes [hive]
kdeschle opened a new pull request, #4902:
URL: https://github.com/apache/hive/pull/4902
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://cwiki.apache.org/confluence/display/Hive/HowToContribute
2. Ensure that you have created an issue on the Hive project JIRA: https://issues.apache.org/jira/projects/HIVE/summary
3. Ensure you have added or run the appropriate tests for your PR:
4. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]HIVE-XXXXX: Your PR title ...'.
5. Be sure to keep the PR description updated to reflect all changes.
6. Please write your PR title to summarize what this PR proposes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
Performance improvements to the Hive JDBC driver to allow for high performance fetching.
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
HIVE-27882: Fix missing classes in standalone jar [Michael Smith]
HIVE-27886: Configure Hadoop security for Kerberos when required [Michael Smith]
HIVE-27887: Provide reasonable defaults for ResultSetMetaData [Michael Smith]
HIVE-27873: Fix getOperationStatus() handling and optimize result fetch path [Kurt Deschler]
HIVE-27872: Support multi-stream parallel fetch in JDBC driver [Kurt Deschler]
HIVE-27874: Support datatype conversion on fetch threads [Kurt Deschler]
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
JDBC performance with large datasets was very slow, especially when connected to Impala via HS2 protocol
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description, screenshot and/or a reproducable example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Hive versions or within the unreleased branches such as master.
If no, write 'No'.
-->
-new JDBC connection parameter 'fetchThreads'
-fetchSize is enforce for prepared statements
-JDBC standalone JAR can be dynamically loaded
-custom hiveconf is no longer required to use the JDBC jar
### Is the change a dependency upgrade?
<!--
If yes, please attach a file with output from mvn dependency:tree to validate a complete upgrade of dependency.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
Ran existing JDBC regression tests
Manual testing multi-stream fetch against Impala
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-1841705149
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=BUG) [4 Bugs](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&resolved=false&types=SECURITY_HOTSPOT) [1 Security Hotspot](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=CODE_SMELL) [27 Code Smells](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=coverage&view=list) No Coverage information
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=duplicated_lines_density&view=list) No Duplication information
 The version of Java (11.0.8) you have used to run this analysis is deprecated and we will stop accepting it soon. Please update to at least Java 17.
Read more [here](https://docs.sonarsource.com/sonarcloud/appendices/scanner-environment/)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615640234
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
Review Comment:
should we `shutdown` the pool?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615625492
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -271,8 +299,18 @@ private void setSchema(List<String> colNames, List<String> colTypes,
colNames.forEach(i -> normalizedColumnNames.add(i.toLowerCase()));
}
+ private void InitEmptyIterator() throws SQLException {
+ try {
+ fetchedRows = RowSetFactory.create(new TRowSet(), protocol);
+ fetchedRowsItr = fetchedRows.iterator();
+ } catch (TException e) {
+ throw new SQLException(e);
+ }
+ }
+
@Override
public void close() throws SQLException {
+ ShutdownPool();
Review Comment:
please use `camel-case` namings
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -290,7 +328,7 @@ public void close() throws SQLException {
client = null;
stmtHandle = null;
isClosed = true;
- operationStatus = null;
+ InitEmptyIterator();
Review Comment:
`camel-case`
why do we need to init empty iterator, but not set them to null?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615623530
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -98,6 +109,7 @@ public static class Builder {
private List<String> colTypes;
private List<JdbcColumnAttributes> colAttributes;
private int fetchSize = 50;
+ private int fetchThreads = 1;
Review Comment:
should we get this from HiveConf?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884409523
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -285,6 +289,16 @@ public HiveConnection(String uri, Properties info) throws SQLException {
this(uri, info, HiveJdbcBrowserClientFactory.get());
}
+ public HiveConnection Clone() throws SQLException {
Review Comment:
switched to copy constructor
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886911626
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
Review Comment:
it doesn't, please see snippet in `convertFuncs` comment
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups (please see https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html).
````
Enum maps are represented internally as arrays. This representation is extremely compact and efficient.
````
I believe this code part does not need to be optimized at the expense of readability.
under the hood, it does the same:
````
public V get(Object key) {
return (isValidKey(key) ?
unmaskNull(vals[((Enum<?>)key).ordinal()]) : null);
}
````
updated code snipped with EnumMap & UnaryOperator:
````
static final EnumMap<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups (please see https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html).
````
Enum maps are represented internally as arrays. This representation is extremely compact and efficient.
````
I believe this code part does not need to be optimized (not even sure if there would be any perf difference) at the expense of readability.
````
static final EnumMap<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups (please see https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html).
````
Enum maps are represented internally as arrays. This representation is extremely compact and efficient.
````
I believe this code part does not need to be optimized at the expense of readability.
````
static final EnumMap<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2543141821
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902) **Quality Gate passed**
Issues
 [27 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2542227613
hi @kdeschle, thanks for addressing the review comments!
Would you be able to rebase this PR?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615636668
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
Review Comment:
nit: space
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615623530
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -98,6 +109,7 @@ public static class Builder {
private List<String> colTypes;
private List<JdbcColumnAttributes> colAttributes;
private int fetchSize = 50;
+ private int fetchThreads = 1;
Review Comment:
should we get this from HiveConf?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615573796
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -95,11 +99,23 @@ public int getScale(int column) throws SQLException {
}
public String getSchemaName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ String tableName = getTableName(column);
Review Comment:
use `HiveTableName.of(tableName)`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615546892
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -177,13 +178,16 @@ public class HiveConnection implements java.sql.Connection {
private TProtocolVersion protocol;
private final int initFetchSize;
private int defaultFetchSize;
+ private int fetchThreads;
Review Comment:
fetchThreadCount
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "aturoczy (via GitHub)" <gi...@apache.org>.
aturoczy commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1407795307
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -285,6 +289,16 @@ public HiveConnection(String uri, Properties info) throws SQLException {
this(uri, info, HiveJdbcBrowserClientFactory.get());
}
+ public HiveConnection Clone() throws SQLException {
Review Comment:
Alos the clone is small c :)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups. I believe this code part does not need to be optimized at the expense of readability.
````
static final Map<Type, UnaryOperator<Object>> convertFuncs = new HashMap<Type, UnaryOperator<Object>>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
List<UnaryOperator<Object>> colConvert = new ArrayList<>(nCols);
for (int i = 0; i < nCols; ++i) {
colConvert.add(convertFuncs.get(schema.getColumnDescriptorAt(i).getType()));
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert.get(i) != null && srcRow[i] != null) {
dstRow[i] = colConvert.get(i).apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2549045298
Sonar reported a number of newly introduced unused import, methods and vars, could we please drop them:
1. https://sonarcloud.io/project/issues?sinceLeakPeriod=true&issueStatuses=OPEN%2CCONFIRMED&pullRequest=4902&id=apache_hive&open=AZPVJuwuQje9E2BL3_-t
2. https://sonarcloud.io/project/issues?sinceLeakPeriod=true&issueStatuses=OPEN%2CCONFIRMED&pullRequest=4902&id=apache_hive&open=AZPFqXK8BQZ6BS4X3Vch
3. https://sonarcloud.io/project/issues?sinceLeakPeriod=true&issueStatuses=OPEN%2CCONFIRMED&pullRequest=4902&id=apache_hive&open=AZPFqXNlBQZ6BS4X3Vc4
4. https://sonarcloud.io/project/issues?sinceLeakPeriod=true&issueStatuses=OPEN%2CCONFIRMED&pullRequest=4902&id=apache_hive&open=AZPFqXNlBQZ6BS4X3Vc5
5. https://sonarcloud.io/project/issues?sinceLeakPeriod=true&issueStatuses=OPEN%2CCONFIRMED&pullRequest=4902&id=apache_hive&open=AZPFqXNlBQZ6BS4X3Vcv
6. https://sonarcloud.io/project/issues?sinceLeakPeriod=true&issueStatuses=OPEN%2CCONFIRMED&pullRequest=4902&id=apache_hive&open=AZPFqXNlBQZ6BS4X3Vcw
7. https://sonarcloud.io/project/issues?sinceLeakPeriod=true&issueStatuses=OPEN%2CCONFIRMED&pullRequest=4902&id=apache_hive&open=AZPFqXHnBQZ6BS4X3Vcg
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884401069
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
+ FetchResult result;
+ while ((result = resultQueue.poll()) != null) {
+ if (result.ex != null) {
+ throw new SQLException(result.ex);
}
+ }
+ }
- row = fetchedRowsItr.next();
- rowsFetched++;
- } catch (SQLException eS) {
- throw eS;
- } catch (Exception ex) {
- throw new SQLException("Error retrieving next row", ex);
+ void ShutdownPool() throws SQLException {
+ if (pool != null) {
+ poolDone = true;
+ DrainQueue();
Review Comment:
This sequence is necessary to ensure that nothing blocks and nothing is left in the queue.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884343289
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
Review Comment:
That leads to generic array creation compile error.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615536609
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -177,13 +178,16 @@ public class HiveConnection implements java.sql.Connection {
private TProtocolVersion protocol;
private final int initFetchSize;
private int defaultFetchSize;
+ private int fetchThreads;
private String initFile = null;
private String wmPool = null, wmApp = null;
private Properties clientInfo;
private Subject loggedInSubject;
private int maxRetries = 1;
private IJdbcBrowserClient browserClient;
+ public TCLIService.Iface GetClient() { return client; }
Review Comment:
naming convention is 'camel-case'
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615538836
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -285,6 +289,16 @@ public HiveConnection(String uri, Properties info) throws SQLException {
this(uri, info, HiveJdbcBrowserClientFactory.get());
}
+ public HiveConnection Clone() throws SQLException {
Review Comment:
if you don't won't to override Object#clone, create a copy constructor
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615630221
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
Review Comment:
could you please move it to the rest or class fields declaration
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ closed pull request #4902: Multi-stream JDBC changes
URL: https://github.com/apache/hive/pull/4902
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2566083935
Thanks for the insights. I was able to repro locally as well and fix the issues.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-1831129897
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=BUG) [4 Bugs](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&resolved=false&types=SECURITY_HOTSPOT) [1 Security Hotspot](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=CODE_SMELL) [27 Code Smells](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=coverage&view=list) No Coverage information
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=duplicated_lines_density&view=list) No Duplication information
 The version of Java (11.0.8) you have used to run this analysis is deprecated and we will stop accepting it soon. Please update to at least Java 17.
Read more [here](https://docs.sonarsource.com/sonarcloud/appendices/scanner-environment/)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615560435
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -293,12 +311,20 @@ protected int getNumRetries() {
@VisibleForTesting
protected HiveConnection(String uri, Properties info,
IJdbcBrowserClientFactory browserClientFactory) throws SQLException {
+ this(uri, info, browserClientFactory, true);
+ }
+
+ @VisibleForTesting
Review Comment:
why do we need this annotation, I don't see use in tests for the new constructor
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615562591
##########
jdbc/src/java/org/apache/hive/jdbc/Utils.java:
##########
@@ -166,6 +166,7 @@ public static class JdbcConnectionParams {
// Set the fetchSize
static final String FETCH_SIZE = "fetchSize";
static final String INIT_FILE = "initFile";
+ static final String FETCH_THREADS = "fetchThreads";
Review Comment:
FETCH_THREADS_COUNT = "fetchThreadCount";
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615625492
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -271,8 +299,18 @@ private void setSchema(List<String> colNames, List<String> colTypes,
colNames.forEach(i -> normalizedColumnNames.add(i.toLowerCase()));
}
+ private void InitEmptyIterator() throws SQLException {
+ try {
+ fetchedRows = RowSetFactory.create(new TRowSet(), protocol);
+ fetchedRowsItr = fetchedRows.iterator();
+ } catch (TException e) {
+ throw new SQLException(e);
+ }
+ }
+
@Override
public void close() throws SQLException {
+ ShutdownPool();
Review Comment:
please use `camel-case` namings
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2566443089
Closed HIVE-27882 as dup of HIVE-28315 and remove from PR desc since that was already fixed.
Updated the other ticket to HIVE-28018.
I'd like to leave the stack as-is now as they address different problems and can be cherrypicked separately to other branches.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2553639631
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902) **Quality Gate passed**
Issues
 [17 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups (please see https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html). I believe this code part does not need to be optimized at the expense of readability.
````
static final EnumMap<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884344776
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
Map would be less efficient than the current ordinal lookup.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884404685
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
+ FetchResult result;
+ while ((result = resultQueue.poll()) != null) {
+ if (result.ex != null) {
+ throw new SQLException(result.ex);
}
+ }
+ }
- row = fetchedRowsItr.next();
- rowsFetched++;
- } catch (SQLException eS) {
- throw eS;
- } catch (Exception ex) {
- throw new SQLException("Error retrieving next row", ex);
+ void ShutdownPool() throws SQLException {
+ if (pool != null) {
+ poolDone = true;
+ DrainQueue();
+ pool.shutdownNow();
+ try {
+ while (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
+ DrainQueue();
Review Comment:
This was part of prescribed strategies for shutting down ExecutorService.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1888486147
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
With this change there was a 1.1% decrease in performance. I can keep the change if you insist.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #4902: Multi-stream JDBC changes
URL: https://github.com/apache/hive/pull/4902
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1428820807
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -30,6 +30,8 @@
*
*/
public class HiveResultSetMetaData implements java.sql.ResultSetMetaData {
+ private static final String DOT = ".";
Review Comment:
all constants should be marked as `static final`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "aturoczy (via GitHub)" <gi...@apache.org>.
aturoczy commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-1829606665
The test failed with incorrect POM :(
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2553375470
this PR causes issues on a Jenkins, build timeouts after 12hr
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1888509077
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
@kdeschle what Map implementation have you used? `EnumMap` or `HashMap`?
`EnumMap`, as in the above code snippet, is 1-to-1 match with your solution (vals array with key.ordinal() index access)
1.1% decrease in generic performance performance or collection access?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2549978295
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902) **Quality Gate passed**
Issues
 [17 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups (please see https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html).
````
Enum maps are represented internally as arrays. This representation is extremely compact and efficient.
````
I believe this code part does not need to be optimized at the expense of readability.
under the hood, it does the same:
````
public V get(Object key) {
return (isValidKey(key) ?
unmaskNull(vals[((Enum<?>)key).ordinal()]) : null);
}
````
updated code snipped with EnumMap & UnaryOperator:
````
static final Map<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>(Type.class) {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2253671858
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.
Feel free to reach out on the dev@hive.apache.org list if the patch is in need of reviews.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2566260891
> Thanks for the insights. I was able to repro locally as well and fix the issues.
Thanks for the fix @kdeschle!
I noticed that we have 2 commits for HIVE-27872, should I squash them into 1
1. Support multi-stream parallel fetch in JDBC driver
2. Don't require HiveConf for JDBC Metadata calls
Also [HIVE-27882](https://issues.apache.org/jira/browse/HIVE-27882): Fix missing classes in standalone jar, mentioned in the PR description, is missing. Is that expected?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884392246
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
Review Comment:
Not here. This is just early shutdown for error case. Unconditional shutdown is from close()
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2545939884
In general LGTM. Would be great to include this perf improvement in the next 4.1.0 Hive release.
I've reopened 1 comment with the refactor recommendation: https://github.com/apache/hive/pull/4902#discussion_r1886898843
Also, Sonar reported a few new issues, we should try to address those.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups (please see https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html).
````
Enum maps are represented internally as arrays. This representation is extremely compact and efficient.
````
I believe this code part does not need to be optimized at the expense of readability.
under the hood, it does the same:
public V get(Object key) {
return (isValidKey(key) ?
unmaskNull(vals[((Enum<?>)key).ordinal()]) : null);
}
````
static final EnumMap<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups (please see https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html).
````
Enum maps are represented internally as arrays. This representation is extremely compact and efficient.
````
I believe this code part does not need to be optimized at the expense of readability.
under the hood, it does the same:
````
public V get(Object key) {
return (isValidKey(key) ?
unmaskNull(vals[((Enum<?>)key).ordinal()]) : null);
}
````
````
static final EnumMap<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884350281
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -360,8 +393,10 @@ protected HiveConnection(String uri, Properties info,
throw new SQLException(new IllegalArgumentException(
"Browser mode is not supported in embedded mode"));
}
- openSession();
- executeInitSql();
+ if (initSession) {
Review Comment:
I don't think it makes sense to add a function here.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2566635263
Thanks Denis!
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615563125
##########
jdbc/src/java/org/apache/hive/jdbc/JdbcColumn.java:
##########
@@ -372,6 +372,20 @@ static int columnScale(Type hiveType, JdbcColumnAttributes columnAttributes)
}
}
+ static boolean columnSigned(Type hiveType, JdbcColumnAttributes columnAttributes)
+ throws SQLException {
+ int columnType = hiveTypeToSqlType(hiveType);
+ switch(columnType) {
+ case Types.TINYINT:
Review Comment:
please format the code
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615562591
##########
jdbc/src/java/org/apache/hive/jdbc/Utils.java:
##########
@@ -166,6 +166,7 @@ public static class JdbcConnectionParams {
// Set the fetchSize
static final String FETCH_SIZE = "fetchSize";
static final String INIT_FILE = "initFile";
+ static final String FETCH_THREADS = "fetchThreads";
Review Comment:
FETCH_THREADS_COUNT = "fetchThreadCount";
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615603512
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
why not use Map?
````
static final Map<Type, UnaryOperator<?>> convertFuncs = new HashMap<Type, UnaryOperator<?>>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
....
}};
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615606095
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
Review Comment:
please use interface (code to interface, not an implementation)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615642168
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
Review Comment:
should we `shutdown` the pool in `finally` block?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #4902: Multi-stream JDBC changes
URL: https://github.com/apache/hive/pull/4902
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-1945135311
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.
Feel free to reach out on the dev@hive.apache.org list if the patch is in need of reviews.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615569568
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -43,7 +45,9 @@ public HiveResultSetMetaData(List<String> columnNames,
}
public String getCatalogName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ // Hive doesn't implement a catalog concept, see
+ // https://issues.apache.org/jira/browse/HIVE-3121.
+ return "";
Review Comment:
should be `default`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615606095
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
Review Comment:
please use interface `List` (code to interface, not an implementation)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615648645
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
+ FetchResult result;
+ while ((result = resultQueue.poll()) != null) {
+ if (result.ex != null) {
+ throw new SQLException(result.ex);
}
+ }
+ }
- row = fetchedRowsItr.next();
- rowsFetched++;
- } catch (SQLException eS) {
- throw eS;
- } catch (Exception ex) {
- throw new SQLException("Error retrieving next row", ex);
+ void ShutdownPool() throws SQLException {
+ if (pool != null) {
+ poolDone = true;
+ DrainQueue();
+ pool.shutdownNow();
+ try {
+ while (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
+ DrainQueue();
Review Comment:
why do you need to call it multiple times?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615643164
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
+ FetchResult result;
+ while ((result = resultQueue.poll()) != null) {
+ if (result.ex != null) {
+ throw new SQLException(result.ex);
}
+ }
+ }
- row = fetchedRowsItr.next();
- rowsFetched++;
- } catch (SQLException eS) {
- throw eS;
- } catch (Exception ex) {
- throw new SQLException("Error retrieving next row", ex);
+ void ShutdownPool() throws SQLException {
Review Comment:
`camel-case`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615546892
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -177,13 +178,16 @@ public class HiveConnection implements java.sql.Connection {
private TProtocolVersion protocol;
private final int initFetchSize;
private int defaultFetchSize;
+ private int fetchThreads;
Review Comment:
fetchThreadCount
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615625946
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -290,7 +328,7 @@ public void close() throws SQLException {
client = null;
stmtHandle = null;
isClosed = true;
- operationStatus = null;
+ InitEmptyIterator();
Review Comment:
`camel-case`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "MikaelSmith (via GitHub)" <gi...@apache.org>.
MikaelSmith commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1408439024
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -285,6 +289,16 @@ public HiveConnection(String uri, Properties info) throws SQLException {
this(uri, info, HiveJdbcBrowserClientFactory.get());
}
+ public HiveConnection Clone() throws SQLException {
Review Comment:
I'm not sure overriding Object#clone makes sense. I've added a comment to describe it's behavior.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "MikaelSmith (via GitHub)" <gi...@apache.org>.
MikaelSmith commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1408352082
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -285,6 +289,16 @@ public HiveConnection(String uri, Properties info) throws SQLException {
this(uri, info, HiveJdbcBrowserClientFactory.get());
}
+ public HiveConnection Clone() throws SQLException {
Review Comment:
It ensures both clients share a session ID.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "MikaelSmith (via GitHub)" <gi...@apache.org>.
MikaelSmith commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1408282244
##########
jdbc/pom.xml:
##########
@@ -302,6 +295,8 @@
<exclude>org.apache.hadoop:hadoop-annotations</exclude>
<exclude>org.apache.hadoop:hadoop-auth</exclude>
<exclude>org.apache.hbase*:*</exclude>
+ <exclude>org.apache.hive:hive-classification</exclude>
+ <exclude>org.apache.hive:hive-storage-api</exclude>
Review Comment:
These have extra trailing whitespace.
```suggestion
<exclude>org.apache.hive:hive-classification</exclude>
<exclude>org.apache.hive:hive-storage-api</exclude>
```
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1888596713
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
Nevermind. It seems it was just variation in my runs. Posted new rev with these changes and some fixes.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884515092
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -307,6 +333,12 @@ protected HiveConnection(String uri, Properties info,
sessConfMap = connParams.getSessionVars();
setupLoginTimeout();
if (isKerberosAuthMode()) {
+ // Ensure UserGroupInformation includes any authorized Kerberos principals.
+ LOG.debug("Configuring Kerberos mode");
+ Configuration config = new Configuration();
+ config.set("hadoop.security.authentication", "Kerberos");
Review Comment:
changed constant.
I don't thing it should always be kerberized.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2401037226
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.
Feel free to reach out on the dev@hive.apache.org list if the patch is in need of reviews.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1888509077
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
@kdeschle what Map implementation have you used? `EnumMap` or `HashMap`?
`EnumMap` as in the above code snippet is 1-to-1 match with your solution (vals array with key.ordinal() index access)
1.1% decrease in generic performance performance or collection access?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups. I believe this code part does not need to be optimized at the expense of readability.
````
static final Map<Type, UnaryOperator<Object>> convertFuncs = new HashMap<Type, UnaryOperator<Object>>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
List<UnaryOperator<Object>> colConvert = new ArrayList<>(nCols);
for (int i = 0; i < nCols; ++i) {
colConvert.add(convertFuncs.get(schema.getColumnDescriptorAt(i).getType()));
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert.get(i) != null && srcRow[i] != null) {
dstRow[i] = colConvert.get(i).apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1887003709
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final List<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
+ static {
+ // These mirror the conversions done in HiveBaseResultSet
+ convertFuncs[Type.BINARY_TYPE.ordinal()] =
+ o -> (o instanceof String) ? ((String)o).getBytes() : 0;
+ convertFuncs[Type.TIMESTAMP_TYPE.ordinal()] =
+ o -> Timestamp.valueOf((String)o);
+ convertFuncs[Type.TIMESTAMPLOCALTZ_TYPE.ordinal()] =
+ o -> TimestampTZUtil.parse((String)o);
+ convertFuncs[Type.DECIMAL_TYPE.ordinal()] =
+ o -> new BigDecimal((String)o);
+ convertFuncs[Type.DATE_TYPE.ordinal()] =
+ o -> Date.valueOf((String) o);
+ convertFuncs[Type. INTERVAL_YEAR_MONTH_TYPE.ordinal()] =
+ o -> HiveIntervalYearMonth.valueOf((String) o);
+ convertFuncs[Type. INTERVAL_DAY_TIME_TYPE.ordinal()] =
+ o -> HiveIntervalDayTime.valueOf((String) o);
+ }
+
+ public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
+ rows = new ArrayList<Object[]>();
+ numColumns = rowSet.numColumns();
+ Iterator<Object[]> srcIter = rowSet.iterator();
+ final int nCols = schema.getSize();
+ ConvertFunc [] colConvert = new ConvertFunc[nCols];
+ for (int i = 0; i < nCols; ++i) {
+ colConvert[i] = convertFuncs[schema.getColumnDescriptorAt(i).getType().ordinal()];
+ }
+
+ while (srcIter.hasNext()) {
+ Object [] srcRow = srcIter.next();
+ Object [] dstRow = new Object[srcRow.length];
+ for (int i = 0; i < nCols; ++i) {
+ if (colConvert[i] != null && srcRow[i] != null) {
+ dstRow[i] = colConvert[i].convert(srcRow[i]);
+ } else {
+ dstRow[i] = srcRow[i];
+ }
+ }
+ rows.add(dstRow);
+ }
+ }
+
+ @Override
+ public ConvertedResultSet addRow(Object[] fields) {
+ throw new UnsupportedOperationException("addRow");
+ }
+
+ @Override
+ public int numColumns() {
+ return numColumns;
+ }
+
+ @Override
+ public int numRows() {
+ return rows.size();
+ }
+
+ public ConvertedResultSet extractSubset(int maxRows) {
+ throw new UnsupportedOperationException("extractSubset");
+ }
+
+ @Override
+ public long getStartOffset() {
+ throw new UnsupportedOperationException("getStartOffset");
+ }
+
+ @Override
+ public void setStartOffset(long startOffset) {
+ throw new UnsupportedOperationException("setStartOffset");
+ }
+
+ @Override
+ public TRowSet toTRowSet() {
+ throw new UnsupportedOperationException("toTRowSet");
+ }
+
+ @Override
+ public Iterator<Object[]> iterator() {
+ return new Iterator<Object[]>() {
+
+ final Iterator<Object[]> iterator = rows.iterator();
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public Object[] next() {
+ return iterator.next();
+ }
+
+ @Override
+ public void remove() {
Review Comment:
no need to override, it's identical to parent method
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2554351028
Any idea what the build timeout was about?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615640234
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
Review Comment:
show we `shutdown` the pool?
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
Review Comment:
should we `shutdown` the pool?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615611526
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -360,8 +393,10 @@ protected HiveConnection(String uri, Properties info,
throw new SQLException(new IllegalArgumentException(
"Browser mode is not supported in embedded mode"));
}
- openSession();
- executeInitSql();
+ if (initSession) {
Review Comment:
could we extract this into a helper method, `initSession(bool)`?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615538836
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -285,6 +289,16 @@ public HiveConnection(String uri, Properties info) throws SQLException {
this(uri, info, HiveJdbcBrowserClientFactory.get());
}
+ public HiveConnection Clone() throws SQLException {
Review Comment:
if you don't want to override Object#clone, create a copy constructor
````
public HiveConnection(HiveConnection hiveConnection) {
this(hiveConnection.getConnectedUrl(), hiveConnection.getClientInfo(), HiveJdbcBrowserClientFactory.get(), false);
// These are set/updated when the session is established.
this.sessHandle = hiveConnection.sessHandle;
this.connParams = hiveConnection.connParams;
this.protocol = hiveConnection.protocol;
this.defaultFetchSize = hiveConnection.defaultFetchSize;
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615587219
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
Review Comment:
why not use `UnaryOperator`?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615633824
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
Review Comment:
use `Logger` instead, `LOG.error()`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615625492
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -271,8 +299,18 @@ private void setSchema(List<String> colNames, List<String> colTypes,
colNames.forEach(i -> normalizedColumnNames.add(i.toLowerCase()));
}
+ private void InitEmptyIterator() throws SQLException {
+ try {
+ fetchedRows = RowSetFactory.create(new TRowSet(), protocol);
+ fetchedRowsItr = fetchedRows.iterator();
+ } catch (TException e) {
+ throw new SQLException(e);
+ }
+ }
+
@Override
public void close() throws SQLException {
+ ShutdownPool();
Review Comment:
please use `camel-case` namings
why do we need to init empty iterator, but not set them to null?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615630221
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
Review Comment:
could you please move it to the rest or class fields declarations
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "aturoczy (via GitHub)" <gi...@apache.org>.
aturoczy commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1407590622
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -285,6 +289,16 @@ public HiveConnection(String uri, Properties info) throws SQLException {
this(uri, info, HiveJdbcBrowserClientFactory.get());
}
+ public HiveConnection Clone() throws SQLException {
Review Comment:
In which use case the Clone is needed? It is not better just to create another connection with the same connection string?
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -30,6 +30,8 @@
*
*/
public class HiveResultSetMetaData implements java.sql.ResultSetMetaData {
+ private static final String DOT = ".";
Review Comment:
It is a bit overloaded to me to create a constant for DOT that is . - I'm fine with the . :-)
cc: @ayushtkn @deniskuzZ Is this a design rule to make this static final? Even in the simplest cases?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615565008
##########
jdbc/src/java/org/apache/hive/jdbc/HiveStatement.java:
##########
@@ -73,12 +74,16 @@ public class HiveStatement implements java.sql.Statement {
private static final int DEFAULT_FETCH_SIZE =
HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_DEFAULT_FETCH_SIZE.defaultIntVal;
+// TODO: Should there be a hiveconf default?
+ private static final int DEFAULT_FETCH_THREADS = 1;
Review Comment:
i think so
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615574502
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -95,11 +99,23 @@ public int getScale(int column) throws SQLException {
}
public String getSchemaName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ String tableName = getTableName(column);
+ int index = tableName.lastIndexOf(DOT);
+ if (index >= 0) {
+ return tableName.substring(0, index);
+ }
+ // Impala usually doesn't return fully qualified column names. Return "" to avoid
+ // giving false results.
+ return "";
}
public String getTableName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ String columnName = getColumnName(column);
Review Comment:
same, use `HiveTableName.of(tableName).getTable()`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615573796
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -95,11 +99,23 @@ public int getScale(int column) throws SQLException {
}
public String getSchemaName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ String tableName = getTableName(column);
Review Comment:
use `HiveTableName.of(tableName).getDb()`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615611526
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -360,8 +393,10 @@ protected HiveConnection(String uri, Properties info,
throw new SQLException(new IllegalArgumentException(
"Browser mode is not supported in embedded mode"));
}
- openSession();
- executeInitSql();
+ if (initSession) {
Review Comment:
could we extract this into a helper method?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615633824
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
Review Comment:
use `Logger` instead
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #4902: Multi-stream JDBC changes
URL: https://github.com/apache/hive/pull/4902
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups. I believe this code part does not need to be optimized at the expense of readability.
````
static final EnumMap<Type, UnaryOperator> convertFuncs = new EnumMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2548695832
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902) **Quality Gate passed**
Issues
 [28 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886898843
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
That is the usage pattern for the HashTable - do the lookups. I believe this code part does not need to be optimized at the expense of readability.
````
static final Map<Type, UnaryOperator> convertFuncs = new HashMap<Type, UnaryOperator>() {{
put(Type.BINARY_TYPE, o -> (o instanceof String) ? ((String) o).getBytes() : 0);
put(Type.TIMESTAMP_TYPE, o -> Timestamp.valueOf((String) o));
put(Type.TIMESTAMPLOCALTZ_TYPE, o -> TimestampTZUtil.parse((String) o));
put(Type.DECIMAL_TYPE, o -> new BigDecimal((String) o));
put(Type.DATE_TYPE, o -> Date.valueOf((String) o));
put(Type.INTERVAL_YEAR_MONTH_TYPE, o -> HiveIntervalYearMonth.valueOf((String) o));
put(Type.INTERVAL_DAY_TIME_TYPE, o -> HiveIntervalDayTime.valueOf((String) o));
}};
public ConvertedResultSet(RowSet rowSet, TableSchema schema) {
rows = new ArrayList<>();
numColumns = rowSet.numColumns();
Iterator<Object[]> srcIter = rowSet.iterator();
final int nCols = schema.getSize();
UnaryOperator[] colConvert = new UnaryOperator[nCols];
for (int i = 0; i < nCols; ++i) {
colConvert[i] = convertFuncs.get(schema.getColumnDescriptorAt(i).getType());
}
while (srcIter.hasNext()) {
Object[] srcRow = srcIter.next();
Object[] dstRow = new Object[srcRow.length];
for (int i = 0; i < nCols; ++i) {
if (colConvert[i] != null && srcRow[i] != null) {
dstRow[i] = colConvert[i].apply(srcRow[i]);
} else {
dstRow[i] = srcRow[i];
}
}
rows.add(dstRow);
}
}
````
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1886908523
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final List<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
Review Comment:
`nTypes` variable can be removed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1887006053
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
Review Comment:
unused import
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2551886475
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902) **Quality Gate passed**
Issues
 [17 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "kdeschle (via GitHub)" <gi...@apache.org>.
kdeschle commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1884421643
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -290,7 +328,7 @@ public void close() throws SQLException {
client = null;
stmtHandle = null;
isClosed = true;
- operationStatus = null;
+ InitEmptyIterator();
Review Comment:
Have to restore initial state in case result is re-opened.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2566106243
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902) **Quality Gate passed**
Issues
 [17 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1888509077
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
@kdeschle what Map implementation have you used? `EnumMap` or `HashMap`?
`EnumMap`, as in the above code snippet, is 1-to-1 match with your solution (vals array with key.ordinal() index access)
1.1% decrease in generic performance or collection access?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1888509077
##########
jdbc/src/java/org/apache/hive/jdbc/ConvertedResultSet.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hive.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
+import org.apache.hadoop.hive.common.type.HiveIntervalYearMonth;
+import org.apache.hadoop.hive.common.type.TimestampTZUtil;
+
+import org.apache.hadoop.hive.serde2.thrift.Type;
+
+import org.apache.hive.service.rpc.thrift.TRowSet;
+import org.apache.hive.service.cli.RowSet;
+import org.apache.hive.service.cli.TableSchema;
+import org.apache.hive.service.cli.TypeDescriptor;
+
+/**
+ * ConvertedResultSet
+ */
+public class ConvertedResultSet implements RowSet {
+
+ private final ArrayList<Object[]> rows;
+ private final int numColumns;
+
+ private interface ConvertFunc {
+ Object convert(Object o);
+ }
+
+ static final int nTypes = Type.values().length;
+ static final ConvertFunc convertFuncs [] = new ConvertFunc [nTypes];
Review Comment:
@kdeschle what Map implementation have you used? EnumMap to HashMap?
EnumMap is 1-to-1 match with your solution (vals array with key.ordinal() index access)
1.1% decrease in generic performance performance or collection access?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615644471
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
+ FetchResult result;
+ while ((result = resultQueue.poll()) != null) {
+ if (result.ex != null) {
+ throw new SQLException(result.ex);
}
+ }
+ }
- row = fetchedRowsItr.next();
- rowsFetched++;
- } catch (SQLException eS) {
- throw eS;
- } catch (Exception ex) {
- throw new SQLException("Error retrieving next row", ex);
+ void ShutdownPool() throws SQLException {
+ if (pool != null) {
+ poolDone = true;
+ DrainQueue();
Review Comment:
do we need to call it before pool shutdown? you throw an exception in that method, so poll might not be closed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615644471
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
+ FetchResult result;
+ while ((result = resultQueue.poll()) != null) {
+ if (result.ex != null) {
+ throw new SQLException(result.ex);
}
+ }
+ }
- row = fetchedRowsItr.next();
- rowsFetched++;
- } catch (SQLException eS) {
- throw eS;
- } catch (Exception ex) {
- throw new SQLException("Error retrieving next row", ex);
+ void ShutdownPool() throws SQLException {
+ if (pool != null) {
+ poolDone = true;
+ DrainQueue();
Review Comment:
do we need to call it before pool shutdown?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615642846
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
Review Comment:
`camel-case`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615637303
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
Review Comment:
nit: space after `->`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615648645
##########
jdbc/src/java/org/apache/hive/jdbc/HiveQueryResultSet.java:
##########
@@ -307,66 +345,220 @@ private void closeOperationHandle(TOperationHandle stmtHandle) throws SQLExcepti
}
}
- /**
- * Moves the cursor down one row from its current position.
- *
- * @see java.sql.ResultSet#next()
- * @throws SQLException
- * if a database access error occurs.
- */
- public boolean next() throws SQLException {
+ ExecutorService pool = null;
+ volatile boolean hasStartRow = false;
+ int poolSize = 1;
+ int threadsStarted = 0;
+
+ private class FetchResult {
+ Exception ex;
+ RowSet fetchedRows;
+ boolean hasMoreRows;
+ long startRow;
+ int numRows;
+ }
+
+ BlockingQueue<FetchResult> resultQueue = new ArrayBlockingQueue<FetchResult>(4);
+ AtomicLong nextStartRow = new AtomicLong(1L);
+
+ volatile InterruptedException interruptException = null;
+ volatile boolean gotLastBatch = false;
+ volatile boolean poolDone = false;
+
+ private boolean nextRowBatch() throws SQLException {
if (isClosed) {
throw new SQLException("Resultset is closed");
}
- if (emptyResultSet || (maxRows > 0 && rowsFetched >= maxRows)) {
+ if ((maxRows > 0 && rowsFetched >= maxRows) || emptyResultSet || fetchDone) {
return false;
}
-
- /*
- * Poll on the operation status, till the operation is complete.
- * We need to wait only for HiveStatement to complete.
- * HiveDatabaseMetaData which also uses this ResultSet returns only after the RPC is complete.
- */
- // when isHasResultSet is set, the query transitioned from running -> complete and is not expected go back to
- // running state when fetching results (implicit state transition)
- if ((statement instanceof HiveStatement) && (operationStatus == null || !operationStatus.isHasResultSet())) {
- operationStatus = ((HiveStatement) statement).waitForOperationToComplete();
+ if (check_operation_status) {
+ TGetOperationStatusResp operationStatus =
+ ((HiveStatement) statement).waitForOperationToComplete();
+ check_operation_status = false;
}
- try {
- TFetchOrientation orientation = TFetchOrientation.FETCH_NEXT;
- if (fetchFirst) {
- // If we are asked to start from beginning, clear the current fetched resultset
- orientation = TFetchOrientation.FETCH_FIRST;
- fetchedRows = null;
- fetchedRowsItr = null;
- fetchFirst = false;
- }
- if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
+// TODO: Could support pool with maxRows by bounding results instead
+ if (rowsFetched < fetchSize || poolSize == 0 || maxRows > 0) {
+ try {
+ int fetchSizeBounded = fetchSize;
+ if (maxRows > 0 && rowsFetched + fetchSize > maxRows) {
+ fetchSizeBounded = maxRows - (int)rowsFetched;
+ }
TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle,
- orientation, fetchSize);
- LOG.debug("HiveQueryResultsFetchReq: {}", fetchReq);
- TFetchResultsResp fetchResp;
- fetchResp = client.FetchResults(fetchReq);
+ orientation, fetchSizeBounded);
+ TFetchResultsResp fetchResp = client.FetchResults(fetchReq);
Utils.verifySuccessWithInfo(fetchResp.getStatus());
-
TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
fetchedRows = RowSetFactory.create(results, protocol);
- fetchedRowsItr = fetchedRows.iterator();
+ fetchDone = !fetchResp.isHasMoreRows() && fetchedRows.numRows() == 0;
+ if (fetchDone) {
+ gotLastBatch = true;
+ }
+ fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ fetchedRows = new ConvertedResultSet(fetchedRows, getSchema());
+ }
+
+ nextStartRow.set(results.getStartRowOffset() + 1 + fetchedRows.numRows());
+ } catch (TException ex) {
+ ex.printStackTrace();
+ throw new SQLException("Error retrieving next row", ex);
+ }
+ } else {
+ if (!gotLastBatch) {
+ if (pool == null) {
+ pool = Executors.newFixedThreadPool(poolSize);
+ }
+ // Add another thread on each row batch up to the limit
+ if(threadsStarted < (hasStartRow ? poolSize : 1)) {
+ final boolean useMainClient = (threadsStarted == 0);
+ threadsStarted++;
+ pool.execute(()->{
+ LOG.debug("Started thread {}", Thread.currentThread().getName());
+
+ final TCLIService.Iface fetchClient;
+ if (useMainClient) {
+ fetchClient = client;
+ } else {
+ try {
+ fetchClient = ((HiveConnection)connection).Clone().GetClient();
+ } catch (SQLException e) {
+ LOG.debug("Multi-stream connection error {}", e.toString());
+ return;
+ }
+ }
+
+ while (!gotLastBatch && !poolDone) {
+ FetchResult result = new FetchResult();
+ try {
+ final TFetchResultsReq fetchReq = new TFetchResultsReq(
+ stmtHandle, orientation, fetchSize);
+ TFetchResultsResp fetchResp = fetchClient.FetchResults(fetchReq);
+ Utils.verifySuccessWithInfo(fetchResp.getStatus());
+ TRowSet results = fetchResp.getResults();
+ if (results.getStartRowOffset() > 0) {
+ hasStartRow = true;
+ }
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ result.numRows = result.fetchedRows.numRows();
+ boolean hasMoreRows = result.numRows > 0 || fetchResp.isHasMoreRows();
+ if (!hasMoreRows) {
+ gotLastBatch = true;
+ }
+ result.hasMoreRows = hasMoreRows;
+ result.fetchedRows = RowSetFactory.create(results, protocol);
+ if (useConvertedResultSet) {
+ result.fetchedRows =
+ new ConvertedResultSet(result.fetchedRows, getSchema());
+ }
+ result.startRow = results.getStartRowOffset() + 1;
+ if (hasStartRow && result.startRow < nextStartRow.get()) {
+ throw new SQLException("Unexpected row offset");
+ }
+ } catch (Exception e) {
+ result.ex = e;
+ }
+
+ try {
+ // Wait for earlier row sets to be added to the queue
+ synchronized(nextStartRow) {
+ if (!poolDone) {
+ if (result.ex == null) {
+ if (hasStartRow) {
+ while (nextStartRow.get() != result.startRow) {
+ nextStartRow.wait();
+ }
+ nextStartRow.set(result.startRow + result.numRows);
+ }
+ poolDone = !result.hasMoreRows;
+ } else {
+ poolDone = true;
+ }
+ resultQueue.put(result);
+ if (hasStartRow) {
+ nextStartRow.notifyAll();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ interruptException = e;
+ break;
+ }
+ }
+ // TODO: shutdown called here can make execute return
+ // RejectedExecutionException
+ // pool.shutdownNow();
+ });
+ }
+ }
+ try {
+ if (interruptException != null) {
+ throw interruptException;
+ }
+ FetchResult result = resultQueue.take();
+ fetchDone = !result.hasMoreRows;
+ if (result.ex != null) {
+ ShutdownPool();
+ throw new SQLException(result.ex);
+ }
+ fetchedRows = result.fetchedRows;
+ } catch (InterruptedException e) {
+ throw new SQLException(e);
}
+ }
- if (!fetchedRowsItr.hasNext()) {
- return false;
+ orientation = TFetchOrientation.FETCH_NEXT;
+ fetchedRowsItr = fetchedRows.iterator();
+
+ return fetchedRowsItr.hasNext();
+ }
+
+ void DrainQueue() throws SQLException {
+ FetchResult result;
+ while ((result = resultQueue.poll()) != null) {
+ if (result.ex != null) {
+ throw new SQLException(result.ex);
}
+ }
+ }
- row = fetchedRowsItr.next();
- rowsFetched++;
- } catch (SQLException eS) {
- throw eS;
- } catch (Exception ex) {
- throw new SQLException("Error retrieving next row", ex);
+ void ShutdownPool() throws SQLException {
+ if (pool != null) {
+ poolDone = true;
+ DrainQueue();
+ pool.shutdownNow();
+ try {
+ while (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
+ DrainQueue();
Review Comment:
why do you need to call it multiple times: before, in and after the term loop?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2132897491
thanks for the PR @MikaelSmith , @kdeschle! In general LGTM, left few comments
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615558262
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -307,6 +333,12 @@ protected HiveConnection(String uri, Properties info,
sessConfMap = connParams.getSessionVars();
setupLoginTimeout();
if (isKerberosAuthMode()) {
+ // Ensure UserGroupInformation includes any authorized Kerberos principals.
+ LOG.debug("Configuring Kerberos mode");
+ Configuration config = new Configuration();
+ config.set("hadoop.security.authentication", "Kerberos");
Review Comment:
use `AuthTypes.KERBEROS.getAuthName`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615573796
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -95,11 +99,23 @@ public int getScale(int column) throws SQLException {
}
public String getSchemaName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ String tableName = getTableName(column);
Review Comment:
use `HiveTableName.of(tableName).getDb()`
##########
jdbc/src/java/org/apache/hive/jdbc/HiveResultSetMetaData.java:
##########
@@ -95,11 +99,23 @@ public int getScale(int column) throws SQLException {
}
public String getSchemaName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ String tableName = getTableName(column);
+ int index = tableName.lastIndexOf(DOT);
+ if (index >= 0) {
+ return tableName.substring(0, index);
+ }
+ // Impala usually doesn't return fully qualified column names. Return "" to avoid
+ // giving false results.
+ return "";
}
public String getTableName(int column) throws SQLException {
- throw new SQLFeatureNotSupportedException("Method not supported");
+ String columnName = getColumnName(column);
Review Comment:
same, use `HiveTableName.of(tableName).getTable()`
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on code in PR #4902:
URL: https://github.com/apache/hive/pull/4902#discussion_r1615558262
##########
jdbc/src/java/org/apache/hive/jdbc/HiveConnection.java:
##########
@@ -307,6 +333,12 @@ protected HiveConnection(String uri, Properties info,
sessConfMap = connParams.getSessionVars();
setupLoginTimeout();
if (isKerberosAuthMode()) {
+ // Ensure UserGroupInformation includes any authorized Kerberos principals.
+ LOG.debug("Configuring Kerberos mode");
+ Configuration config = new Configuration();
+ config.set("hadoop.security.authentication", "Kerberos");
Review Comment:
use `AuthTypes.KERBEROS.getAuthName`
should it be always kerberized?
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2566479554
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902) **Quality Gate passed**
Issues
 [17 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=4902&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=4902&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=4902&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=4902)
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2566483517
merged to master
thanks @MikaelSmith & @kdeschle for the contribution!
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2553440482
i checked the test bundle and there are no failures. rebased with master
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] Multi-stream JDBC changes [hive]
Posted by "deniskuzZ (via GitHub)" <gi...@apache.org>.
deniskuzZ commented on PR #4902:
URL: https://github.com/apache/hive/pull/4902#issuecomment-2554997806
it seems that all splits are done except split-13. It's already more than 5hr
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org