You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ja...@apache.org on 2015/03/20 06:15:18 UTC
[1/5] drill git commit: DRILL-1967: Add null checks to resources in
cleanup of Parquet writer that may not be initialized if no schema has been
provided to it by the time it is cleaned up.
Repository: drill
Updated Branches:
refs/heads/master 4d398edf8 -> 9c9ee8c43
DRILL-1967: Add null checks to resources in cleanup of Parquet writer that may not be initialized if no schema has been provided to it by the time it is cleaned up.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/838fd08a
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/838fd08a
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/838fd08a
Branch: refs/heads/master
Commit: 838fd08aec6378d3709842b7f93cfabf1e14f87a
Parents: 4d398ed
Author: Jason Altekruse <al...@gmail.com>
Authored: Thu Mar 19 19:50:01 2015 -0700
Committer: Jason Altekruse <al...@gmail.com>
Committed: Thu Mar 19 19:50:01 2015 -0700
----------------------------------------------------------------------
.../apache/drill/exec/store/parquet/ParquetRecordWriter.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/838fd08a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
index 8bf9a92..4a8ff5e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java
@@ -315,8 +315,12 @@ public class ParquetRecordWriter extends ParquetOutputRecordWriter {
parquetFileWriter.endBlock();
parquetFileWriter.end(extraMetaData);
}
- store.close();
- ColumnChunkPageWriteStoreExposer.close(pageStore);
+ if (store != null) {
+ store.close();
+ }
+ if (pageStore != null) {
+ ColumnChunkPageWriteStoreExposer.close(pageStore);
+ }
if(oContext!=null){
oContext.close();
}
[4/5] drill git commit: DRILL-1735: Have closing of JDBC connection
free embedded-server resources.
Posted by ja...@apache.org.
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/DriverTest.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/DriverTest.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/DriverTest.java
new file mode 100644
index 0000000..7935215
--- /dev/null
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/DriverTest.java
@@ -0,0 +1,371 @@
+/*
+ * 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.drill.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Properties;
+
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.test.DrillTest;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+//import static org.hamcrest.CoreMatchers.*;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.nullValue;
+
+/**
+ * (Some) unit and integration tests for org.apache.drill.jdbc.Driver.
+ */
+public class DriverTest extends DrillTest {
+
+ // TODO: Move Jetty status server disabling to DrillTest.
+ private static final String STATUS_SERVER_PROPERTY_NAME =
+ ExecConstants.HTTP_ENABLE;
+
+ private static final String origJettyPropValue =
+ System.getProperty( STATUS_SERVER_PROPERTY_NAME, "true" );
+
+ // Disable Jetty status server so unit tests run (outside Maven setup).
+ // (TODO: Move this to base test class and/or have Jetty try other ports.)
+ @BeforeClass
+ public static void setUpClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, "false" );
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, origJettyPropValue );
+ }
+
+ private Driver uut = new Driver();
+
+
+ ////////////////////////////////////////
+ // Tests of methods defined by JDBC/java.sql.Driver:
+
+ // Tests for connect() (defined by JDBC/java.sql.Driver):
+
+ @Test
+ public void test_connect_declinesEmptyUrl()
+ throws SQLException
+ {
+ assertThat( uut.connect( "", null ), nullValue() );
+ }
+
+ @Test
+ public void test_connect_declinesNonUrl()
+ throws SQLException
+ {
+ assertThat( uut.connect( "whatever", null ), nullValue() );
+ }
+
+ @Test
+ public void test_connect_declinesNonJdbcUrl()
+ throws SQLException
+ {
+ assertThat( uut.connect( "file:///something", null ), nullValue() );
+ }
+
+ @Test
+ public void test_connect_declinesNonDrillJdbcUrl()
+ throws SQLException
+ {
+ assertThat( uut.connect( "jdbc:somedb:whatever", null ), nullValue() );
+ }
+
+ @Test
+ public void test_connect_declinesNotQuiteDrillUrl()
+ throws SQLException
+ {
+ assertThat( uut.connect( "jdbc:drill", null ), nullValue() );
+ }
+
+ // TODO Determine whether this "jdbc:drill:" should be valid or error.
+ @Ignore( "Just hangs, trying to connect to non-existent local zookeeper." )
+ @Test
+ public void test_connect_acceptsMinimalDrillJdbcUrl()
+ throws SQLException
+ {
+ assertThat( uut.connect( "jdbc:drill:", null ), nullValue() );
+ fail( "Not implemented yet" );
+ }
+
+ // TODO: Determine rules for Drill JDBC URLs. (E.g., is "zk=..." always
+ // required? What other properties are allowed? What is disallowed?)
+ @Ignore( "Just hangs, trying to connect to non-existent local zookeeper." )
+ @Test
+ public void test_connect_DECIDEWHICHBogusDrillJdbcUrl()
+ throws SQLException
+ {
+ assertThat( uut.connect( "jdbc:drill:x=y;z;;a=b=c=d;what=ever", null ),
+ nullValue() );
+ fail( "Not implemented yet" );
+ }
+
+ @Test
+ public void test_connect_acceptsLocalZkDrillJdbcUrl()
+ throws SQLException
+ {
+ Connection connection = uut.connect( "jdbc:drill:zk=local", null );
+ assertThat( connection, not( nullValue() ) );
+ connection.close();
+ }
+
+ @Test
+ public void test_connect_acceptsLocalViaProperties()
+ throws SQLException
+ {
+ Properties props = new Properties();
+ props.put( "zk", "local" );
+
+ Connection connection = uut.connect( "jdbc:drill:", props );
+ assertThat( connection, not( nullValue() ) );
+ connection.close();
+ }
+
+ // TODO: Determine which other cases to test, including cases of Properties
+ // parameter values to test.
+
+
+ // Tests for acceptsURL(String) (defined by JDBC/java.sql.Driver):
+
+ @Test
+ public void test_acceptsURL_acceptsDrillUrlMinimal()
+ throws SQLException
+ {
+ assertThat( uut.acceptsURL("jdbc:drill:"), equalTo( true ) );
+ }
+
+ @Test
+ public void test_acceptsURL_acceptsDrillPlusJunk()
+ throws SQLException
+ {
+ assertThat( uut.acceptsURL("jdbc:drill:should it check this?"),
+ equalTo( true ) );
+ }
+
+ @Test
+ public void test_acceptsURL_rejectsNonDrillJdbcUrl()
+ throws SQLException
+ {
+ assertThat( uut.acceptsURL("jdbc:notdrill:whatever"), equalTo( false ) );
+ }
+
+ @Test
+ public void test_acceptsURL_rejectsNonDrillJdbc2()
+ throws SQLException
+ {
+ assertThat( uut.acceptsURL("jdbc:optiq:"), equalTo( false ) );
+ }
+
+ @Test
+ public void test_acceptsURL_rejectsNonJdbcUrl()
+ throws SQLException
+ {
+ assertThat( uut.acceptsURL("drill:"), equalTo( false ) );
+ }
+
+
+ // Tests for getPropertyInfo(String, Properties) (defined by
+ // JDBC/java.sql.Driver):
+ // TODO: Determine what properties (if any) should be returned.
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getPropertyInfo()
+ throws SQLException
+ {
+ fail( "Not implemented yet" );
+ }
+
+
+ // Tests for getMajorVersion() (defined by JDBC/java.sql.Driver):
+ // TODO: Determine what major version number should be.
+ @Test
+ public void test_getMajorVersion() {
+ assertThat( uut.getMajorVersion(), org.hamcrest.CoreMatchers.is( 0 ) );
+ }
+
+
+ // Tests for getMinorVersion() (defined by JDBC/java.sql.Driver):
+ // TODO: Determine what minor version number should be.
+ @Test
+ public void test_getMinorVersion() {
+ assertThat( uut.getMinorVersion(), org.hamcrest.core.Is.is( 0 ) );
+ }
+
+
+ // Tests for jdbcCompliant() (defined by JDBC/java.sql.Driver):
+ // TODO Determine what we choose to return. If it doesn't match what
+ // java.sql.Driver's documentation says, document that on DrillResultSet and
+ // where users (programmers) can see it.
+ @Ignore( "Seemingly: Bug: Returns true, but hasn't passed compliance tests." )
+ @Test
+ public void test_jdbcCompliant() {
+ // Expect false because not known to have "[passed] the JDBC compliance
+ // tests."
+ assertThat( uut.jdbcCompliant(), equalTo( false ) );
+ }
+
+ // Tests for XXX (defined by JDBC/java.sql.Driver):
+ // Defined by JDBC/java.sql.Driver: getParentLogger()
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getParentLogger()
+ {
+ fail( "Not implemented yet" );
+ }
+
+
+ // Tests for XXX (defined by JDBC/java.sql.Driver):
+ // Defined by JDBC/java.sql.Driver: "[driver] should create and instance of
+ // itself and register it with the DriverManager.
+ @Test
+ public void test_Driver_registersWithManager()
+ throws SQLException
+ {
+ assertThat( DriverManager.getDriver( "jdbc:drill:whatever" ),
+ instanceOf( Driver.class ) );
+ }
+
+
+ ////////////////////////////////////////
+ // Tests of methods defined by net.hydromatic.avatica.UnregisteredDriver.
+
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_load() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getConnectStringPrefix() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getFactoryClassNameJdbcVersion()
+ {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_createDriverVersion() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_createHandler() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_unregisteredDriver() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_createFactory() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_createHandler1() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getFactoryClassNameJdbcVersion1()
+ {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_createDriverVersion1() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getConnectionProperties() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_instantiateFactory() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getConnectStringPrefix1() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_getDriverVersion() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_register() {
+ fail( "Not implemented yet" );
+ }
+
+ ////////////////////////////////////////
+
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_hashCode() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_equals() {
+ fail( "Not implemented yet" );
+ }
+
+ @Ignore( "Deferred pending need." )
+ @Test
+ public void test_toString() {
+ fail( "Not implemented yet" );
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/JdbcTest.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/JdbcTest.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/JdbcTest.java
index 40b1445..6d8d8e9 100644
--- a/exec/jdbc/src/test/java/org/apache/drill/jdbc/JdbcTest.java
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/JdbcTest.java
@@ -48,7 +48,7 @@ public class JdbcTest extends ExecTest {
private static CachingConnectionFactory factory;
@BeforeClass
- public static void setUp() {
+ public static void setUpTestCase() {
factory = new SingleConnectionCachingFactory(new ConnectionFactory() {
@Override
public Connection createConnection(ConnectionInfo info) throws Exception {
@@ -92,11 +92,24 @@ public class JdbcTest extends ExecTest {
}
}
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment race
+ // conditions are fixed (not just DRILL-2245 fixes).
+ ///**
+ // * Calls {@link ResultSet#next} on given {@code ResultSet} until it returns
+ // * false. (For TEMPORARY workaround for query cancelation race condition.)
+ // */
+ //private static void nextUntilEnd(final ResultSet resultSet) throws SQLException {
+ // while (resultSet.next()) {
+ // }
+ //}
+
protected static void changeSchema(Connection conn, String schema) {
final String query = String.format("use %s", schema);
- try {
- Statement s = conn.createStatement();
+ try ( Statement s = conn.createStatement() ) {
ResultSet r = s.executeQuery(query);
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(r);
} catch (SQLException e) {
throw new RuntimeException("unable to change schema", e);
}
@@ -114,7 +127,7 @@ public class JdbcTest extends ExecTest {
}
@AfterClass
- public static void clearUp() throws Exception {
+ public static void tearDownTestCase() throws Exception {
factory.close();
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ConnectionCloseTest.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ConnectionCloseTest.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ConnectionCloseTest.java
new file mode 100644
index 0000000..32240d6
--- /dev/null
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ConnectionCloseTest.java
@@ -0,0 +1,102 @@
+/**
+ * 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.drill.jdbc.test;
+
+import java.sql.Connection;
+
+import org.apache.drill.common.util.TestTools;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.jdbc.Driver;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestRule;
+import org.slf4j.Logger;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+
+/**
+* Test for DRILL-1735: Closing local JDBC connection didn't shut down
+* local DrillBit to free resources (plus QueryResultBatch buffer allocation leak
+* in DrillCursor.next(), lack of DrillMetrics reset, vectors buffer leak under
+* DrillCursor/DrillResultSet, and other problems).
+*/
+public class Bug1735ConnectionCloseTest extends JdbcTestQueryBase {
+
+ static final Logger logger = getLogger( Bug1735ConnectionCloseTest.class );
+
+ @Rule
+ public TestRule TIMEOUT = TestTools.getTimeoutRule( 120_000 /* ms */ );
+
+ // TODO: Move Jetty status server disabling to DrillTest.
+
+ private static final String STATUS_SERVER_PROPERTY_NAME =
+ ExecConstants.HTTP_ENABLE;
+
+ private static final String origStatusServerPropValue =
+ System.getProperty( STATUS_SERVER_PROPERTY_NAME, "true" );
+
+ // Disable Jetty status server so unit tests run (outside Maven setup).
+ // (TODO: Move this to base test class and/or have Jetty try other ports.)
+ @BeforeClass
+ public static void setUpClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, "false" );
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, origStatusServerPropValue );
+ }
+
+
+ // Basic sanity test (too small to detect original connection close problem
+ // but would detect QueryResultBatch release and metrics problems).
+
+ private static final int SMALL_ITERATION_COUNT = 3;
+
+ @Test
+ public void testCloseDoesntLeakResourcesBasic() throws Exception {
+ for ( int i = 1; i <= SMALL_ITERATION_COUNT; i++ ) {
+ logger.info( "iteration " + i + ":" );
+ System.out.println( "iteration " + i + ":" );
+ Connection connection = new Driver().connect( "jdbc:drill:zk=local", null );
+ connection.close();
+ }
+ }
+
+
+ // Test large enough to detect connection close problem (at least on
+ // developer's machine).
+
+ private static final int LARGE_ITERATION_COUNT = 1000;
+
+ @Ignore( "Normally suppressed because slow" )
+ @Test
+ public void testCloseDoesntLeakResourcesMany() throws Exception {
+ for ( int i = 1; i <= LARGE_ITERATION_COUNT; i++ ) {
+ logger.info( "iteration " + i + ":" );
+ System.out.println( "iteration " + i + ":" );
+ Connection connection = new Driver().connect( "jdbc:drill:zk=local", null );
+ connection.close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ResultSetCloseReleasesBuffersTest.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ResultSetCloseReleasesBuffersTest.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ResultSetCloseReleasesBuffersTest.java
new file mode 100644
index 0000000..1434640
--- /dev/null
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/Bug1735ResultSetCloseReleasesBuffersTest.java
@@ -0,0 +1,89 @@
+/**
+ * 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.drill.jdbc.test;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import org.apache.drill.exec.ExecConstants;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.base.Function;
+
+public class Bug1735ResultSetCloseReleasesBuffersTest extends JdbcTestQueryBase {
+
+ // TODO: Move Jetty status server disabling to DrillTest.
+
+ private static final String STATUS_SERVER_PROPERTY_NAME =
+ ExecConstants.HTTP_ENABLE;
+
+ private static final String origStatusServerPropValue =
+ System.getProperty( STATUS_SERVER_PROPERTY_NAME, "true" );
+
+ // Disable Jetty status server so unit tests run (outside Maven setup).
+ // (TODO: Move this to base test class and/or have Jetty try other ports.)
+ @BeforeClass
+ public static void setUpClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, "false" );
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, origStatusServerPropValue );
+ }
+
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment race
+ // conditions are fixed (not just DRILL-2245 fixes).
+ ///**
+ // * Calls {@link ResultSet#next} on given {@code ResultSet} until it returns
+ // * false. (For TEMPORARY workaround for query cancelation race condition.)
+ // */
+ //private void nextUntilEnd(final ResultSet resultSet) throws SQLException {
+ // while (resultSet.next()) {
+ // }
+ //}
+
+ @Test
+ public void test() throws Exception {
+ JdbcAssert
+ .withNoDefaultSchema()
+ .withConnection(
+ new Function<Connection, Void>() {
+ public Void apply( Connection connection ) {
+ try {
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery( "USE dfs_test.tmp" );
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // resultSet.close( resultSet );
+ statement.close();
+ // connection.close() is in withConnection(...)
+ return null;
+ } catch ( SQLException e ) {
+ throw new RuntimeException( e );
+ }
+ }
+ });
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcDistQuery.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcDistQuery.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcDistQuery.java
index b2f86ea..a8f47c6 100644
--- a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcDistQuery.java
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcDistQuery.java
@@ -22,6 +22,7 @@ import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -38,7 +39,7 @@ import org.junit.rules.TestRule;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
-public class TestJdbcDistQuery extends JdbcTest{
+public class TestJdbcDistQuery extends JdbcTest {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestJdbcDistQuery.class);
@@ -52,6 +53,17 @@ public class TestJdbcDistQuery extends JdbcTest{
}
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment race
+ // conditions are fixed (not just DRILL-2245 fixes).
+ /**
+ * Calls {@link ResultSet#next} on given {@code ResultSet} until it returns
+ * false. (For TEMPORARY workaround for query cancelation race condition.)
+ */
+ private static void nextUntilEnd(final ResultSet resultSet) throws SQLException {
+ while (resultSet.next()) {
+ }
+ }
+
@Test
public void testSimpleQuerySingleFile() throws Exception{
testQuery(String.format("select R_REGIONKEY, R_NAME "
@@ -183,13 +195,17 @@ public class TestJdbcDistQuery extends JdbcTest{
private void testQuery(String sql) throws Exception{
boolean success = false;
try (Connection c = DriverManager.getConnection("jdbc:drill:zk=local", null);) {
+ // ???? TODO: What is this currently redundant one-time loop for? (If
+ // it's keep around to make it easy to switch to looping multiple times
+ // (e.g., for debugging) then define a constant field or local variable
+ // for the number of iterations.)
for (int x = 0; x < 1; x++) {
Stopwatch watch = new Stopwatch().start();
Statement s = c.createStatement();
ResultSet r = s.executeQuery(sql);
boolean first = true;
ResultSetMetaData md = r.getMetaData();
- if (first == true) {
+ if (first) {
for (int i = 1; i <= md.getColumnCount(); i++) {
System.out.print(md.getColumnName(i));
System.out.print('\t');
@@ -235,6 +251,9 @@ public class TestJdbcDistQuery extends JdbcTest{
String[] expected = {"fullname", "occupation", "postal_code"};
Assert.assertEquals(3, md.getColumnCount());
Assert.assertArrayEquals(expected, columns.toArray());
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment race
+ // conditions are fixed (not just DRILL-2245 fixes).
+ nextUntilEnd(r);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
index 682fca3..09ad5db 100644
--- a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
@@ -21,6 +21,7 @@ import static org.junit.Assert.assertEquals;
import java.sql.Connection;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
@@ -32,6 +33,17 @@ import com.google.common.base.Function;
public class TestJdbcQuery extends JdbcTestQueryBase{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestJdbcQuery.class);
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment race
+ // conditions are fixed (not just DRILL-2245 fixes).
+ ///**
+ // * Calls {@link ResultSet#next} on given {@code ResultSet} until it returns
+ // * false. (For TEMPORARY workaround for query cancelation race condition.)
+ // */
+ //private void nextUntilEnd(final ResultSet resultSet) throws SQLException {
+ // while (resultSet.next()) {
+ // }
+ //}
+
@Test
@Ignore
public void testJsonQuery() throws Exception{
@@ -214,6 +226,9 @@ public class TestJdbcQuery extends JdbcTestQueryBase{
"\ninterval year: " + intervalYear + " intervalDay: " + intervalDay +
" date_interval_add: " + ts1.toString() + "date_int_add: " + date1.toString());
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
statement.close();
return null;
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestMetadataDDL.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestMetadataDDL.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestMetadataDDL.java
index 93925fe..41a595d 100644
--- a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestMetadataDDL.java
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestMetadataDDL.java
@@ -21,6 +21,7 @@ import static org.junit.Assert.assertTrue;
import java.sql.Connection;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
import java.util.Set;
@@ -40,6 +41,17 @@ import com.google.common.collect.ImmutableSet;
*/
public class TestMetadataDDL extends JdbcTestQueryBase {
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment race
+ // conditions are fixed (not just DRILL-2245 fixes).
+ ///**
+ // * Calls {@link ResultSet#next} on given {@code ResultSet} until it returns
+ // * false. (For TEMPORARY workaround for query cancelation race condition.)
+ // */
+ //private void nextUntilEnd(final ResultSet resultSet) throws SQLException {
+ // while (resultSet.next()) {
+ // }
+ //}
+
@BeforeClass
public static void generateHive() throws Exception{
new HiveTestDataGenerator().generateTestData();
@@ -197,14 +209,23 @@ public class TestMetadataDDL extends JdbcTestQueryBase {
JdbcAssert.withNoDefaultSchema().withConnection(new Function<Connection, Void>() {
public Void apply(Connection connection) {
try {
+ ResultSet resultSet;
Statement statement = connection.createStatement();
- statement.executeQuery("USE dfs_test.tmp").close();
+ resultSet = statement.executeQuery("USE dfs_test.tmp");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
// INFORMATION_SCHEMA already has a table named "TABLES". Now create a table with same name in "dfs_test.tmp" schema
- statement.executeQuery("CREATE OR REPLACE VIEW `TABLES` AS SELECT key FROM hive_test.kv").close();
+ resultSet = statement.executeQuery("CREATE OR REPLACE VIEW `TABLES` AS SELECT key FROM hive_test.kv");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
// Test describe of `TABLES` with no schema qualifier
- ResultSet resultSet = statement.executeQuery("DESCRIBE `TABLES`");
+ resultSet = statement.executeQuery("DESCRIBE `TABLES`");
Set<String> result = JdbcAssert.toStringSet(resultSet);
resultSet.close();
ImmutableSet<String> expected = ImmutableSet.of("COLUMN_NAME=key; DATA_TYPE=INTEGER; IS_NULLABLE=NO");
@@ -222,7 +243,11 @@ public class TestMetadataDDL extends JdbcTestQueryBase {
assertTrue(String.format("Generated string:\n%s\ndoes not match:\n%s", result, expected), expected.equals(result));
// drop created view
- statement.executeQuery("DROP VIEW `TABLES`").close();
+ resultSet = statement.executeQuery("DROP VIEW `TABLES`");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
statement.close();
return null;
@@ -420,13 +445,17 @@ public class TestMetadataDDL extends JdbcTestQueryBase {
JdbcAssert.withNoDefaultSchema().withConnection(new Function<Connection, Void>() {
public Void apply(Connection connection) {
try {
+ ResultSet resultSet;
Statement statement = connection.createStatement();
// change default schema
- statement.executeQuery("USE dfs_test.`default`");
+ resultSet = statement.executeQuery("USE dfs_test.`default`");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// show files
- ResultSet resultSet = statement.executeQuery("show files from `/tmp`");
+ resultSet = statement.executeQuery("show files from `/tmp`");
System.out.println(JdbcAssert.toString(resultSet));
resultSet.close();
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestViews.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestViews.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestViews.java
index 0f9e25e..8042a6d 100644
--- a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestViews.java
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestViews.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue;
import java.io.File;
import java.sql.Connection;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
import org.apache.commons.io.FileUtils;
@@ -33,7 +34,7 @@ import org.junit.Test;
import com.google.common.base.Function;
-/** Contains tests for creating/droping and using views in Drill. */
+/** Contains tests for creating/dropping and using views in Drill. */
public class TestViews extends JdbcTestQueryBase {
@BeforeClass
@@ -48,6 +49,17 @@ public class TestViews extends JdbcTestQueryBase {
}
}
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment race
+ // conditions are fixed (not just DRILL-2245 fixes).
+ ///**
+ // * Calls {@link ResultSet#next} on given {@code ResultSet} until it returns
+ // * false. (For TEMPORARY workaround for query cancelation race condition.)
+ // */
+ //private void nextUntilEnd(final ResultSet resultSet) throws SQLException {
+ // while (resultSet.next()) {
+ // }
+ //}
+
/** Helper test method for view tests */
private void testViewHelper(final String viewCreate, final String viewName,
final String viewQuery, final String queryResult) throws Exception{
@@ -55,12 +67,16 @@ public class TestViews extends JdbcTestQueryBase {
public Void apply(Connection connection) {
try {
Statement statement = connection.createStatement();
+ ResultSet resultSet;
// change default schema
- statement.executeQuery("USE dfs_test.tmp");
+ resultSet = statement.executeQuery("USE dfs_test.tmp");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// create view
- ResultSet resultSet = statement.executeQuery(viewCreate);
+ resultSet = statement.executeQuery(viewCreate);
String result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
String viewCreateResult = "ok=true; summary=View '" + viewName + "' created successfully in 'dfs_test.tmp' schema";
@@ -74,7 +90,10 @@ public class TestViews extends JdbcTestQueryBase {
assertTrue(String.format("Generated string:\n%s\ndoes not match:\n%s", result, queryResult),
queryResult.equals(result));
- statement.executeQuery("drop view " + viewName).close();
+ resultSet = statement.executeQuery("drop view " + viewName);
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
statement.close();
return null;
@@ -240,16 +259,23 @@ public class TestViews extends JdbcTestQueryBase {
public Void apply(Connection connection) {
try {
Statement statement = connection.createStatement();
+ ResultSet resultSet;
// change default schema
- statement.executeQuery("USE dfs_test.tmp");
+ resultSet = statement.executeQuery("USE dfs_test.tmp");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// create view
- statement.executeQuery(
+ resultSet = statement.executeQuery(
"CREATE VIEW testview3(regionid) AS SELECT region_id FROM cp.`region.json`");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// query from view
- ResultSet resultSet = statement.executeQuery("SELECT regionid FROM testview3 LIMIT 1");
+ resultSet = statement.executeQuery("SELECT regionid FROM testview3 LIMIT 1");
String result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
String expected = "regionid=0";
@@ -278,16 +304,23 @@ public class TestViews extends JdbcTestQueryBase {
public Void apply(Connection connection) {
try {
Statement statement = connection.createStatement();
+ ResultSet resultSet;
// change default schema
- statement.executeQuery("USE dfs_test.tmp");
+ resultSet = statement.executeQuery("USE dfs_test.tmp");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// create view
- statement.executeQuery(
+ resultSet = statement.executeQuery(
"CREATE VIEW testview3 AS SELECT * FROM hive_test.kv");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// show tables on view
- ResultSet resultSet = statement.executeQuery("SHOW TABLES like 'testview3'");
+ resultSet = statement.executeQuery("SHOW TABLES like 'testview3'");
String result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
String expected = "TABLE_SCHEMA=dfs_test.tmp; TABLE_NAME=testview3";
@@ -322,7 +355,11 @@ public class TestViews extends JdbcTestQueryBase {
assertTrue(String.format("Generated string:\n%s\ndoes not match:\n%s", result, expected),
expected.equals(result));
- statement.executeQuery("drop view testview3").close();
+ resultSet = statement.executeQuery("drop view testview3");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
statement.close();
return null;
@@ -347,12 +384,16 @@ public class TestViews extends JdbcTestQueryBase {
public Void apply(Connection connection) {
try {
Statement statement = connection.createStatement();
+ ResultSet resultSet;
// change default schema
- statement.executeQuery("USE cp");
+ resultSet = statement.executeQuery("USE cp");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// create a view with full schema identifier
- ResultSet resultSet = statement.executeQuery("CREATE VIEW dfs_test.tmp.testview AS SELECT * FROM hive_test.kv");
+ resultSet = statement.executeQuery("CREATE VIEW dfs_test.tmp.testview AS SELECT * FROM hive_test.kv");
String result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
String expected = "ok=true; summary=View 'testview' created successfully in 'dfs_test.tmp' schema";
@@ -367,7 +408,11 @@ public class TestViews extends JdbcTestQueryBase {
assertTrue(String.format("Generated string:\n%s\ndoes not match:\n%s", result, expected),
expected.equals(result));
- statement.executeQuery("drop view dfs_test.tmp.testview").close();
+ resultSet = statement.executeQuery("drop view dfs_test.tmp.testview");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
statement.close();
return null;
@@ -384,12 +429,16 @@ public class TestViews extends JdbcTestQueryBase {
public Void apply(Connection connection) {
try {
Statement statement = connection.createStatement();
+ ResultSet resultSet;
// change default schema
- statement.executeQuery("USE dfs_test");
+ resultSet = statement.executeQuery("USE dfs_test");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// create a view with partial schema identifier
- ResultSet resultSet = statement.executeQuery("CREATE VIEW tmp.testview AS SELECT * FROM hive_test.kv");
+ resultSet = statement.executeQuery("CREATE VIEW tmp.testview AS SELECT * FROM hive_test.kv");
String result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
String expected = "ok=true; summary=View 'testview' created successfully in 'dfs_test.tmp' schema";
@@ -405,14 +454,22 @@ public class TestViews extends JdbcTestQueryBase {
expected.equals(result));
// change the default schema and query
- statement.executeQuery("USE dfs_test.tmp");
+ resultSet = statement.executeQuery("USE dfs_test.tmp");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+
resultSet = statement.executeQuery("SELECT key FROM testview LIMIT 1");
result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
assertTrue(String.format("Generated string:\n%s\ndoes not match:\n%s", result, expected),
expected.equals(result));
- statement.executeQuery("drop view tmp.testview").close();
+ resultSet = statement.executeQuery("drop view tmp.testview");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
statement.close();
return null;
@@ -429,12 +486,16 @@ public class TestViews extends JdbcTestQueryBase {
public Void apply(Connection connection) {
try {
Statement statement = connection.createStatement();
+ ResultSet resultSet;
// change default schema
- statement.executeQuery("USE cp");
+ resultSet = statement.executeQuery("USE cp");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
// create a view with full schema identifier
- ResultSet resultSet = statement.executeQuery(
+ resultSet = statement.executeQuery(
"CREATE VIEW dfs_test.tmp.testViewResolvingTablesInWorkspaceSchema AS " +
"SELECT region_id, sales_city FROM `region.json`");
String result = JdbcAssert.toString(resultSet).trim();
@@ -453,7 +514,11 @@ public class TestViews extends JdbcTestQueryBase {
assertTrue(String.format("Generated string:\n%s\ndoes not match:\n%s", result, expected),
expected.equals(result));
- statement.executeQuery("drop view dfs_test.tmp.testViewResolvingTablesInWorkspaceSchema").close();
+ resultSet = statement.executeQuery("drop view dfs_test.tmp.testViewResolvingTablesInWorkspaceSchema");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
statement.close();
return null;
@@ -472,7 +537,7 @@ public class TestViews extends JdbcTestQueryBase {
Statement statement = connection.createStatement();
// create a view
- ResultSet resultSet = statement.executeQuery(
+ ResultSet resultSet = statement.executeQuery(
"CREATE VIEW testCreateViewWhenViewAlreadyExists AS SELECT region_id, sales_city FROM cp.`region.json`");
String result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
@@ -482,7 +547,7 @@ public class TestViews extends JdbcTestQueryBase {
expected.equals(result));
// try to create the view with same name
- resultSet = statement.executeQuery(
+ resultSet = statement.executeQuery(
"CREATE VIEW testCreateViewWhenViewAlreadyExists AS SELECT region_id FROM cp.`region.json`");
result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
@@ -491,7 +556,7 @@ public class TestViews extends JdbcTestQueryBase {
expected.equals(result));
// try creating the view with same name but with a OR REPLACE clause
- resultSet = statement.executeQuery(
+ resultSet = statement.executeQuery(
"CREATE OR REPLACE VIEW testCreateViewWhenViewAlreadyExists AS SELECT region_id FROM cp.`region.json`");
result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
@@ -500,7 +565,11 @@ public class TestViews extends JdbcTestQueryBase {
assertTrue(String.format("Generated string:\n%s\ndoes not match:\n%s", result, expected),
expected.equals(result));
- statement.executeQuery("drop view dfs_test.tmp.testCreateViewWhenViewAlreadyExists").close();
+ resultSet = statement.executeQuery("drop view dfs_test.tmp.testCreateViewWhenViewAlreadyExists");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
statement.close();
return null;
@@ -512,7 +581,7 @@ public class TestViews extends JdbcTestQueryBase {
}
private void createViewHelper(Statement statement, String schema, String viewName, String query) throws Exception {
- ResultSet resultSet = statement.executeQuery(query);
+ ResultSet resultSet = statement.executeQuery(query);
String result = JdbcAssert.toString(resultSet).trim();
resultSet.close();
String expected = String.format("ok=true; summary=View '%s' created successfully in '%s' schema", viewName, schema);
@@ -531,7 +600,11 @@ public class TestViews extends JdbcTestQueryBase {
if (schema != null && !schema.isEmpty()) {
viewName = schema + "." + viewName;
}
- statement.executeQuery("drop view innerView").close();
+ ResultSet resultSet = statement.executeQuery("drop view innerView");
+ // TODO: Purge nextUntilEnd(...) and calls when remaining fragment
+ // race conditions are fixed (not just DRILL-2245 fixes).
+ // nextUntilEnd(resultSet);
+ resultSet.close();
}
@Test
[2/5] drill git commit: DRILL-2503: AsmUtil ClassTransformer
MergeAdapter - add option to pass through ClassReader.EXPAND_FRAMES to
satisfy complaint from ASM - rationalize AsmUtils methods' argument lists
Posted by ja...@apache.org.
DRILL-2503: AsmUtil ClassTransformer MergeAdapter - add option to pass through ClassReader.EXPAND_FRAMES to satisfy complaint from ASM - rationalize AsmUtils methods' argument lists
TestBugFixes (in ...drill.jdbc.test)
- created this to hold random bug fix tests
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/7af5f9a0
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/7af5f9a0
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/7af5f9a0
Branch: refs/heads/master
Commit: 7af5f9a01b1433bb8bb62c1e04a6cef68d629b48
Parents: 838fd08
Author: Chris Westin <cw...@yahoo.com>
Authored: Thu Mar 19 17:13:35 2015 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Thu Mar 19 22:14:14 2015 -0700
----------------------------------------------------------------------
.../org/apache/drill/exec/compile/AsmUtil.java | 25 +++---
.../drill/exec/compile/ClassTransformer.java | 5 +-
.../apache/drill/exec/compile/MergeAdapter.java | 58 +++-----------
.../apache/drill/jdbc/test/TestBugFixes.java | 84 ++++++++++++++++++++
4 files changed, 109 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/7af5f9a0/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AsmUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AsmUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AsmUtil.java
index 032aebd..81904df 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AsmUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AsmUtil.java
@@ -42,12 +42,12 @@ public class AsmUtil {
/**
* Check to see if a class is well-formed.
*
- * @param classNode the class to check
- * @param logTag a tag to print to the log if a problem is found
* @param logger the logger to write to if a problem is found
+ * @param logTag a tag to print to the log if a problem is found
+ * @param classNode the class to check
* @return true if the class is ok, false otherwise
*/
- public static boolean isClassOk(final Logger logger, final ClassNode classNode, final String logTag) {
+ public static boolean isClassOk(final Logger logger, final String logTag, final ClassNode classNode) {
final StringWriter sw = new StringWriter();
final ClassWriter verifyWriter = new ClassWriter(ClassWriter.COMPUTE_FRAMES);
classNode.accept(verifyWriter);
@@ -71,26 +71,27 @@ public class AsmUtil {
/**
* Check to see if a class is well-formed.
*
- * @param classBytes the bytecode of the class to check
- * @param logTag a tag to print to the log if a problem is found
* @param logger the logger to write to if a problem is found
+ * @param logTag a tag to print to the log if a problem is found
+ * @param classBytes the bytecode of the class to check
* @return true if the class is ok, false otherwise
*/
- public static boolean isClassBytesOk(final Logger logger, final byte[] classBytes, final String logTag) {
- final ClassNode classNode = classFromBytes(classBytes);
- return isClassOk(logger, classNode, logTag);
+ public static boolean isClassBytesOk(final Logger logger, final String logTag, final byte[] classBytes) {
+ final ClassNode classNode = classFromBytes(classBytes, 0);
+ return isClassOk(logger, logTag, classNode);
}
/**
* Create a ClassNode from bytecode.
*
* @param classBytes the bytecode
+ * @param asmReaderFlags flags for ASM; see {@link org.objectweb.asm.ClassReader#accept(org.objectweb.asm.ClassVisitor, int)}
* @return the ClassNode
*/
- public static ClassNode classFromBytes(final byte[] classBytes) {
+ public static ClassNode classFromBytes(final byte[] classBytes, final int asmReaderFlags) {
final ClassNode classNode = new ClassNode(CompilationConfig.ASM_API_VERSION);
final ClassReader classReader = new ClassReader(classBytes);
- classReader.accept(classNode, 0);
+ classReader.accept(classNode, asmReaderFlags);
return classNode;
}
@@ -99,9 +100,9 @@ public class AsmUtil {
*
* <p>Writes at level DEBUG.
*
+ * @param logger the logger to write to
* @param logTag a tag to print to the log
* @param classNode the class
- * @param logger the logger to write to
*/
public static void logClass(final Logger logger, final String logTag, final ClassNode classNode) {
logger.debug(logTag);
@@ -122,7 +123,7 @@ public class AsmUtil {
* @param logger the logger to write to
*/
public static void logClassFromBytes(final Logger logger, final String logTag, final byte[] classBytes) {
- final ClassNode classNode = classFromBytes(classBytes);
+ final ClassNode classNode = classFromBytes(classBytes, 0);
logClass(logger, logTag, classNode);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/7af5f9a0/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
index 493f6ce..d4d74dd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
@@ -30,6 +30,7 @@ import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.server.options.OptionValue;
import org.apache.drill.exec.server.options.TypeValidators.EnumeratedStringValidator;
import org.codehaus.commons.compiler.CompileException;
+import org.objectweb.asm.ClassReader;
import org.objectweb.asm.tree.ClassNode;
import com.google.common.annotations.VisibleForTesting;
@@ -226,8 +227,8 @@ public class ClassTransformer {
Map<String, ClassNode> classesToMerge = Maps.newHashMap();
for (byte[] clazz : implementationClasses) {
totalBytecodeSize += clazz.length;
- final ClassNode node = AsmUtil.classFromBytes(clazz);
- if (!AsmUtil.isClassOk(logger, node, "implementationClasses")) {
+ final ClassNode node = AsmUtil.classFromBytes(clazz, ClassReader.EXPAND_FRAMES);
+ if (!AsmUtil.isClassOk(logger, "implementationClasses", node)) {
throw new IllegalStateException("Problem found with implementationClasses");
}
classesToMerge.put(node.name, node);
http://git-wip-us.apache.org/repos/asf/drill/blob/7af5f9a0/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
index 1522102..82bd413 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
@@ -17,8 +17,6 @@
*/
package org.apache.drill.exec.compile;
-import java.io.PrintWriter;
-import java.io.StringWriter;
import java.lang.reflect.Modifier;
import java.util.Collection;
import java.util.Iterator;
@@ -39,7 +37,6 @@ import org.objectweb.asm.commons.SimpleRemapper;
import org.objectweb.asm.tree.ClassNode;
import org.objectweb.asm.tree.FieldNode;
import org.objectweb.asm.tree.MethodNode;
-import org.objectweb.asm.util.TraceClassVisitor;
import com.google.common.collect.Sets;
@@ -191,10 +188,10 @@ class MergeAdapter extends ClassVisitor {
public static MergedClassResult getMergedClass(final ClassSet set, final byte[] precompiledClass,
ClassNode generatedClass, final boolean scalarReplace) {
if (verifyBytecode) {
- if (!AsmUtil.isClassBytesOk(logger, precompiledClass, "precompiledClass")) {
+ if (!AsmUtil.isClassBytesOk(logger, "precompiledClass", precompiledClass)) {
throw new IllegalStateException("Problem found in precompiledClass");
}
- if ((generatedClass != null) && !AsmUtil.isClassOk(logger, generatedClass, "generatedClass")) {
+ if ((generatedClass != null) && !AsmUtil.isClassOk(logger, "generatedClass", generatedClass)) {
throw new IllegalStateException("Problem found in generatedClass");
}
}
@@ -225,7 +222,7 @@ class MergeAdapter extends ClassVisitor {
*/
generatedClass.accept(new ValueHolderReplacementVisitor(mergeGenerator, verifyBytecode));
if (verifyBytecode) {
- if (!AsmUtil.isClassOk(logger, generatedMerged, "generatedMerged")) {
+ if (!AsmUtil.isClassOk(logger, "generatedMerged", generatedMerged)) {
throw new IllegalStateException("Problem found with generatedMerged");
}
}
@@ -266,15 +263,13 @@ class MergeAdapter extends ClassVisitor {
}
}
-
- static class RemapClasses extends Remapper {
-
+ private static class RemapClasses extends Remapper {
final Set<String> innerClasses = Sets.newHashSet();
ClassSet top;
ClassSet current;
- public RemapClasses(ClassSet set) {
- super();
- this.current = set;
+
+ public RemapClasses(final ClassSet set) {
+ current = set;
ClassSet top = set;
while (top.parent != null) {
top = top.parent;
@@ -283,11 +278,9 @@ class MergeAdapter extends ClassVisitor {
}
@Override
- public String map(String typeName) {
-
+ public String map(final String typeName) {
// remap the names of all classes that start with the old class name.
if (typeName.startsWith(top.precompiled.slash)) {
-
// write down all the sub classes.
if (typeName.startsWith(current.precompiled.slash + "$")) {
innerClasses.add(typeName);
@@ -295,45 +288,12 @@ class MergeAdapter extends ClassVisitor {
return typeName.replace(top.precompiled.slash, top.generated.slash);
}
+
return typeName;
}
public Set<String> getInnerClasses() {
return innerClasses;
}
-
- }
-
- private static final void check(ClassNode node) {
- Exception e = null;
- String error = "";
-
- try {
- ClassWriter cw = new ClassWriter(ClassWriter.COMPUTE_FRAMES);
- ClassVisitor cv = new DrillCheckClassAdapter(CompilationConfig.ASM_API_VERSION, cw, true);
- node.accept(cv);
-
- StringWriter sw = new StringWriter();
- PrintWriter pw = new PrintWriter(sw);
- DrillCheckClassAdapter.verify(new ClassReader(cw.toByteArray()), false, pw);
-
- error = sw.toString();
- } catch (Exception ex) {
- e = ex;
- }
-
- if (!error.isEmpty() || e != null) {
- StringWriter sw2 = new StringWriter();
- PrintWriter pw2 = new PrintWriter(sw2);
- TraceClassVisitor v = new TraceClassVisitor(pw2);
- node.accept(v);
- if (e != null) {
- throw new RuntimeException("Failure validating class. ByteCode: \n" +
- sw2.toString() + "\n\n====ERRROR====\n" + error, e);
- } else {
- throw new RuntimeException("Failure validating class. ByteCode: \n" +
- sw2.toString() + "\n\n====ERRROR====\n" + error);
- }
- }
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/7af5f9a0/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestBugFixes.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestBugFixes.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestBugFixes.java
new file mode 100644
index 0000000..c316484
--- /dev/null
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/TestBugFixes.java
@@ -0,0 +1,84 @@
+/**
+ * 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.drill.jdbc.test;
+
+import org.junit.Test;
+
+public class TestBugFixes extends JdbcTestQueryBase {
+// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestBugFixes.class);
+
+ @Test
+ public void testDrill2503() throws Exception {
+ final StringBuilder sb = new StringBuilder();
+ sb.append(
+ "SELECT "
+ + " CASE "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " WHEN 'y' = 'x' THEN 0 "
+ + " END "
+ + " FROM INFORMATION_SCHEMA.CATALOGS "
+ );
+
+ testQuery(sb.toString());
+ }
+}
[3/5] drill git commit: DRILL-2353: Add interpreter based partition
pruning.
Posted by ja...@apache.org.
DRILL-2353: Add interpreter based partition pruning.
Integrate Jacques's interpreter based partition pruning with Jason's interpreter refactoring that removed interpreter module and added that functionality in the exec module.
Ensure boolean operators are correctly handled when traversing expression tree to find partition filters.
Resolve merge conflicts after rebasing to master branch.
Additional fixes for handling OR conditions.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/48c9c01d
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/48c9c01d
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/48c9c01d
Branch: refs/heads/master
Commit: 48c9c01df011d63fa2c118624fe91c085bbe1883
Parents: 7af5f9a
Author: Aman Sinha <as...@mapr.com>
Authored: Sun Feb 15 23:04:50 2015 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Thu Mar 19 22:14:57 2015 -0700
----------------------------------------------------------------------
.../planner/sql/HivePartitionDescriptor.java | 6 +
.../org/apache/drill/exec/expr/TestPrune.java | 38 +++
.../fn/interpreter/InterpreterEvaluator.java | 11 +
.../planner/FileSystemPartitionDescriptor.java | 22 +-
.../drill/exec/planner/PartitionDescriptor.java | 7 +
.../exec/planner/logical/DrillRuleSets.java | 8 +-
.../partition/FindPartitionConditions.java | 302 +++++++++++++++++++
.../logical/partition/PruneScanRule.java | 296 ++++++++++++++++++
.../org/apache/drill/exec/expr/TestPrune.java | 47 +++
.../exec/planner/logical/FilterSplitTest.java | 170 +++++++++++
10 files changed, 904 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
index e6ca21e..8307dff 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
@@ -53,4 +53,10 @@ public class HivePartitionDescriptor implements PartitionDescriptor {
public int getMaxHierarchyLevel() {
return MAX_NESTED_SUBDIRS;
}
+
+ @Override
+ public Integer getIdIfValid(String name) {
+ return partitionMap.get(name);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/interpreter/src/test/java/org/apache/drill/exec/expr/TestPrune.java
----------------------------------------------------------------------
diff --git a/exec/interpreter/src/test/java/org/apache/drill/exec/expr/TestPrune.java b/exec/interpreter/src/test/java/org/apache/drill/exec/expr/TestPrune.java
new file mode 100644
index 0000000..7e75165
--- /dev/null
+++ b/exec/interpreter/src/test/java/org/apache/drill/exec/expr/TestPrune.java
@@ -0,0 +1,38 @@
+/*******************************************************************************
+ * 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.drill.exec.expr;
+
+import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.util.TestTools;
+import org.junit.Test;
+
+public class TestPrune extends BaseTestQuery {
+
+ String MULTILEVEL = TestTools.getWorkingPath() + "/../java-exec/src/test/resources/multilevel";
+
+ @Test
+ public void pruneCompound() throws Exception {
+ test(String.format("select * from dfs.`%s/csv` where x is null and dir1 in ('Q1', 'Q2')", MULTILEVEL));
+ }
+
+ @Test
+ public void pruneSimple() throws Exception {
+ test(String.format("select * from dfs.`%s/csv` where dir1 in ('Q1', 'Q2')", MULTILEVEL));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
index 4f8e126..35c35ec 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/interpreter/InterpreterEvaluator.java
@@ -19,6 +19,9 @@ package org.apache.drill.exec.expr.fn.interpreter;
import com.google.common.base.Preconditions;
import io.netty.buffer.DrillBuf;
+
+import java.lang.reflect.Field;
+
import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.expression.BooleanOperator;
import org.apache.drill.common.expression.CastExpression;
@@ -31,6 +34,10 @@ import org.apache.drill.common.expression.NullExpression;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.expression.TypedNullConstant;
import org.apache.drill.common.expression.ValueExpressions;
+import org.apache.drill.common.expression.ValueExpressions.BooleanExpression;
+import org.apache.drill.common.expression.ValueExpressions.DateExpression;
+import org.apache.drill.common.expression.ValueExpressions.TimeExpression;
+import org.apache.drill.common.expression.ValueExpressions.TimeStampExpression;
import org.apache.drill.common.expression.visitors.AbstractExprVisitor;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.exec.expr.DrillFuncHolderExpr;
@@ -42,7 +49,11 @@ import org.apache.drill.exec.expr.annotations.Output;
import org.apache.drill.exec.expr.annotations.Param;
import org.apache.drill.exec.expr.fn.DrillSimpleFuncHolder;
import org.apache.drill.exec.expr.holders.BitHolder;
+import org.apache.drill.exec.expr.holders.DateHolder;
+import org.apache.drill.exec.expr.holders.NullableBigIntHolder;
import org.apache.drill.exec.expr.holders.NullableBitHolder;
+import org.apache.drill.exec.expr.holders.TimeHolder;
+import org.apache.drill.exec.expr.holders.TimeStampHolder;
import org.apache.drill.exec.expr.holders.ValueHolder;
import org.apache.drill.exec.ops.QueryDateTimeInfo;
import org.apache.drill.exec.ops.UdfUtilities;
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/main/java/org/apache/drill/exec/planner/FileSystemPartitionDescriptor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/FileSystemPartitionDescriptor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/FileSystemPartitionDescriptor.java
index 4c1f8e8..9ad14b1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/FileSystemPartitionDescriptor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/FileSystemPartitionDescriptor.java
@@ -17,6 +17,11 @@
*/
package org.apache.drill.exec.planner;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+
+
// partition descriptor for file system based tables
public class FileSystemPartitionDescriptor implements PartitionDescriptor {
@@ -24,10 +29,14 @@ public class FileSystemPartitionDescriptor implements PartitionDescriptor {
private final String partitionLabel;
private final int partitionLabelLength;
+ private final Map<String, Integer> partitions = Maps.newHashMap();
public FileSystemPartitionDescriptor(String partitionLabel) {
this.partitionLabel = partitionLabel;
this.partitionLabelLength = partitionLabel.length();
+ for(int i =0; i < 10; i++){
+ partitions.put(partitionLabel + i, i);
+ }
}
@Override
@@ -38,11 +47,22 @@ public class FileSystemPartitionDescriptor implements PartitionDescriptor {
@Override
public boolean isPartitionName(String name) {
- return name.matches(partitionLabel +"[0-9]");
+ return partitions.containsKey(name);
+ }
+
+ @Override
+ public Integer getIdIfValid(String name) {
+ return partitions.get(name);
}
@Override
public int getMaxHierarchyLevel() {
return MAX_NESTED_SUBDIRS;
}
+
+ public String getName(int index){
+ return partitionLabel + index;
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PartitionDescriptor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PartitionDescriptor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PartitionDescriptor.java
index 02a6a8f..35fdae9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PartitionDescriptor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PartitionDescriptor.java
@@ -31,6 +31,13 @@ public interface PartitionDescriptor {
// Given a column name return boolean to indicate if its a partition column or not
public boolean isPartitionName(String name);
+ /**
+ * Check to see if the name is a partition name.
+ * @param name The field name you want to compare to partition names.
+ * @return Return index if valid, otherwise return null;
+ */
+ public Integer getIdIfValid(String name);
+
// Maximum level of partition nesting/ hierarchy supported
public int getMaxHierarchyLevel();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
index 496bc9a..b1a7189 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
@@ -24,6 +24,7 @@ import java.util.List;
import net.hydromatic.optiq.tools.RuleSet;
import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.planner.logical.partition.PruneScanRule;
import org.apache.drill.exec.planner.physical.ConvertCountToDirectScan;
import org.apache.drill.exec.planner.physical.FilterPrule;
import org.apache.drill.exec.planner.physical.HashAggPrule;
@@ -103,8 +104,11 @@ public class DrillRuleSets {
// PushSortPastProjectRule.INSTANCE, //
DrillPushProjIntoScan.INSTANCE,
- DrillPushPartitionFilterIntoScan.FILTER_ON_PROJECT,
- DrillPushPartitionFilterIntoScan.FILTER_ON_SCAN,
+
+// DrillPushPartitionFilterIntoScan.FILTER_ON_PROJECT,
+// DrillPushPartitionFilterIntoScan.FILTER_ON_SCAN,
+ PruneScanRule.getFilterOnProject(context),
+ PruneScanRule.getFilterOnScan(context),
////////////////////////////////
DrillScanRule.INSTANCE,
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/FindPartitionConditions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/FindPartitionConditions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/FindPartitionConditions.java
new file mode 100644
index 0000000..3acf29d
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/FindPartitionConditions.java
@@ -0,0 +1,302 @@
+/**
+ * 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.drill.exec.planner.logical.partition;
+
+import java.util.ArrayDeque;
+import java.util.BitSet;
+import java.util.Deque;
+import java.util.List;
+import java.util.Iterator;
+
+import org.eigenbase.rex.RexBuilder;
+import org.eigenbase.rex.RexCall;
+import org.eigenbase.rex.RexCorrelVariable;
+import org.eigenbase.rex.RexDynamicParam;
+import org.eigenbase.rex.RexFieldAccess;
+import org.eigenbase.rex.RexInputRef;
+import org.eigenbase.rex.RexLiteral;
+import org.eigenbase.rex.RexNode;
+import org.eigenbase.rex.RexOver;
+import org.eigenbase.rex.RexRangeRef;
+import org.eigenbase.rex.RexVisitorImpl;
+import org.eigenbase.sql.SqlKind;
+import org.eigenbase.sql.SqlOperator;
+import org.eigenbase.sql.fun.SqlRowOperator;
+import org.eigenbase.sql.fun.SqlStdOperatorTable;
+import org.eigenbase.util.Stacks;
+import org.eigenbase.util.Util;
+
+import com.google.common.collect.Lists;
+
+
+public class FindPartitionConditions extends RexVisitorImpl<Void> {
+ /** Whether an expression is a directory filter, and if so, whether
+ * it can be pushed into the scan.
+ */
+ enum PushDirFilter {
+ NO_PUSH, PUSH
+ }
+
+ /**
+ * During top-down traversal of the expression tree, keep track of the
+ * boolean operators such that if a directory filter is found, it will
+ * be added as a child of the current boolean operator.
+ *
+ * NOTE: this auxiliary class is necessary because RexNodes are immutable.
+ * If they were mutable, we could have easily added/dropped inputs as we
+ * encountered directory filters.
+ */
+ public class BooleanOpState {
+ private SqlOperator booleanOp;
+ private List<RexNode> children = Lists.newArrayList();
+ public BooleanOpState(SqlOperator op) {
+ booleanOp = op;
+ }
+ public SqlOperator getOp() {
+ return booleanOp;
+ }
+ public void addChild(RexNode n) {
+ children.add(n);
+ }
+ public List<RexNode> getChildren() {
+ return children;
+ }
+ public void clear() {
+ children.clear();
+ }
+ }
+
+ private final BitSet dirs;
+
+ private final List<PushDirFilter> pushStatusStack = Lists.newArrayList();
+ private final Deque<SqlOperator> parentCallTypeStack = new ArrayDeque<SqlOperator>();
+ private final Deque<BooleanOpState> opStack = new ArrayDeque<BooleanOpState>();
+
+ private RexBuilder builder = null;
+ private RexNode resultCondition = null;
+
+ public FindPartitionConditions(BitSet dirs) {
+ // go deep
+ super(true);
+ this.dirs = dirs;
+ }
+
+ public FindPartitionConditions(BitSet dirs, RexBuilder builder) {
+ // go deep
+ super(true);
+ this.dirs = dirs;
+ this.builder = builder;
+ }
+
+ public void analyze(RexNode exp) {
+ assert pushStatusStack.isEmpty();
+
+ exp.accept(this);
+
+ // Deal with top of stack
+ assert pushStatusStack.size() == 1;
+ assert parentCallTypeStack.isEmpty();
+ PushDirFilter rootPushDirFilter = pushStatusStack.get(0);
+ if (rootPushDirFilter == PushDirFilter.PUSH) {
+ // The entire subtree was directory filter, so add it to the result.
+ addResult(exp);
+ }
+ pushStatusStack.clear();
+ }
+
+ public RexNode getFinalCondition() {
+ return resultCondition;
+ }
+
+ private Void pushVariable() {
+ pushStatusStack.add(PushDirFilter.NO_PUSH);
+ return null;
+ }
+
+ private void addResult(RexNode exp) {
+ // when we find a directory filter, add it to the current boolean operator's
+ // children (if one exists)
+ if (!opStack.isEmpty()) {
+ BooleanOpState op = opStack.peek();
+ op.addChild(exp);
+ } else {
+ resultCondition = exp;
+ }
+ }
+
+ /**
+ * For an OR node that is marked as NO_PUSH, there could be 3 situations:
+ * 1. left child has a partition condition, right child does not. In this case, we should not push any child of this OR
+ * 2. left child does not have partition condition, right child has one. Again, we should not push any child of this OR
+ * 3. left and right child both have partition condition but both sides may have had other non-partition conditions. In
+ * this case, we can push the partition conditions by building a new OR combining both children.
+ * In this method we clear the children of the OR for cases 1 and 2 and leave it alone for case 3
+ */
+ private void clearOrChildrenIfSingle() {
+ if (!opStack.isEmpty()) {
+ BooleanOpState op = opStack.peek();
+ assert op.getOp().getKind() == SqlKind.OR;
+ if (op.getChildren().size() == 1) {
+ op.clear();
+ }
+ }
+ }
+
+ /**
+ * If the top of the parentCallTypeStack is an AND or OR, get the corresponding
+ * top item from the BooleanOpState stack and examine its children - these must
+ * be the directory filters we are interested in. Create a new filter condition
+ * using the boolean operation and the children. Add this new filter as a child
+ * of the parent boolean operator - thus the filter condition gets built bottom-up.
+ */
+ private void popAndBuildFilter() {
+ SqlOperator op1 = null;
+ if (!parentCallTypeStack.isEmpty()) {
+ op1 = parentCallTypeStack.pop();
+ }
+ if (op1 != null
+ && (op1.getKind() == SqlKind.AND || op1.getKind() == SqlKind.OR)
+ && !opStack.isEmpty()) {
+ BooleanOpState op = opStack.pop();
+ int size = op.getChildren().size();
+ RexNode newFilter = null;
+ if (size > 1) {
+ newFilter = builder.makeCall(op.getOp(), op.getChildren());
+ } else if (size == 1) {
+ newFilter = op.getChildren().get(0);
+ }
+ if (newFilter != null) {
+ // add this new filter to my parent boolean operator's children
+ if (!opStack.isEmpty()) {
+ op = opStack.peek();
+ op.addChild(newFilter);
+ } else {
+ resultCondition = newFilter;
+ }
+ }
+ }
+ }
+
+
+ public Void visitInputRef(RexInputRef inputRef) {
+ if(dirs.get(inputRef.getIndex())){
+ pushStatusStack.add(PushDirFilter.PUSH);
+ }else{
+ pushStatusStack.add(PushDirFilter.NO_PUSH);
+ }
+ return null;
+ }
+
+ public Void visitLiteral(RexLiteral literal) {
+ pushStatusStack.add(PushDirFilter.PUSH);
+ return null;
+ }
+
+ public Void visitOver(RexOver over) {
+ // assume NO_PUSH until proven otherwise
+ analyzeCall(over, PushDirFilter.NO_PUSH);
+ return null;
+ }
+
+ public Void visitCorrelVariable(RexCorrelVariable correlVariable) {
+ return pushVariable();
+ }
+
+ public Void visitCall(RexCall call) {
+ // assume PUSH until proven otherwise
+ analyzeCall(call, PushDirFilter.PUSH);
+ return null;
+ }
+
+ private void analyzeCall(RexCall call, PushDirFilter callPushDirFilter) {
+ parentCallTypeStack.push(call.getOperator());
+ if (call.getKind() == SqlKind.AND || call.getKind() == SqlKind.OR) {
+ opStack.push(new BooleanOpState(call.getOperator()));
+ }
+
+ // visit operands, pushing their states onto stack
+ super.visitCall(call);
+
+ // look for NO_PUSH operands
+ int operandCount = call.getOperands().size();
+ List<PushDirFilter> operandStack = Util.last(pushStatusStack, operandCount);
+ for (PushDirFilter operandPushDirFilter : operandStack) {
+ if (operandPushDirFilter == PushDirFilter.NO_PUSH) {
+ callPushDirFilter = PushDirFilter.NO_PUSH;
+ }
+ }
+
+ // Even if all operands are PUSH, the call itself may
+ // be non-deterministic.
+ if (!call.getOperator().isDeterministic()) {
+ callPushDirFilter = PushDirFilter.NO_PUSH;
+ } else if (call.getOperator().isDynamicFunction()) {
+ // For now, treat it same as non-deterministic.
+ callPushDirFilter = PushDirFilter.NO_PUSH;
+ }
+
+ // Row operator itself can't be reduced to a PUSH
+ if ((callPushDirFilter == PushDirFilter.PUSH)
+ && (call.getOperator() instanceof SqlRowOperator)) {
+ callPushDirFilter = PushDirFilter.NO_PUSH;
+ }
+
+
+ if (callPushDirFilter == PushDirFilter.NO_PUSH) {
+ if (call.getKind() == SqlKind.AND) {
+ // one or more children is not a push-able directory filter. If this is an AND, add
+ // all the ones that are push-able directory filters.
+ for (int iOperand = 0; iOperand < operandCount; ++iOperand) {
+ PushDirFilter pushDirFilter = operandStack.get(iOperand);
+ RexNode n = call.getOperands().get(iOperand);
+ if (pushDirFilter == PushDirFilter.PUSH && !(n.getKind() == SqlKind.AND || n.getKind() == SqlKind.OR)) {
+ addResult(n);
+ }
+ }
+ } else if (call.getKind() == SqlKind.OR) {
+ clearOrChildrenIfSingle();
+ }
+ }
+ else if (callPushDirFilter == PushDirFilter.PUSH && !(call.getKind() == SqlKind.AND || call.getKind() == SqlKind.OR)) {
+ addResult(call);
+ }
+
+ // pop operands off of the stack
+ operandStack.clear();
+
+ // pop this parent call operator off the stack and build the intermediate filters as we go
+ popAndBuildFilter();
+
+ // push PushDirFilter result for this call onto stack
+ pushStatusStack.add(callPushDirFilter);
+ }
+
+ public Void visitDynamicParam(RexDynamicParam dynamicParam) {
+ return pushVariable();
+ }
+
+ public Void visitRangeRef(RexRangeRef rangeRef) {
+ return pushVariable();
+ }
+
+ public Void visitFieldAccess(RexFieldAccess fieldAccess) {
+ return pushVariable();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java
new file mode 100644
index 0000000..b8c9ebf
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java
@@ -0,0 +1,296 @@
+/**
+ * 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.drill.exec.planner.logical.partition;
+
+import java.util.BitSet;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import net.hydromatic.optiq.util.BitSets;
+
+import org.apache.drill.common.expression.ErrorCollectorImpl;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.common.types.Types;
+import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
+import org.apache.drill.exec.expr.fn.interpreter.InterpreterEvaluator;
+import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.FileGroupScan;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.planner.FileSystemPartitionDescriptor;
+import org.apache.drill.exec.planner.logical.DrillFilterRel;
+import org.apache.drill.exec.planner.logical.DrillOptiq;
+import org.apache.drill.exec.planner.logical.DrillParseContext;
+import org.apache.drill.exec.planner.logical.DrillProjectRel;
+import org.apache.drill.exec.planner.logical.DrillRel;
+import org.apache.drill.exec.planner.logical.DrillScanRel;
+import org.apache.drill.exec.planner.logical.RelOptHelper;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.store.dfs.FileSelection;
+import org.apache.drill.exec.store.dfs.FormatSelection;
+import org.apache.drill.exec.vector.NullableBitVector;
+import org.apache.drill.exec.vector.NullableVarCharVector;
+import org.eigenbase.rel.RelNode;
+import org.eigenbase.relopt.RelOptRule;
+import org.eigenbase.relopt.RelOptRuleCall;
+import org.eigenbase.relopt.RelOptRuleOperand;
+import org.eigenbase.relopt.RelOptUtil;
+import org.eigenbase.rex.RexNode;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+public abstract class PruneScanRule extends RelOptRule {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PruneScanRule.class);
+
+ public static final RelOptRule getFilterOnProject(QueryContext context){
+ return new PruneScanRule(
+ RelOptHelper.some(DrillFilterRel.class, RelOptHelper.some(DrillProjectRel.class, RelOptHelper.any(DrillScanRel.class))),
+ "PruneScanRule:Filter_On_Project",
+ context) {
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final DrillScanRel scan = (DrillScanRel) call.rel(2);
+ GroupScan groupScan = scan.getGroupScan();
+ // this rule is applicable only for dfs based partition pruning
+ return groupScan instanceof FileGroupScan && groupScan.supportsPartitionFilterPushdown();
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
+ final DrillProjectRel projectRel = (DrillProjectRel) call.rel(1);
+ final DrillScanRel scanRel = (DrillScanRel) call.rel(2);
+ doOnMatch(call, filterRel, projectRel, scanRel);
+ };
+ };
+ }
+
+ public static final RelOptRule getFilterOnScan(QueryContext context){
+ return new PruneScanRule(
+ RelOptHelper.some(DrillFilterRel.class, RelOptHelper.any(DrillScanRel.class)),
+ "PruneScanRule:Filter_On_Scan", context) {
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final DrillScanRel scan = (DrillScanRel) call.rel(1);
+ GroupScan groupScan = scan.getGroupScan();
+ // this rule is applicable only for dfs based partition pruning
+ return groupScan instanceof FileGroupScan && groupScan.supportsPartitionFilterPushdown();
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
+ final DrillScanRel scanRel = (DrillScanRel) call.rel(1);
+ doOnMatch(call, filterRel, null, scanRel);
+ }
+ };
+ }
+
+ final QueryContext context;
+
+ private PruneScanRule(RelOptRuleOperand operand, String id, QueryContext context) {
+ super(operand, id);
+ this.context = context;
+ }
+
+ protected void doOnMatch(RelOptRuleCall call, DrillFilterRel filterRel, DrillProjectRel projectRel, DrillScanRel scanRel) {
+ PlannerSettings settings = context.getPlannerSettings();
+ FileSystemPartitionDescriptor descriptor = new FileSystemPartitionDescriptor(settings.getFsPartitionColumnLabel());
+ final BufferAllocator allocator = context.getAllocator();
+
+
+ RexNode condition = null;
+ if(projectRel == null){
+ condition = filterRel.getCondition();
+ }else{
+ // get the filter as if it were below the projection.
+ condition = RelOptUtil.pushFilterPastProject(filterRel.getCondition(), projectRel);
+ }
+
+ Map<Integer, String> dirNames = Maps.newHashMap();
+ List<String> fieldNames = scanRel.getRowType().getFieldNames();
+ BitSet columnBitset = new BitSet();
+ BitSet dirBitset = new BitSet();
+ {
+ int colIndex = 0;
+ for(String field : fieldNames){
+ final Integer dirIndex = descriptor.getIdIfValid(field);
+ if(dirIndex != null){
+ dirNames.put(dirIndex, field);
+ dirBitset.set(dirIndex);
+ columnBitset.set(colIndex);
+ }
+ colIndex++;
+ }
+ }
+
+ if(dirBitset.isEmpty()){
+ return;
+ }
+
+ FindPartitionConditions c = new FindPartitionConditions(columnBitset, filterRel.getCluster().getRexBuilder());
+ c.analyze(condition);
+ RexNode pruneCondition = c.getFinalCondition();
+
+ if(pruneCondition == null){
+ return;
+ }
+
+ // set up the partitions
+ final FormatSelection origSelection = (FormatSelection)scanRel.getDrillTable().getSelection();
+ final List<String> files = origSelection.getAsFiles();
+ final String selectionRoot = origSelection.getSelection().selectionRoot;
+ List<PathPartition> partitions = Lists.newLinkedList();
+
+ // let's only deal with one batch of files for now.
+ if(files.size() > Character.MAX_VALUE){
+ return;
+ }
+
+ for(String f : files){
+ partitions.add(new PathPartition(descriptor.getMaxHierarchyLevel(), selectionRoot, f));
+ }
+
+ final NullableBitVector output = new NullableBitVector(MaterializedField.create("", Types.optional(MinorType.BIT)), allocator);
+ final VectorContainer container = new VectorContainer();
+
+ try{
+ final NullableVarCharVector[] vectors = new NullableVarCharVector[descriptor.getMaxHierarchyLevel()];
+ for(int dirIndex : BitSets.toIter(dirBitset)){
+ NullableVarCharVector vector = new NullableVarCharVector(MaterializedField.create(dirNames.get(dirIndex), Types.optional(MinorType.VARCHAR)), allocator);
+ vector.allocateNew(5000, partitions.size());
+ vectors[dirIndex] = vector;
+ container.add(vector);
+ }
+
+ // populate partition vectors.
+ int record = 0;
+ for(Iterator<PathPartition> iter = partitions.iterator(); iter.hasNext(); record++){
+ final PathPartition partition = iter.next();
+ for(int dirIndex : BitSets.toIter(dirBitset)){
+ if(partition.dirs[dirIndex] == null){
+ vectors[dirIndex].getMutator().setNull(record);
+ }else{
+ byte[] bytes = partition.dirs[dirIndex].getBytes(Charsets.UTF_8);
+ vectors[dirIndex].getMutator().setSafe(record, bytes, 0, bytes.length);
+ }
+ }
+ }
+
+ for(NullableVarCharVector v : vectors){
+ if(v == null){
+ continue;
+ }
+ v.getMutator().setValueCount(partitions.size());
+ }
+
+
+ // materialize the expression
+ logger.debug("Attempting to prune {}", pruneCondition);
+ LogicalExpression expr = DrillOptiq.toDrill(new DrillParseContext(), scanRel, pruneCondition);
+ ErrorCollectorImpl errors = new ErrorCollectorImpl();
+ LogicalExpression materializedExpr = ExpressionTreeMaterializer.materialize(expr, container, errors, context.getFunctionRegistry());
+ if (errors.getErrorCount() != 0) {
+ logger.warn("Failure while materializing expression [{}]. Errors: {}", expr, errors);
+ }
+
+ output.allocateNew(partitions.size());
+ InterpreterEvaluator.evaluate(partitions.size(), context, container, output, materializedExpr);
+ record = 0;
+
+ List<String> newFiles = Lists.newArrayList();
+ for(Iterator<PathPartition> iter = partitions.iterator(); iter.hasNext(); record++){
+ PathPartition part = iter.next();
+ if(!output.getAccessor().isNull(record) && output.getAccessor().get(record) == 1){
+ newFiles.add(part.file);
+ }
+ }
+
+ if(newFiles.isEmpty()){
+ newFiles.add(files.get(0));
+ }
+
+ if(newFiles.size() == files.size()){
+ return;
+ }
+
+ logger.debug("Pruned {} => {}", files, newFiles);
+
+
+ final FileSelection newFileSelection = new FileSelection(newFiles, origSelection.getSelection().selectionRoot, true);
+ final FileGroupScan newScan = ((FileGroupScan)scanRel.getGroupScan()).clone(newFileSelection);
+ final DrillScanRel newScanRel =
+ new DrillScanRel(scanRel.getCluster(),
+ scanRel.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
+ scanRel.getTable(),
+ newScan,
+ scanRel.getRowType(),
+ scanRel.getColumns());
+
+ RelNode inputRel = newScanRel;
+
+ if(projectRel != null){
+ inputRel = projectRel.copy(projectRel.getTraitSet(), Collections.singletonList(inputRel));
+ }
+
+ final RelNode newFilter = filterRel.copy(filterRel.getTraitSet(), Collections.singletonList(inputRel));
+ call.transformTo(newFilter);
+
+ }catch(Exception e){
+ logger.warn("Exception while trying to prune partition.", e);
+ }finally{
+ container.clear();
+ if(output !=null){
+ output.clear();
+ }
+ }
+ }
+
+ private static class PathPartition {
+ final String[] dirs;
+ final String file;
+
+ public PathPartition(int max, String selectionRoot, String file){
+ this.file = file;
+ int start = file.indexOf(selectionRoot) + selectionRoot.length();
+ String postPath = file.substring(start);
+ if(postPath.charAt(0) == '/'){
+ postPath = postPath.substring(1);
+ }
+ String[] mostDirs = postPath.split("/");
+ this.dirs = new String[max];
+ int maxLoop = Math.min(max, mostDirs.length - 1);
+ for(int i =0; i < maxLoop; i++){
+ this.dirs[i] = mostDirs[i];
+ }
+ }
+
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/test/java/org/apache/drill/exec/expr/TestPrune.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/expr/TestPrune.java b/exec/java-exec/src/test/java/org/apache/drill/exec/expr/TestPrune.java
new file mode 100644
index 0000000..d15555e
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/expr/TestPrune.java
@@ -0,0 +1,47 @@
+/*******************************************************************************
+ * 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.drill.exec.expr;
+
+import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.util.TestTools;
+import org.junit.Test;
+
+public class TestPrune extends BaseTestQuery {
+
+ String MULTILEVEL = TestTools.getWorkingPath() + "/../java-exec/src/test/resources/multilevel";
+
+ @Test
+ public void pruneCompound1() throws Exception {
+ test(String.format("select * from dfs.`%s/csv` where x is null and dir1 in ('Q1', 'Q2')", MULTILEVEL));
+ }
+
+ @Test
+ public void pruneSimple1() throws Exception {
+ test(String.format("select * from dfs.`%s/csv` where dir1 in ('Q1', 'Q2')", MULTILEVEL));
+ }
+
+ @Test
+ public void pruneCompound2() throws Exception {
+ String query1 = String.format("select * from dfs_test.`%s/parquet` where (dir0=1995 and o_totalprice < 40000) or (dir0=1996 and o_totalprice < 40000)", MULTILEVEL);
+ String query2 = String.format("select * from dfs_test.`%s/parquet` where dir0=1995 and o_totalprice < 40000", MULTILEVEL);
+ String query3 = String.format("select * from dfs_test.`%s/parquet` where (dir0=1995 and o_totalprice < 40000) or dir0=1996", MULTILEVEL);
+ String query4 = String.format("select * from dfs_test.`%s/parquet` where dir0=1995 or dir0=1996", MULTILEVEL);
+ test(query3);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/48c9c01d/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/FilterSplitTest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/FilterSplitTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/FilterSplitTest.java
new file mode 100644
index 0000000..7c85c19
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/FilterSplitTest.java
@@ -0,0 +1,170 @@
+/**
+ * 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.drill.exec.planner.logical;
+
+import static org.junit.Assert.*;
+
+import java.util.BitSet;
+
+import net.hydromatic.optiq.impl.java.JavaTypeFactory;
+import net.hydromatic.optiq.jdbc.JavaTypeFactoryImpl;
+
+import org.apache.drill.exec.planner.logical.partition.FindPartitionConditions;
+import org.eigenbase.reltype.RelDataType;
+import org.eigenbase.rex.RexBuilder;
+import org.eigenbase.rex.RexNode;
+import org.eigenbase.sql.fun.SqlStdOperatorTable;
+import org.eigenbase.sql.type.SqlTypeName;
+import org.junit.Test;
+
+public class FilterSplitTest {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FilterSplitTest.class);
+
+ final JavaTypeFactory t = new JavaTypeFactoryImpl();
+ final RexBuilder builder = new RexBuilder(t);
+ final RelDataType intType = t.createSqlType(SqlTypeName.INTEGER);
+ final RelDataType sType = t.createSqlType(SqlTypeName.VARCHAR, 20);
+
+ @Test
+ public void simpleCompound() {
+ // a < 1 AND dir0 in (2,3)
+ RexNode n = and(
+ lt(c(0), lit(1)),
+ or(
+ eq(c(1), lit(2)),
+ eq(c(1), lit(3))
+ )
+ );
+
+ BitSet bs = new BitSet();
+ bs.set(1);
+ FindPartitionConditions c = new FindPartitionConditions(bs, builder);
+ c.analyze(n);
+
+ RexNode partNode = c.getFinalCondition();
+ assertEquals(n.toString(), "AND(<($0, 1), OR(=($1, 2), =($1, 3)))");
+ assertEquals(partNode.toString(), "OR(=($1, 2), =($1, 3))");
+ }
+
+ @Test
+ public void twoLevelDir() {
+ // (dir0 = 1 and dir1 = 2) OR (dir0 = 3 and dir1 = 4)
+ RexNode n = or(
+ and(
+ eq(c(1), lit(1)),
+ eq(c(2), lit(2))
+ ),
+ and(
+ eq(c(1), lit(3)),
+ eq(c(2), lit(4))
+ )
+
+ );
+
+ BitSet bs = new BitSet();
+ bs.set(1);
+ bs.set(2);
+ FindPartitionConditions c = new FindPartitionConditions(bs, builder);
+ c.analyze(n);
+
+ RexNode partNode = c.getFinalCondition();
+ assertEquals("OR(AND(=($1, 1), =($2, 2)), AND(=($1, 3), =($2, 4)))", n.toString());
+ assertEquals("OR(AND(=($1, 1), =($2, 2)), AND(=($1, 3), =($2, 4)))", partNode.toString());
+ }
+
+ @Test
+ public void badOr() {
+ // (dir0 = 1 and dir1 = 2) OR (a < 5)
+ RexNode n = or(
+ and(
+ eq(c(1), lit(1)),
+ eq(c(2), lit(2))
+ ),
+ lt(c(0), lit(5))
+
+ );
+
+ BitSet bs = new BitSet();
+ bs.set(1);
+ bs.set(2);
+ FindPartitionConditions c = new FindPartitionConditions(bs, builder);
+ c.analyze(n);
+
+ RexNode partNode = c.getFinalCondition();
+ assertEquals("OR(AND(=($1, 1), =($2, 2)), <($0, 5))", n.toString());
+ assertTrue(partNode == null);
+ }
+
+
+ @Test
+ public void badFunc() {
+ // (dir0 = 1 and dir1 = 2) OR (a < 5)
+ RexNode n = fn(
+ cs(0),
+ cs(1)
+ );
+
+ BitSet bs = new BitSet();
+ bs.set(1);
+ bs.set(2);
+ FindPartitionConditions c = new FindPartitionConditions(bs, builder);
+ c.analyze(n);
+
+ RexNode partNode = c.getFinalCondition();
+ assertEquals("||($0, $1)", n.toString());
+ assertTrue(partNode == null);
+ }
+
+
+ private RexNode and(RexNode...nodes){
+ return builder.makeCall(SqlStdOperatorTable.AND, nodes);
+ }
+
+ private RexNode fn(RexNode...nodes){
+ return builder.makeCall(SqlStdOperatorTable.CONCAT, nodes);
+ }
+
+ private RexNode or(RexNode...nodes){
+ return builder.makeCall(SqlStdOperatorTable.OR, nodes);
+ }
+
+ private RexNode lt(RexNode left, RexNode right){
+ return builder.makeCall(SqlStdOperatorTable.LESS_THAN, left, right);
+ }
+
+ private RexNode eq(RexNode left, RexNode right){
+ return builder.makeCall(SqlStdOperatorTable.EQUALS, left, right);
+ }
+
+ private RexNode lit(int value){
+ return builder.makeLiteral(value, intType, true);
+ }
+
+ private RexNode c(int index){
+ return builder.makeInputRef(intType, index);
+ }
+
+
+ private RexNode cs(int index){
+ return builder.makeInputRef(sType, index);
+ }
+
+ private RexNode str(String s){
+ return builder.makeLiteral(s);
+ }
+}
[5/5] drill git commit: DRILL-1735: Have closing of JDBC connection
free embedded-server resources.
Posted by ja...@apache.org.
DRILL-1735: Have closing of JDBC connection free embedded-server resources.
Hooked up closing of JDBC connection to shut down embedded Drillbit, and then
fixed chain of bugs exposed by that:
1. Added test org.apache.drill.jdbc.test.Bug1735ConnectionCloseTest.
2. Hooked up connection handler in Driver to actually close JDBC connection.
3. Released a QueryResultsBatch in DrillCursor.
4. Reset DrillMetrics in BootStrapContext.close() (so stopping local DrillBit
and starting new DrillBit doesn't yield "duplicate metric" error.)
5. Checked cursor/row state before trying to retrieve value in DrillResultSet's
column accessor methods.
- Added org.apache.drill.jdbc.JdbcApiSqlException (for JDBC-level errors).
- Added org.apache.drill.jdbc.InvalidCursorStateSqlException.
[SqlAccessor, AvaticaDrillSqlAccessor, DrillConnectionImpl, DrillCursor,
DrillResultSet, InvalidCursorStateSqlException, JdbcApiSqlException,
DrillResultSetTest]
6. Released vectors in DrillResultSet.cleanup().
Added org.apache.drill.jdbc.test.Bug1735ResultSetCloseReleasesBuffersTest.
7. Delayed last-chunk batch until COMPLETED batch in QueryResultHandler.
8. Added nextUntilEnd(...) workarounds for fragment cancelation race condition
to TestView and other JDBC module/subproject tests.
9. Tracked open statements in orer to close at connection close (DrillStatementRegistry, etc.)
10. Commented out nextUntilEnd(...) workarounds for fragment cancelation race
condition.
11 Miscellaneous:
- Added some toString() methods
- Adjusted some logging (e.g., "// log.debug(...)" -> "log.trace(...)".
- Cleaned up a bit. [DrillCursor, DrillResultSet, QueryResultHandler]
- Added a few documentation comments.
- Added various TODO comments.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/9c9ee8c4
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/9c9ee8c4
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/9c9ee8c4
Branch: refs/heads/master
Commit: 9c9ee8c435c19c90636ce770fef9d05b5d3ae12e
Parents: 48c9c01
Author: dbarclay <db...@maprtech.com>
Authored: Tue Dec 16 14:05:11 2014 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Thu Mar 19 22:14:57 2015 -0700
----------------------------------------------------------------------
.../apache/drill/exec/client/DrillClient.java | 31 +-
.../drill/exec/physical/impl/ScreenCreator.java | 10 +-
.../drill/exec/rpc/user/QueryResultHandler.java | 274 ++++++++++----
.../drill/exec/server/BootStrapContext.java | 4 +-
.../org/apache/drill/exec/server/Drillbit.java | 15 +-
.../drill/exec/vector/accessor/SqlAccessor.java | 11 +
.../apache/drill/exec/work/foreman/Foreman.java | 3 +-
.../exec/work/fragment/FragmentExecutor.java | 57 ++-
.../java/org/apache/drill/exec/ExecTest.java | 2 +
.../drill/jdbc/AvaticaDrillSqlAccessor.java | 62 ++--
.../apache/drill/jdbc/DrillAccessorList.java | 2 +
.../drill/jdbc/DrillConnectionConfig.java | 2 +
.../apache/drill/jdbc/DrillConnectionImpl.java | 40 +-
.../java/org/apache/drill/jdbc/DrillCursor.java | 79 +++-
.../drill/jdbc/DrillPreparedStatement.java | 6 +-
.../org/apache/drill/jdbc/DrillResultSet.java | 38 +-
.../org/apache/drill/jdbc/DrillStatement.java | 6 +-
.../drill/jdbc/DrillStatementRegistry.java | 57 ++-
.../main/java/org/apache/drill/jdbc/Driver.java | 4 +-
.../jdbc/InvalidCursorStateSqlException.java | 97 +++++
.../apache/drill/jdbc/JdbcApiSqlException.java | 155 ++++++++
.../apache/drill/jdbc/DrillResultSetTest.java | 159 ++++++++
.../java/org/apache/drill/jdbc/DriverTest.java | 371 +++++++++++++++++++
.../java/org/apache/drill/jdbc/JdbcTest.java | 21 +-
.../jdbc/test/Bug1735ConnectionCloseTest.java | 102 +++++
...ug1735ResultSetCloseReleasesBuffersTest.java | 89 +++++
.../drill/jdbc/test/TestJdbcDistQuery.java | 23 +-
.../apache/drill/jdbc/test/TestJdbcQuery.java | 15 +
.../apache/drill/jdbc/test/TestMetadataDDL.java | 41 +-
.../org/apache/drill/jdbc/test/TestViews.java | 127 +++++--
30 files changed, 1680 insertions(+), 223 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
index c3a873c..6d4c86c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
@@ -40,6 +40,7 @@ import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
import org.apache.drill.exec.proto.UserBitShared;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
import org.apache.drill.exec.proto.UserBitShared.QueryType;
import org.apache.drill.exec.proto.UserProtos;
import org.apache.drill.exec.proto.UserProtos.Property;
@@ -61,9 +62,10 @@ import com.google.common.util.concurrent.AbstractCheckedFuture;
import com.google.common.util.concurrent.SettableFuture;
/**
- * Thin wrapper around a UserClient that handles connect/close and transforms String into ByteBuf
+ * Thin wrapper around a UserClient that handles connect/close and transforms
+ * String into ByteBuf.
*/
-public class DrillClient implements Closeable, ConnectionThrottle{
+public class DrillClient implements Closeable, ConnectionThrottle {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillClient.class);
DrillConfig config;
@@ -223,7 +225,7 @@ public class DrillClient implements Closeable, ConnectionThrottle{
if (this.ownsAllocator && allocator != null) {
allocator.close();
}
- if(ownsZkConnection) {
+ if (ownsZkConnection) {
try {
this.clusterCoordinator.close();
} catch (IOException e) {
@@ -234,6 +236,7 @@ public class DrillClient implements Closeable, ConnectionThrottle{
eventLoopGroup.shutdownGracefully();
}
+ // TODO: Did DRILL-1735 changes cover this TODO?:
// TODO: fix tests that fail when this is called.
//allocator.close();
connected = false;
@@ -247,9 +250,9 @@ public class DrillClient implements Closeable, ConnectionThrottle{
* @throws RpcException
*/
public List<QueryResultBatch> runQuery(QueryType type, String plan) throws RpcException {
- UserProtos.RunQuery query = newBuilder().setResultsMode(STREAM_FULL).setType(type).setPlan(plan).build() ;
+ UserProtos.RunQuery query = newBuilder().setResultsMode(STREAM_FULL).setType(type).setPlan(plan).build();
ListHoldingResultsListener listener = new ListHoldingResultsListener(query);
- client.submitQuery(listener,query);
+ client.submitQuery(listener, query);
return listener.getResults();
}
@@ -282,7 +285,9 @@ public class DrillClient implements Closeable, ConnectionThrottle{
/**
* Submits a Logical plan for direct execution (bypasses parsing)
*
- * @param plan the plan to execute
+ * @param plan the plan to execute
+ * @return a handle for the query result
+ * @throws RpcException
*/
public void runQuery(QueryType type, String plan, UserResultsListener resultsListener) {
client.submitQuery(resultsListener, newBuilder().setResultsMode(STREAM_FULL).setType(type).setPlan(plan).build());
@@ -294,6 +299,7 @@ public class DrillClient implements Closeable, ConnectionThrottle{
private UserProtos.RunQuery query ;
public ListHoldingResultsListener(UserProtos.RunQuery query) {
+ logger.debug( "Listener created for query \"\"\"{}\"\"\"", query );
this.query = query;
}
@@ -323,11 +329,20 @@ public class DrillClient implements Closeable, ConnectionThrottle{
@Override
public void resultArrived(QueryResultBatch result, ConnectionThrottle throttle) {
-// logger.debug("Result arrived. Is Last Chunk: {}. Full Result: {}", result.getHeader().getIsLastChunk(), result);
+ logger.debug(
+ "Result arrived: Query state: {}. Is last chunk: {}. Result: {}",
+ result.getHeader().getQueryState(),
+ result.getHeader().getIsLastChunk(),
+ result );
results.add(result);
if (result.getHeader().getIsLastChunk()) {
future.set(results);
}
+ else {
+ assert QueryState.PENDING == result.getHeader().getQueryState()
+ : "For non-last chunk, expected query state of PENDING but got "
+ + result.getHeader().getQueryState();
+ }
}
public List<QueryResultBatch> getResults() throws RpcException{
@@ -340,7 +355,9 @@ public class DrillClient implements Closeable, ConnectionThrottle{
@Override
public void queryIdArrived(QueryId queryId) {
+ logger.debug( "Query ID arrived: {}", queryId );
}
+
}
private class FutureHandler extends AbstractCheckedFuture<Void, RpcException> implements RpcConnectionHandler<ServerConnection>, DrillRpcFuture<Void>{
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java
index 2d1a136..404c453 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScreenCreator.java
@@ -78,6 +78,7 @@ public class ScreenCreator implements RootCreator<Screen>{
public ScreenRoot(FragmentContext context, RecordBatch incoming, Screen config) throws OutOfMemoryException {
super(context, config);
+ // TODO Edit: That "as such" doesn't make sense.
assert context.getConnection() != null : "A screen root should only be run on the driving node which is connected directly to the client. As such, this should always be true.";
this.context = context;
this.incoming = incoming;
@@ -86,15 +87,15 @@ public class ScreenCreator implements RootCreator<Screen>{
@Override
public boolean innerNext() {
- if(!ok){
+ if (!ok) {
stop();
context.fail(this.listener.ex);
return false;
}
IterOutcome outcome = next(incoming);
-// logger.debug("Screen Outcome {}", outcome);
- switch(outcome){
+ logger.trace("Screen Outcome {}", outcome);
+ switch (outcome) {
case STOP: {
this.internalStop();
boolean verbose = context.getOptions().getOption(ExecConstants.ENABLE_VERBOSE_ERRORS_KEY).bool_val;
@@ -177,7 +178,7 @@ public class ScreenCreator implements RootCreator<Screen>{
@Override
public void stop() {
- if(!oContext.isClosed()){
+ if (!oContext.isClosed()) {
internalStop();
}
sendCount.waitForSendComplete();
@@ -213,7 +214,6 @@ public class ScreenCreator implements RootCreator<Screen>{
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
index b079428..c05b127 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
@@ -20,9 +20,14 @@ package org.apache.drill.exec.rpc.user;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.DrillBuf;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
+import javax.annotation.Nullable;
+
import org.apache.drill.exec.proto.UserBitShared;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.proto.UserBitShared.QueryResult;
@@ -36,69 +41,203 @@ import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
/**
- * Encapsulates the future management of query submissions. This entails a potential race condition. Normal ordering is:
- * 1. Submit query to be executed. 2. Receive QueryHandle for buffer management 3. Start receiving results batches for
- * query.
- *
- * However, 3 could potentially occur before 2. As such, we need to handle this case and then do a switcheroo.
- *
+ * Encapsulates the future management of query submissions. This entails a
+ * potential race condition. Normal ordering is:
+ * <ul>
+ * <li>1. Submit query to be executed. </li>
+ * <li>2. Receive QueryHandle for buffer management. </li>
+ * <li>3. Start receiving results batches for query. </li>
+ * </ul>
+ * However, 3 could potentially occur before 2. Because of that, we need to
+ * handle this case and then do a switcheroo.
*/
public class QueryResultHandler {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryResultHandler.class);
-
- private ConcurrentMap<QueryId, UserResultsListener> resultsListener = Maps.newConcurrentMap();
-
-
- public RpcOutcomeListener<QueryId> getWrappedListener(UserResultsListener listener) {
- return new SubmissionListener(listener);
+ private static final org.slf4j.Logger logger =
+ org.slf4j.LoggerFactory.getLogger(QueryResultHandler.class);
+
+ /**
+ * Current listener for results, for each active query.
+ * <p>
+ * Concurrency: Access by SubmissionLister for query-ID message vs.
+ * access by batchArrived is not otherwise synchronized.
+ * </p>
+ */
+ private final ConcurrentMap<QueryId, UserResultsListener> queryIdToResultsListenersMap =
+ Maps.newConcurrentMap();
+
+ /**
+ * Any is-last-chunk batch being deferred until the next batch
+ * (normally one with COMPLETED) arrives, per active query.
+ * <ul>
+ * <li>Last-chunk batch is added (and not passed on) when it arrives.</li>
+ * <li>Last-chunk batch is removed (and passed on) when next batch arrives
+ * and has state {@link QueryState.COMPLETED}.</li>
+ * <li>Last-chunk batch is removed (and not passed on) when next batch
+ * arrives and has state {@link QueryState.CANCELED} or
+ * {@link QueryState.FAILED}.</li>
+ * </ul>
+ */
+ private final Map<QueryId, QueryResultBatch> queryIdToDeferredLastChunkBatchesMap =
+ new ConcurrentHashMap<>();
+
+
+ public RpcOutcomeListener<QueryId> getWrappedListener(UserResultsListener resultsListener) {
+ return new SubmissionListener(resultsListener);
}
- public void batchArrived(ConnectionThrottle throttle, ByteBuf pBody, ByteBuf dBody) throws RpcException {
- final QueryResult result = RpcBus.get(pBody, QueryResult.PARSER);
- final QueryResultBatch batch = new QueryResultBatch(result, (DrillBuf) dBody);
- final boolean failed = (batch.getHeader().getQueryState() == QueryState.FAILED);
-
- assert failed || batch.getHeader().getErrorCount() == 0 : "Error count for the query batch is non-zero but QueryState != FAILED";
-
- UserResultsListener l = resultsListener.get(result.getQueryId());
- // logger.debug("For QueryId [{}], retrieved result listener {}", result.getQueryId(), l);
- if (l == null) {
- BufferingListener bl = new BufferingListener();
- l = resultsListener.putIfAbsent(result.getQueryId(), bl);
- // if we had a successful insert, use that reference. Otherwise, just throw away the new bufering listener.
- if (l == null) {
- l = bl;
+ /**
+ * Maps internal low-level API protocol to {@link UserResultsListener}-level
+ * API protocol, deferring sending is-last-chunk batches until (internal)
+ * COMPLETED batch.
+ */
+ public void batchArrived( ConnectionThrottle throttle,
+ ByteBuf pBody, ByteBuf dBody ) throws RpcException {
+ final QueryResult queryResult = RpcBus.get( pBody, QueryResult.PARSER );
+ // Current batch coming in. (Not necessarily passed along now or ever.)
+ final QueryResultBatch inputBatch = new QueryResultBatch( queryResult,
+ (DrillBuf) dBody );
+
+ final QueryId queryId = queryResult.getQueryId();
+ final QueryState queryState = inputBatch.getHeader().getQueryState();
+
+ logger.debug( "batchArrived: isLastChunk: {}, queryState: {}, queryId = {}",
+ inputBatch.getHeader().getIsLastChunk(), queryState, queryId );
+ logger.trace( "batchArrived: currentBatch = {}", inputBatch );
+
+ final boolean isFailureBatch = QueryState.FAILED == queryState;
+ final boolean isCompletionBatch = QueryState.COMPLETED == queryState;
+ final boolean isLastChunkBatchToDelay =
+ inputBatch.getHeader().getIsLastChunk() && QueryState.PENDING == queryState;
+ final boolean isTerminalBatch;
+ switch ( queryState ) {
+ case PENDING:
+ isTerminalBatch = false;
+ break;
+ case FAILED:
+ case CANCELED:
+ case COMPLETED:
+ isTerminalBatch = true;
+ break;
+ default:
+ logger.error( "Unexpected/unhandled QueryState " + queryState
+ + " (for query " + queryId + ")" );
+ isTerminalBatch = false;
+ break;
+ }
+ assert isFailureBatch || inputBatch.getHeader().getErrorCount() == 0
+ : "Error count for the query batch is non-zero but QueryState != FAILED";
+
+ UserResultsListener resultsListener = queryIdToResultsListenersMap.get( queryId );
+ logger.trace( "For QueryId [{}], retrieved results listener {}", queryId,
+ resultsListener );
+ if ( null == resultsListener ) {
+ // WHO?? didn't get query ID response and set submission listener yet,
+ // so install a buffering listener for now
+
+ BufferingResultsListener bl = new BufferingResultsListener();
+ resultsListener = queryIdToResultsListenersMap.putIfAbsent( queryId, bl );
+ // If we had a successful insertion, use that reference. Otherwise, just
+ // throw away the new buffering listener.
+ if ( null == resultsListener ) {
+ resultsListener = bl;
}
- if (result.getQueryId().toString().equals("")) {
+ // TODO: Is there a more direct way to detect a Query ID in whatever
+ // state this string comparison detects?
+ if ( queryId.toString().equals( "" ) ) {
failAll();
}
}
- if(failed) {
- String message = buildErrorMessage(batch);
- l.submissionFailed(new RpcException(message));
- resultsListener.remove(result.getQueryId(), l);
- }else{
- try {
- l.resultArrived(batch, throttle);
- } catch (Exception e) {
- batch.release();
- l.submissionFailed(new RpcException(e));
+ try {
+ if (isFailureBatch) {
+ // Failure case--pass on via submissionFailed(...).
+
+ try {
+ String message = buildErrorMessage(inputBatch);
+ resultsListener.submissionFailed(new RpcException(message));
+ }
+ finally {
+ inputBatch.release();
+ }
+ // Note: Listener and any delayed batch are removed in finally below.
+ } else {
+ // A successful (data, completion, or cancelation) case--pass on via
+ // resultArrived, delaying any last-chunk batches until following
+ // COMPLETED batch and omitting COMPLETED batch.
+
+ // If is last-chunk batch, save until next batch for query (normally a
+ // COMPLETED batch) comes in:
+ if ( isLastChunkBatchToDelay ) {
+ // We have a (non-failure) is-last-chunk batch--defer it until we get
+ // the query's COMPLETED batch.
+
+ QueryResultBatch expectNone;
+ assert null == ( expectNone =
+ queryIdToDeferredLastChunkBatchesMap.get( queryId ) )
+ : "Already have pending last-batch QueryResultBatch " + expectNone
+ + " (at receiving last-batch QueryResultBatch " + inputBatch
+ + ") for query " + queryId;
+ queryIdToDeferredLastChunkBatchesMap.put( queryId, inputBatch );
+ // Can't release batch now; will release at terminal batch in
+ // finally below.
+ } else {
+ // We have a batch triggering sending out a batch (maybe same one,
+ // maybe deferred one.
+
+ // Batch to send out in response to current batch.
+ final QueryResultBatch outputBatch;
+ if ( isCompletionBatch ) {
+ // We have a COMPLETED batch--we should have a saved is-last-chunk
+ // batch, and we must pass that on now (that we've seen COMPLETED).
+
+ outputBatch = queryIdToDeferredLastChunkBatchesMap.get( queryId );
+ assert null != outputBatch
+ : "No pending last-batch QueryResultsBatch saved, at COMPLETED"
+ + " QueryResultsBatch " + inputBatch + " for query " + queryId;
+ } else {
+ // We have a non--last-chunk PENDING batch or a CANCELED
+ // batch--pass it on.
+ outputBatch = inputBatch;
+ }
+ // Note to release input batch if it's not the batch we're sending out.
+ final boolean releaseInputBatch = outputBatch != inputBatch;
+
+ try {
+ resultsListener.resultArrived( outputBatch, throttle );
+ // That releases outputBatch if successful.
+ } catch ( Exception e ) {
+ outputBatch.release();
+ resultsListener.submissionFailed(new RpcException(e));
+ }
+ finally {
+ if ( releaseInputBatch ) {
+ inputBatch.release();
+ }
+ }
+ }
}
- }
+ } finally {
+ if ( isTerminalBatch ) {
+ // Remove and release any deferred is-last-chunk batch:
+ QueryResultBatch anyUnsentLastChunkBatch =
+ queryIdToDeferredLastChunkBatchesMap.remove( queryId );
+ if ( null != anyUnsentLastChunkBatch ) {
+ anyUnsentLastChunkBatch.release();
+ }
- if (
- (failed || result.getIsLastChunk())
- &&
- (!(l instanceof BufferingListener) || ((BufferingListener)l).output != null)
- ) {
- resultsListener.remove(result.getQueryId(), l);
+ // TODO: What exactly are we checking for? How should we really check
+ // for it?
+ if ( (! ( resultsListener instanceof BufferingResultsListener )
+ || ((BufferingResultsListener) resultsListener).output != null ) ) {
+ queryIdToResultsListenersMap.remove( queryId, resultsListener );
+ }
+ }
}
}
protected String buildErrorMessage(QueryResultBatch batch) {
StringBuilder sb = new StringBuilder();
- for (UserBitShared.DrillPBError error:batch.getHeader().getErrorList()) {
+ for (UserBitShared.DrillPBError error : batch.getHeader().getErrorList()) {
sb.append(error.getMessage());
sb.append("\n");
}
@@ -106,12 +245,12 @@ public class QueryResultHandler {
}
private void failAll() {
- for (UserResultsListener l : resultsListener.values()) {
+ for (UserResultsListener l : queryIdToResultsListenersMap.values()) {
l.submissionFailed(new RpcException("Received result without QueryId"));
}
}
- private class BufferingListener implements UserResultsListener {
+ private static class BufferingResultsListener implements UserResultsListener {
private ConcurrentLinkedQueue<QueryResultBatch> results = Queues.newConcurrentLinkedQueue();
private volatile boolean finished = false;
@@ -174,39 +313,42 @@ public class QueryResultHandler {
}
private class SubmissionListener extends BaseRpcOutcomeListener<QueryId> {
- private UserResultsListener listener;
+ private UserResultsListener resultsListener;
- public SubmissionListener(UserResultsListener listener) {
+ public SubmissionListener(UserResultsListener resultsListener) {
super();
- this.listener = listener;
+ this.resultsListener = resultsListener;
}
@Override
public void failed(RpcException ex) {
- listener.submissionFailed(ex);
+ resultsListener.submissionFailed(ex);
}
@Override
public void success(QueryId queryId, ByteBuf buf) {
- listener.queryIdArrived(queryId);
- logger.debug("Received QueryId {} succesfully. Adding listener {}", queryId, listener);
- UserResultsListener oldListener = resultsListener.putIfAbsent(queryId, listener);
-
- // we need to deal with the situation where we already received results by the time we got the query id back. In
- // that case, we'll need to transfer the buffering listener over, grabbing a lock against reception of additional
- // results during the transition
+ resultsListener.queryIdArrived(queryId);
+ logger.debug("Received QueryId {} successfully. Adding results listener {}.",
+ queryId, resultsListener);
+ UserResultsListener oldListener =
+ queryIdToResultsListenersMap.putIfAbsent(queryId, resultsListener);
+
+ // We need to deal with the situation where we already received results by
+ // the time we got the query id back. In that case, we'll need to
+ // transfer the buffering listener over, grabbing a lock against reception
+ // of additional results during the transition.
if (oldListener != null) {
logger.debug("Unable to place user results listener, buffering listener was already in place.");
- if (oldListener instanceof BufferingListener) {
- resultsListener.remove(oldListener);
- boolean all = ((BufferingListener) oldListener).transferTo(this.listener);
+ if (oldListener instanceof BufferingResultsListener) {
+ queryIdToResultsListenersMap.remove(oldListener);
+ boolean all = ((BufferingResultsListener) oldListener).transferTo(this.resultsListener);
// simply remove the buffering listener if we already have the last response.
if (all) {
- resultsListener.remove(oldListener);
+ queryIdToResultsListenersMap.remove(oldListener);
} else {
- boolean replaced = resultsListener.replace(queryId, oldListener, listener);
+ boolean replaced = queryIdToResultsListenersMap.replace(queryId, oldListener, resultsListener);
if (!replaced) {
- throw new IllegalStateException();
+ throw new IllegalStateException(); // TODO: Say what the problem is!
}
}
} else {
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
index 3da2ea9..d0a998e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
@@ -30,6 +30,7 @@ import org.apache.drill.exec.rpc.TransportCheck;
import com.codahale.metrics.MetricRegistry;
+// TODO: Doc. What kind of context? (For what aspects, RPC? What kind of data?)
public class BootStrapContext implements Closeable{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BootStrapContext.class);
@@ -68,7 +69,8 @@ public class BootStrapContext implements Closeable{
return allocator;
}
- public void close(){
+ public void close() {
+ DrillMetrics.resetMetrics();
loop.shutdownGracefully();
allocator.close();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
index b606707..0d8c892 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
@@ -237,7 +237,7 @@ public class Drillbit implements AutoCloseable {
registrationHandle = coord.register(md);
startJetty();
- Runtime.getRuntime().addShutdownHook(new ShutdownThread());
+ Runtime.getRuntime().addShutdownHook(new ShutdownThread(this));
}
@Override
@@ -269,15 +269,20 @@ public class Drillbit implements AutoCloseable {
logger.info("Shutdown completed.");
}
- private class ShutdownThread extends Thread {
- ShutdownThread() {
- setName("Drillbit-ShutdownHook");
+ private static class ShutdownThread extends Thread {
+ private static int idCounter = 0;
+ private final Drillbit drillbit;
+
+ ShutdownThread( Drillbit drillbit ) {
+ this.drillbit = drillbit;
+ idCounter++;
+ setName("Drillbit-ShutdownHook#" + idCounter );
}
@Override
public void run() {
logger.info("Received shutdown request.");
- close();
+ drillbit.close();
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/vector/accessor/SqlAccessor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/accessor/SqlAccessor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/accessor/SqlAccessor.java
index b8480b4..b69ae54 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/accessor/SqlAccessor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/accessor/SqlAccessor.java
@@ -26,10 +26,18 @@ import java.sql.Timestamp;
import org.apache.drill.exec.vector.accessor.AbstractSqlAccessor.InvalidAccessException;
+// TODO: Doc.
public interface SqlAccessor {
+ // TODO: Document (renamed) index.
+ // TODO: Rename ambiguous "index" (JDBC (1-based) column index? other index?)
+ // TODO: Doc./Spec.: What happens if index is invalid?
+
public abstract boolean isNull(int index);
+ // TODO: Clean: This interface refers to type InvalidAccessException
+ // defined in class implementing this interface.
+
public abstract BigDecimal getBigDecimal(int index) throws InvalidAccessException;
public abstract boolean getBoolean(int index) throws InvalidAccessException;
@@ -56,6 +64,9 @@ public interface SqlAccessor {
public abstract Reader getReader(int index) throws InvalidAccessException;
+ // TODO: Doc./Spec.: What should happen if called on non-string type? (Most
+ // are convertible to string. Does that result in error or conversion?)
+ // Similar question for many other methods.
public abstract String getString(int index) throws InvalidAccessException;
public abstract Time getTime(int index) throws InvalidAccessException;
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
index 9650ee5..bfb6de8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
@@ -803,7 +803,8 @@ public class Foreman implements Runnable {
// record all fragments for status purposes.
for (PlanFragment planFragment : fragments) {
-// logger.debug("Tracking intermediate remote node {} with data {}", f.getAssignment(), f.getFragmentJson());
+ logger.trace("Tracking intermediate remote node {} with data {}",
+ planFragment.getAssignment(), planFragment.getFragmentJson());
queryManager.addFragmentStatusTracker(planFragment, false);
if (planFragment.getLeafFragment()) {
leafFragmentMap.put(planFragment.getAssignment(), planFragment);
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentExecutor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentExecutor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentExecutor.java
index b6176db..5592707 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentExecutor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentExecutor.java
@@ -41,6 +41,8 @@ import org.apache.drill.exec.work.foreman.DrillbitStatusListener;
public class FragmentExecutor implements Runnable {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FragmentExecutor.class);
+ // TODO: REVIEW: Can't this be AtomicReference<FragmentState> (so that
+ // debugging and logging don't show just integer values--and for type safety)?
private final AtomicInteger state = new AtomicInteger(FragmentState.AWAITING_ALLOCATION_VALUE);
private final FragmentRoot rootOperator;
private final FragmentContext fragmentContext;
@@ -48,13 +50,26 @@ public class FragmentExecutor implements Runnable {
private volatile boolean closed;
private RootExec root;
+
public FragmentExecutor(final FragmentContext context, final FragmentRoot rootOperator,
- final StatusReporter listener) {
+ final StatusReporter listener) {
this.fragmentContext = context;
this.rootOperator = rootOperator;
this.listener = listener;
}
+ @Override
+ public String toString() {
+ return
+ super.toString()
+ + "[closed = " + closed
+ + ", state = " + state
+ + ", rootOperator = " + rootOperator
+ + ", fragmentContext = " + fragmentContext
+ + ", listener = " + listener
+ + "]";
+ }
+
public FragmentStatus getStatus() {
/*
* If the query is not in a running state, the operator tree is still being constructed and
@@ -73,10 +88,15 @@ public class FragmentExecutor implements Runnable {
}
public void cancel() {
+ logger.debug("Cancelling fragment {}", fragmentContext.getHandle());
+
// Note this will be called outside of run(), from another thread
+ // Change state checked by main loop to terminate it (if not already done):
updateState(FragmentState.CANCELLED);
- logger.debug("Cancelled Fragment {}", fragmentContext.getHandle());
+
fragmentContext.cancel();
+
+ logger.debug("Cancelled fragment {}", fragmentContext.getHandle());
}
public void receivingFragmentFinished(FragmentHandle handle) {
@@ -106,18 +126,20 @@ public class FragmentExecutor implements Runnable {
logger.debug("Starting fragment runner. {}:{}",
fragmentHandle.getMajorFragmentId(), fragmentHandle.getMinorFragmentId());
if (!updateStateOrFail(FragmentState.AWAITING_ALLOCATION, FragmentState.RUNNING)) {
- logger.warn("Unable to set fragment state to RUNNING. Cancelled or failed?");
+ logger.warn("Unable to set fragment state to RUNNING. Cancelled or failed?");
return;
}
/*
- * Run the query until root.next returns false.
- * Note that we closeOutResources() here if we're done. That's because this can also throw
- * exceptions that we want to treat as failures of the request, even if the request did fine
- * up until this point. Any failures there will be caught in the catch clause below, which
- * will be reported to the user. If they were to come from the finally clause, the uncaught
- * exception there will simply terminate this thread without alerting the user -- the
- * behavior then is to hang.
+ * Run the query until root.next returns false OR cancel() changes the
+ * state.
+ * Note that we closeOutResources() here if we're done. That's because
+ * this can also throw exceptions that we want to treat as failures of the
+ * request, even if the request did fine up until this point. Any
+ * failures there will be caught in the catch clause below, which will be
+ * reported to the user. If they were to come from the finally clause,
+ * the uncaught exception there will simply terminate this thread without
+ * alerting the user--the behavior then is to hang.
*/
while (state.get() == FragmentState.RUNNING_VALUE) {
if (!root.next()) {
@@ -191,7 +213,7 @@ public class FragmentExecutor implements Runnable {
/**
* Updates the fragment state with the given state
*
- * @param to target state
+ * @param to target state
*/
private void updateState(final FragmentState to) {
state.set(to.getNumber());
@@ -201,8 +223,8 @@ public class FragmentExecutor implements Runnable {
/**
* Updates the fragment state only if the current state matches the expected.
*
- * @param expected expected current state
- * @param to target state
+ * @param expected expected current state
+ * @param to target state
* @return true only if update succeeds
*/
private boolean checkAndUpdateState(final FragmentState expected, final FragmentState to) {
@@ -229,8 +251,8 @@ public class FragmentExecutor implements Runnable {
* Update the state if current state matches expected or fail the fragment if state transition fails even though
* fragment is not in a terminal state.
*
- * @param expected current expected state
- * @param to target state
+ * @param expected current expected state
+ * @param to target state
* @return true only if update succeeds
*/
private boolean updateStateOrFail(final FragmentState expected, final FragmentState to) {
@@ -257,8 +279,9 @@ public class FragmentExecutor implements Runnable {
// if the defunct Drillbit was running our Foreman, then cancel the query
final DrillbitEndpoint foremanEndpoint = FragmentExecutor.this.fragmentContext.getForemanEndpoint();
if (unregisteredDrillbits.contains(foremanEndpoint)) {
- logger.warn("Foreman : {} no longer active. Cancelling fragment {}.",
- foremanEndpoint.getAddress(), QueryIdHelper.getQueryIdentifier(fragmentContext.getHandle()));
+ logger.warn("Foreman {} no longer active. Cancelling fragment {}.",
+ foremanEndpoint.getAddress(),
+ QueryIdHelper.getQueryIdentifier(fragmentContext.getHandle()));
FragmentExecutor.this.cancel();
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/java-exec/src/test/java/org/apache/drill/exec/ExecTest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/ExecTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/ExecTest.java
index 0272b23..8a1aecb 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/ExecTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/ExecTest.java
@@ -25,6 +25,8 @@ public class ExecTest extends DrillTest {
@After
public void clear(){
+ // TODO: (Re DRILL-1735) Check whether still needed now that
+ // BootstrapContext.close() resets the metrics.
DrillMetrics.resetMetrics();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/AvaticaDrillSqlAccessor.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/AvaticaDrillSqlAccessor.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/AvaticaDrillSqlAccessor.java
index cf5829a..3702257 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/AvaticaDrillSqlAccessor.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/AvaticaDrillSqlAccessor.java
@@ -43,67 +43,77 @@ import org.apache.drill.exec.vector.accessor.SqlAccessor;
public class AvaticaDrillSqlAccessor implements Accessor{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AvaticaDrillSqlAccessor.class);
- private SqlAccessor a;
+ private SqlAccessor underlyingAccessor;
private DrillCursor cursor;
public AvaticaDrillSqlAccessor(SqlAccessor drillSqlAccessor, DrillCursor cursor) {
super();
- this.a = drillSqlAccessor;
+ this.underlyingAccessor = drillSqlAccessor;
this.cursor = cursor;
}
- private int row(){
- return cursor.currentRecord;
+ private int getCurrentRecordNumber() throws SQLException {
+ if ( cursor.getResultSet().isBeforeFirst() ) {
+ throw new InvalidCursorStateSqlException(
+ "Result set cursor is positioned before all rows. Call next() first." );
+ }
+ else if ( cursor.getResultSet().isAfterLast() ) {
+ throw new InvalidCursorStateSqlException(
+ "Result set cursor is already positioned past all rows." );
+ }
+ else {
+ return cursor.getCurrentRecordNumber();
+ }
}
@Override
- public boolean wasNull() {
- return a.isNull(row());
+ public boolean wasNull() throws SQLException {
+ return underlyingAccessor.isNull(getCurrentRecordNumber());
}
@Override
public String getString() throws SQLException {
- return a.getString(row());
+ return underlyingAccessor.getString(getCurrentRecordNumber());
}
@Override
public boolean getBoolean() throws SQLException {
- return a.getBoolean(row());
+ return underlyingAccessor.getBoolean(getCurrentRecordNumber());
}
@Override
public byte getByte() throws SQLException {
- return a.getByte(row());
+ return underlyingAccessor.getByte(getCurrentRecordNumber());
}
@Override
public short getShort() throws SQLException {
- return a.getShort(row());
+ return underlyingAccessor.getShort(getCurrentRecordNumber());
}
@Override
public int getInt() throws SQLException {
- return a.getInt(row());
+ return underlyingAccessor.getInt(getCurrentRecordNumber());
}
@Override
public long getLong() throws SQLException {
- return a.getLong(row());
+ return underlyingAccessor.getLong(getCurrentRecordNumber());
}
@Override
public float getFloat() throws SQLException {
- return a.getFloat(row());
+ return underlyingAccessor.getFloat(getCurrentRecordNumber());
}
@Override
public double getDouble() throws SQLException {
- return a.getDouble(row());
+ return underlyingAccessor.getDouble(getCurrentRecordNumber());
}
@Override
public BigDecimal getBigDecimal() throws SQLException {
- return a.getBigDecimal(row());
+ return underlyingAccessor.getBigDecimal(getCurrentRecordNumber());
}
@Override
@@ -113,32 +123,32 @@ public class AvaticaDrillSqlAccessor implements Accessor{
@Override
public byte[] getBytes() throws SQLException {
- return a.getBytes(row());
+ return underlyingAccessor.getBytes(getCurrentRecordNumber());
}
@Override
public InputStream getAsciiStream() throws SQLException {
- return a.getStream(row());
+ return underlyingAccessor.getStream(getCurrentRecordNumber());
}
@Override
public InputStream getUnicodeStream() throws SQLException {
- return a.getStream(row());
+ return underlyingAccessor.getStream(getCurrentRecordNumber());
}
@Override
public InputStream getBinaryStream() throws SQLException {
- return a.getStream(row());
+ return underlyingAccessor.getStream(getCurrentRecordNumber());
}
@Override
public Object getObject() throws SQLException {
- return a.getObject(row());
+ return underlyingAccessor.getObject(getCurrentRecordNumber());
}
@Override
public Reader getCharacterStream() throws SQLException {
- return a.getReader(row());
+ return underlyingAccessor.getReader(getCurrentRecordNumber());
}
@Override
@@ -168,17 +178,17 @@ public class AvaticaDrillSqlAccessor implements Accessor{
@Override
public Date getDate(Calendar calendar) throws SQLException {
- return a.getDate(row());
+ return underlyingAccessor.getDate(getCurrentRecordNumber());
}
@Override
public Time getTime(Calendar calendar) throws SQLException {
- return a.getTime(row());
+ return underlyingAccessor.getTime(getCurrentRecordNumber());
}
@Override
public Timestamp getTimestamp(Calendar calendar) throws SQLException {
- return a.getTimestamp(row());
+ return underlyingAccessor.getTimestamp(getCurrentRecordNumber());
}
@Override
@@ -198,12 +208,12 @@ public class AvaticaDrillSqlAccessor implements Accessor{
@Override
public String getNString() throws SQLException {
- return a.getString(row());
+ return underlyingAccessor.getString(getCurrentRecordNumber());
}
@Override
public Reader getNCharacterStream() throws SQLException {
- return a.getReader(row());
+ return underlyingAccessor.getReader(getCurrentRecordNumber());
}
@Override
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillAccessorList.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillAccessorList.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillAccessorList.java
index 82d51f1..ccf2658 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillAccessorList.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillAccessorList.java
@@ -31,6 +31,8 @@ public class DrillAccessorList extends BasicList<Accessor>{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillAccessorList.class);
private Accessor[] accessors = new Accessor[0];
+ // TODO Rename to lastColumnAccessed and/or document.
+ // TODO Why 1, rather than, say, -1?
private int lastColumn = 1;
public void generateAccessors(DrillCursor cursor, RecordBatchLoader currentBatch){
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionConfig.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionConfig.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionConfig.java
index 54e31b1..de08cda 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionConfig.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionConfig.java
@@ -31,9 +31,11 @@ public class DrillConnectionConfig extends ConnectionConfigImpl {
}
public boolean isLocal(){
+ // TODO Why doesn't this call getZookeeperConnectionString()?
return "local".equals(props.getProperty("zk"));
}
+ // TODO: Check: Shouldn't something validate that URL has "zk" parameter?
public String getZookeeperConnectionString(){
return props.getProperty("zk");
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionImpl.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionImpl.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionImpl.java
index f19aab0..e590778 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionImpl.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillConnectionImpl.java
@@ -44,17 +44,18 @@ import org.apache.drill.exec.server.RemoteServiceSet;
* Abstract to allow newer versions of JDBC to add methods.
* </p>
*/
-abstract class DrillConnectionImpl extends AvaticaConnection implements org.apache.drill.jdbc.DrillConnection {
- public final DrillStatementRegistry registry = new DrillStatementRegistry();
- final DrillConnectionConfig config;
-
+abstract class DrillConnectionImpl extends AvaticaConnection implements DrillConnection {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillConnection.class);
+ final DrillStatementRegistry openStatementsRegistry = new DrillStatementRegistry();
+ final DrillConnectionConfig config;
+
private final DrillClient client;
private final BufferAllocator allocator;
private Drillbit bit;
private RemoteServiceSet serviceSet;
+
protected DrillConnectionImpl(Driver driver, AvaticaFactory factory, String url, Properties info) throws SQLException {
super(driver, factory, url, info);
this.config = new DrillConnectionConfig(info);
@@ -71,7 +72,7 @@ abstract class DrillConnectionImpl extends AvaticaConnection implements org.apac
this.allocator = new TopLevelAllocator(dConfig);
RemoteServiceSet set = GlobalServiceSetReference.SETS.get();
if (set == null) {
- // we're embedded, start a local drill bit.
+ // We're embedded; start a local drill bit.
serviceSet = RemoteServiceSet.getLocalServiceSet();
set = serviceSet;
try {
@@ -89,6 +90,10 @@ abstract class DrillConnectionImpl extends AvaticaConnection implements org.apac
} else {
final DrillConfig dConfig = DrillConfig.forClient();
this.allocator = new TopLevelAllocator(dConfig);
+ // TODO: Check: Why does new DrillClient() create another DrillConfig,
+ // with enableServerConfigs true, and cause scanning for function
+ // implementations (needed by a server, but not by a client-only
+ // process, right?)? Probably pass dConfig to construction.
this.client = new DrillClient();
this.client.connect(config.getZookeeperConnectionString(), info);
}
@@ -121,21 +126,24 @@ abstract class DrillConnectionImpl extends AvaticaConnection implements org.apac
}
@Override
- public DrillStatement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability)
- throws SQLException {
- DrillStatement statement = (DrillStatement) super.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
- registry.addStatement(statement);
+ public DrillStatement createStatement(int resultSetType, int resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
+ DrillStatement statement =
+ (DrillStatement) super.createStatement(resultSetType, resultSetConcurrency,
+ resultSetHoldability);
return statement;
}
@Override
- public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency,
- int resultSetHoldability) throws SQLException {
+ public PreparedStatement prepareStatement(String sql, int resultSetType,
+ int resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
try {
DrillPrepareResult prepareResult = new DrillPrepareResult(sql);
- DrillPreparedStatement statement = (DrillPreparedStatement) factory.newPreparedStatement(this, prepareResult,
- resultSetType, resultSetConcurrency, resultSetHoldability);
- registry.addStatement(statement);
+ DrillPreparedStatement statement =
+ (DrillPreparedStatement) factory.newPreparedStatement(
+ this, prepareResult, resultSetType, resultSetConcurrency,
+ resultSetHoldability);
return statement;
} catch (RuntimeException e) {
throw Helper.INSTANCE.createException("Error while preparing statement [" + sql + "]", e);
@@ -160,6 +168,10 @@ abstract class DrillConnectionImpl extends AvaticaConnection implements org.apac
}
void cleanup() {
+ // First close any open JDBC Statement objects, to close any open ResultSet
+ // objects and release their buffers/vectors.
+ openStatementsRegistry.close();
+
client.close();
allocator.close();
if (bit != null) {
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillCursor.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillCursor.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillCursor.java
index fbe611f..cddd999 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillCursor.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillCursor.java
@@ -31,32 +31,50 @@ import org.apache.drill.exec.record.RecordBatchLoader;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.user.QueryResultBatch;
-public class DrillCursor implements Cursor{
+public class DrillCursor implements Cursor {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillCursor.class);
private static final String UNKNOWN = "--UNKNOWN--";
+ /** The associated java.sql.ResultSet implementation. */
+ private final DrillResultSet resultSet;
+
+ private final RecordBatchLoader currentBatch;
+ private final DrillResultSet.ResultsListener resultsListener;
+
+ // TODO: Doc.: Say what's started (set of rows? just current result batch?)
private boolean started = false;
private boolean finished = false;
- private final RecordBatchLoader currentBatch;
- private final DrillResultSet.Listener listener;
+ // TODO: Doc.: Say what "readFirstNext" means.
private boolean redoFirstNext = false;
+ // TODO: Doc.: First what? (First batch? record? "next" call/operation?)
private boolean first = true;
private DrillColumnMetaDataList columnMetaDataList;
private BatchSchema schema;
- final DrillResultSet results;
- int currentRecord = 0;
+ /** Zero-based index of current record in record batch. */
+ private int currentRecordNumber = -1;
private long recordBatchCount;
private final DrillAccessorList accessors = new DrillAccessorList();
- public DrillCursor(DrillResultSet results) {
- super();
- this.results = results;
- currentBatch = results.currentBatch;
- this.listener = results.listener;
+ /**
+ *
+ * @param resultSet the associated ResultSet implementation
+ */
+ public DrillCursor(final DrillResultSet resultSet) {
+ this.resultSet = resultSet;
+ currentBatch = resultSet.currentBatch;
+ resultsListener = resultSet.resultslistener;
+ }
+
+ public DrillResultSet getResultSet() {
+ return resultSet;
+ }
+
+ protected int getCurrentRecordNumber() {
+ return currentRecordNumber;
}
@Override
@@ -65,12 +83,20 @@ public class DrillCursor implements Cursor{
return accessors;
}
+ // TODO: Doc.: Specify what the return value actually means. (The wording
+ // "Moves to the next row" and "Whether moved" from the documentation of the
+ // implemented interface (net.hydromatic.avatica.Cursor) doesn't address
+ // moving past last row or how to evaluate "whether moved" on the first call.
+ // In particular, document what the return value indicates about whether we're
+ // currently at a valid row (or whether next() can be called again, or
+ // whatever it does indicate), especially the first time this next() called
+ // for a new result.
@Override
public boolean next() throws SQLException {
if (!started) {
started = true;
redoFirstNext = true;
- } else if(redoFirstNext && !finished) {
+ } else if (redoFirstNext && !finished) {
redoFirstNext = false;
return true;
}
@@ -79,27 +105,32 @@ public class DrillCursor implements Cursor{
return false;
}
- if (currentRecord+1 < currentBatch.getRecordCount()) {
- currentRecord++;
+ if (currentRecordNumber + 1 < currentBatch.getRecordCount()) {
+ // Next index is in within current batch--just increment to that record.
+ currentRecordNumber++;
return true;
} else {
+ // Next index is not in current batch (including initial empty batch--
+ // (try to) get next batch.
try {
- QueryResultBatch qrb = listener.getNext();
+ QueryResultBatch qrb = resultsListener.getNext();
recordBatchCount++;
while (qrb != null && qrb.getHeader().getRowCount() == 0 && !first) {
qrb.release();
- qrb = listener.getNext();
+ qrb = resultsListener.getNext();
recordBatchCount++;
}
first = false;
if (qrb == null) {
+ currentBatch.clear();
finished = true;
return false;
} else {
- currentRecord = 0;
+ currentRecordNumber = 0;
boolean changed = currentBatch.load(qrb.getHeader().getDef(), qrb.getData());
+ qrb.release();
schema = currentBatch.getSchema();
if (changed) {
updateColumns();
@@ -119,8 +150,8 @@ public class DrillCursor implements Cursor{
void updateColumns() {
accessors.generateAccessors(this, currentBatch);
columnMetaDataList.updateColumnMetaData(UNKNOWN, UNKNOWN, UNKNOWN, schema);
- if (results.changeListener != null) {
- results.changeListener.schemaChanged(schema);
+ if (getResultSet().changeListener != null) {
+ getResultSet().changeListener.schemaChanged(schema);
}
}
@@ -130,7 +161,17 @@ public class DrillCursor implements Cursor{
@Override
public void close() {
- results.cleanup();
+ // currentBatch is owned by resultSet and cleaned up by
+ // DrillResultSet.cleanup()
+
+ // listener is owned by resultSet and cleaned up by
+ // DrillResultSet.cleanup()
+
+ // Clean up result set (to deallocate any buffers).
+ getResultSet().cleanup();
+ // TODO: CHECK: Something might need to set statement.openResultSet to
+ // null. Also, AvaticaResultSet.close() doesn't check whether already
+ // closed and skip calls to cursor.close(), statement.onResultSetClose()
}
@Override
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillPreparedStatement.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillPreparedStatement.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillPreparedStatement.java
index cfcee8c..4397c2f 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillPreparedStatement.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillPreparedStatement.java
@@ -30,11 +30,13 @@ import net.hydromatic.avatica.AvaticaPreparedStatement;
* {@link net.hydromatic.avatica.AvaticaFactory#newPreparedStatement}.
* </p>
*/
-abstract class DrillPreparedStatement extends AvaticaPreparedStatement implements DrillRemoteStatement {
+abstract class DrillPreparedStatement extends AvaticaPreparedStatement
+ implements DrillRemoteStatement {
protected DrillPreparedStatement(DrillConnectionImpl connection, AvaticaPrepareResult prepareResult,
int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
super(connection, prepareResult, resultSetType, resultSetConcurrency, resultSetHoldability);
+ connection.openStatementsRegistry.addStatement(this);
}
@Override
@@ -45,6 +47,6 @@ abstract class DrillPreparedStatement extends AvaticaPreparedStatement implement
@Override
public void cleanup() {
final DrillConnectionImpl connection1 = (DrillConnectionImpl) connection;
- connection1.registry.removeStatement(this);
+ connection1.openStatementsRegistry.removeStatement(this);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillResultSet.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillResultSet.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillResultSet.java
index 77b2c37..0ce33f4 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillResultSet.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillResultSet.java
@@ -19,6 +19,8 @@ package org.apache.drill.jdbc;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.TimeZone;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
@@ -28,6 +30,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import net.hydromatic.avatica.AvaticaPrepareResult;
import net.hydromatic.avatica.AvaticaResultSet;
import net.hydromatic.avatica.AvaticaStatement;
+import net.hydromatic.avatica.Cursor;
+import net.hydromatic.avatica.Cursor.Accessor;
import org.apache.drill.exec.client.DrillClient;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
@@ -46,7 +50,7 @@ public class DrillResultSet extends AvaticaResultSet {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillResultSet.class);
SchemaChangeListener changeListener;
- final Listener listener = new Listener();
+ final ResultsListener resultslistener = new ResultsListener();
private volatile QueryId queryId;
private final DrillClient client;
final RecordBatchLoader currentBatch;
@@ -70,10 +74,11 @@ public class DrillResultSet extends AvaticaResultSet {
}
synchronized void cleanup() {
- if (queryId != null && !listener.completed) {
+ if (queryId != null && ! resultslistener.completed) {
client.cancelQuery(queryId);
}
- listener.close();
+ resultslistener.close();
+ currentBatch.clear();
}
@Override
@@ -81,29 +86,32 @@ public class DrillResultSet extends AvaticaResultSet {
// Next may be called after close has been called (for example after a user cancel) which in turn
// sets the cursor to null. So we must check before we call next.
// TODO: handle next() after close is called in the Avatica code.
- if(super.cursor!=null){
+ if (super.cursor != null) {
return super.next();
- }else{
+ } else {
return false;
}
-
}
-
- @Override protected DrillResultSet execute() throws SQLException{
+ @Override
+ protected DrillResultSet execute() throws SQLException{
// Call driver's callback. It is permitted to throw a RuntimeException.
DrillConnectionImpl connection = (DrillConnectionImpl) statement.getConnection();
- connection.getClient().runQuery(QueryType.SQL, this.prepareResult.getSql(), listener);
+ connection.getClient().runQuery(QueryType.SQL, this.prepareResult.getSql(),
+ resultslistener);
connection.getDriver().handler.onStatementExecute(statement, null);
super.execute();
// don't return with metadata until we've achieved at least one return message.
try {
- listener.latch.await();
- cursor.next();
+ resultslistener.latch.await();
+ boolean notAtEnd = cursor.next();
+ assert notAtEnd;
} catch (InterruptedException e) {
+ // TODO: Check: Should this call Thread.currentThread.interrupt()? If
+ // not, at least document why this is empty.
}
return this;
@@ -117,7 +125,7 @@ public class DrillResultSet extends AvaticaResultSet {
}
}
- class Listener implements UserResultsListener {
+ class ResultsListener implements UserResultsListener {
private static final int MAX = 100;
private volatile RpcException ex;
volatile boolean completed = false;
@@ -131,6 +139,7 @@ public class DrillResultSet extends AvaticaResultSet {
final LinkedBlockingDeque<QueryResultBatch> queue = Queues.newLinkedBlockingDeque();
+ // TODO: Doc.: Release what if what is first relative to what?
private boolean releaseIfFirst() {
if (receivedMessage.compareAndSet(false, true)) {
latch.countDown();
@@ -158,14 +167,14 @@ public class DrillResultSet extends AvaticaResultSet {
return;
}
- // if we're in a closed state, just release the message.
+ // If we're in a closed state, just release the message.
if (closed) {
result.release();
completed = true;
return;
}
- // we're active, let's add to the queue.
+ // We're active; let's add to the queue.
queue.add(result);
if (queue.size() >= MAX - 1) {
throttle.setAutoRead(false);
@@ -185,6 +194,7 @@ public class DrillResultSet extends AvaticaResultSet {
}
+ // TODO: Doc.: Specify whether result can be null and what that means.
public QueryResultBatch getNext() throws RpcException, InterruptedException {
while (true) {
if (ex != null) {
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatement.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatement.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatement.java
index fec126e..d934c7c 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatement.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatement.java
@@ -19,10 +19,12 @@ package org.apache.drill.jdbc;
import net.hydromatic.avatica.AvaticaStatement;
-public abstract class DrillStatement extends AvaticaStatement implements DrillRemoteStatement {
+public abstract class DrillStatement extends AvaticaStatement
+ implements DrillRemoteStatement {
DrillStatement(DrillConnectionImpl connection, int resultSetType, int resultSetConcurrency, int resultSetHoldability) {
super(connection, resultSetType, resultSetConcurrency, resultSetHoldability);
+ connection.openStatementsRegistry.addStatement(this);
}
@Override
@@ -33,7 +35,7 @@ public abstract class DrillStatement extends AvaticaStatement implements DrillRe
@Override
public void cleanup() {
final DrillConnectionImpl connection1 = (DrillConnectionImpl) connection;
- connection1.registry.removeStatement(this);
+ connection1.openStatementsRegistry.removeStatement(this);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatementRegistry.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatementRegistry.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatementRegistry.java
index cc797fa..adbbb64 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatementRegistry.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/DrillStatementRegistry.java
@@ -17,12 +17,61 @@
*/
package org.apache.drill.jdbc;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Registry of open statements (for a connection), for closing them when a
+ * connection is closed.
+ * <p>
+ * Concurrency: Not thread-safe. (Creating statements, closing statements,
+ * and closing connection cannot be concurrent (unless concurrency is
+ * coordinated elsewhere).)
+ * </p>
+ */
class DrillStatementRegistry {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillStatementRegistry.class);
+ private static final Logger logger = getLogger( DrillStatementRegistry.class );
+
+ /** ... (using as IdentityHash*Set*) */
+ private final Map<Statement, Object> openStatements = new IdentityHashMap<>();
+
+
+ public void addStatement( Statement statement ) {
+ logger.debug( "Adding to open-statements registry: " + statement );
+ openStatements.put( statement, statement );
+ }
+
+ public void removeStatement( Statement statement ) {
+ logger.debug( "Removing from open-statements registry: " + statement );
+ openStatements.remove( statement );
+ }
+
+ public void close() {
+ // Note: Can't call close() on statement during iteration of map because
+ // close() calls our removeStatement(...), which modifies the map.
- public void addStatement(DrillRemoteStatement statement){}
- public void removeStatement(DrillRemoteStatement statement){}
+ // Copy set of open statements to other collection before closing:
+ final List<Statement> copiedList = new ArrayList<>( openStatements.keySet() );
- public void close(){}
+ for ( final Statement statement : copiedList ) {
+ try {
+ logger.debug( "Auto-closing (via open-statements registry): " + statement );
+ statement.close();
+ }
+ catch ( SQLException e ) {
+ logger.error( "Error auto-closing statement " + statement + ": " + e, e );
+ // Otherwise ignore the error, to close which statements we can close.
+ }
+ }
+ openStatements.clear();
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/Driver.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/Driver.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/Driver.java
index 974e786..55453e8 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/Driver.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/Driver.java
@@ -71,9 +71,11 @@ public class Driver extends UnregisteredDriver {
@Override
protected Handler createHandler() {
- return new HandlerImpl();
+ return new DrillHandler();
}
+ // Any reference to class loads class, and loading class instantiates an
+ // instance and has it register itself:
static {
new Driver().register();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/InvalidCursorStateSqlException.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/InvalidCursorStateSqlException.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/InvalidCursorStateSqlException.java
new file mode 100644
index 0000000..7b04371
--- /dev/null
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/InvalidCursorStateSqlException.java
@@ -0,0 +1,97 @@
+/*
+ * 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.drill.jdbc;
+
+import java.sql.ResultSet;
+
+/**
+ * SQLException for invalid-cursor-state conditions, e.g., calling a column
+ * accessor method before calling {@link ResultSet#next()} or after
+ * {@link ResultSet#next()} returns false.
+ *
+ */
+class InvalidCursorStateSqlException extends JdbcApiSqlException {
+
+ private static final long serialVersionUID = 2014_12_09L;
+
+
+ /**
+ * See {@link JdbcApiSqlException#JdbcApiSqlException(String, String, int)}.
+ */
+ public InvalidCursorStateSqlException( String reason,
+ String SQLState,
+ int vendorCode ) {
+ super( reason, SQLState, vendorCode );
+ }
+
+ /**
+ * See {@link JdbcApiSqlException#JdbcApiSqlException(String, String)}.
+ */
+ public InvalidCursorStateSqlException( String reason, String SQLState ) {
+ super( reason, SQLState );
+ }
+
+ /**
+ * See {@link JdbcApiSqlException#JdbcApiSqlException(String)}.
+ */
+ public InvalidCursorStateSqlException( String reason ) {
+ super( reason );
+ }
+
+ /**
+ * See {@link JdbcApiSqlException#JdbcApiSqlException()}.
+ * */
+ public InvalidCursorStateSqlException() {
+ super();
+ }
+
+ /**
+ * See {@link JdbcApiSqlException#JdbcApiSqlException(Throwable cause)}.
+ */
+ public InvalidCursorStateSqlException( Throwable cause ) {
+ super( cause );
+ }
+
+ /**
+ * See {@link JdbcApiSqlException#JdbcApiSqlException(String, Throwable)}.
+ */
+ public InvalidCursorStateSqlException( String reason, Throwable cause ) {
+ super( reason, cause );
+ }
+
+ /**
+ * See
+ * {@link JdbcApiSqlException#JdbcApiSqlException(String, String, Throwable)}.
+ */
+ public InvalidCursorStateSqlException( String reason, String sqlState,
+ Throwable cause ) {
+ super( reason, sqlState, cause );
+ }
+
+ /**
+ * See
+ * {@link JdbcApiSqlException#JdbcApiSqlException(String, String, int, Throwable)}.
+ */
+ public InvalidCursorStateSqlException( String reason,
+ String sqlState,
+ int vendorCode,
+ Throwable cause ) {
+ super( reason, sqlState, vendorCode, cause );
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/main/java/org/apache/drill/jdbc/JdbcApiSqlException.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/JdbcApiSqlException.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/JdbcApiSqlException.java
new file mode 100644
index 0000000..d6b05fb
--- /dev/null
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/JdbcApiSqlException.java
@@ -0,0 +1,155 @@
+/*
+ * 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.drill.jdbc;
+
+import java.sql.SQLException;
+
+/**
+ * SQLException for JDBC API calling sequence/state problems.
+ *
+ * <p>
+ * {@code JdbcApiSqlException} is intended for errors in using the JDBC API,
+ * such as calling {@link ResultSet#getString} before calling
+ * {@link ResultSet#next}.
+ * </p>
+ * <p>
+ * ({@code JdbcApiSqlException} is not for errors that are not under direct
+ * control of the programmer writing JDBC API calls, for example, invalid SQL
+ * syntax, errors from SQL-vs.-data mismatches, data file format errors,
+ * resource availability errors, or internal Drill errors.)
+ * </p>
+ * <p>
+ * TODO: Consider having a DrillSqlException (in part for reviewing,
+ * coordinating, and revising the many uses of SQLException in the code).
+ * </p>
+ * <p>
+ * TODO: Consider using ANSI-/XOPEN-standard SQL State values. (See:
+ * </p>
+ * <ul>
+ * <li>
+ * <a href="
+ * http://stackoverflow.com/questions/1399574/what-are-all-the-possible-values-for-sqlexception-getsqlstate
+ * ">
+ * http://stackoverflow.com/questions/1399574/what-are-all-the-possible-values-for-sqlexception-getsqlstate
+ * </a>
+ * </li>
+ * <li>
+ * <a href="
+ * https://github.com/olamedia/kanon/blob/master/src/mvc-model/storageDrivers/SQLSTATE.txt
+ * ">
+ * https://github.com/olamedia/kanon/blob/master/src/mvc-model/storageDrivers/SQLSTATE.txt
+ * </a>
+ * </li>
+ * <li>
+ * <a href="
+ * http://kanon-framework.googlecode.com/svn/trunk/src/mvc-model/storageDrivers/SQLSTATE.txt
+ * ">
+ * http://kanon-framework.googlecode.com/svn/trunk/src/mvc-model/storageDrivers/SQLSTATE.txt
+ * </a>
+ * </li>
+ * <li>
+ * <a href="
+ * http://www-01.ibm.com/support/knowledgecenter/api/content/nl/en-us/SSVHEW_6.2.0/com.ibm.rcp.tools.doc.db2e/adg/sql11.html
+ * ">
+ * http://www-01.ibm.com/support/knowledgecenter/api/content/nl/en-us/SSVHEW_6.2.0/com.ibm.rcp.tools.doc.db2e/adg/sql11.html
+ * </a>
+ * </li>
+ * <li>
+ * <a href="
+ * ftp://ftp.software.ibm.com/ps/products/db2/info/vr6/htm/db2m0/db2state.htm
+ * ">
+ * ftp://ftp.software.ibm.com/ps/products/db2/info/vr6/htm/db2m0/db2state.htm
+ * </a>
+ * </li>
+ * <li>
+ * <a href="
+ * https://docs.oracle.com/cd/E15817_01/appdev.111/b31230/ch2.htm
+ * ">
+ * https://docs.oracle.com/cd/E15817_01/appdev.111/b31230/ch2.htm
+ * </a>
+ * </li>
+ * </ul>
+ * <p>
+ * etc.)
+ * </p>
+ */
+class JdbcApiSqlException extends SQLException {
+
+ private static final long serialVersionUID = 2014_12_12L;
+
+
+ /**
+ * See {@link SQLException#SQLException(String, String, int)}.
+ */
+ public JdbcApiSqlException( String reason, String SQLState, int vendorCode ) {
+ super( reason, SQLState, vendorCode );
+ }
+
+ /**
+ * See {@link SQLException#SQLException(String, String)}.
+ */
+ public JdbcApiSqlException( String reason, String SQLState ) {
+ super( reason, SQLState );
+ }
+
+ /**
+ * See {@link SQLException#SQLException(String)}.
+ */
+ public JdbcApiSqlException( String reason ) {
+ super( reason );
+ }
+
+ /**
+ * See {@link SQLException#SQLException()}.
+ * */
+ public JdbcApiSqlException() {
+ super();
+ }
+
+ /**
+ * See {@link SQLException#SQLException(Throwable cause)}.
+ */
+ public JdbcApiSqlException( Throwable cause ) {
+ super( cause );
+ }
+
+ /**
+ * See {@link SQLException#SQLException(String, Throwable)}.
+ */
+ public JdbcApiSqlException( String reason, Throwable cause ) {
+ super( reason, cause );
+ }
+
+ /**
+ * See {@link SQLException#SQLException(String, String, Throwable)}.
+ */
+ public JdbcApiSqlException( String reason, String sqlState, Throwable cause ) {
+ super( reason, sqlState, cause );
+ }
+
+ /**
+ * See {@link SQLException#SQLException(String, String, int, Throwable)}.
+ */
+ public JdbcApiSqlException( String reason,
+ String sqlState,
+ int vendorCode,
+ Throwable cause ) {
+ super( reason, sqlState, vendorCode, cause );
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/9c9ee8c4/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java
new file mode 100644
index 0000000..de19615
--- /dev/null
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java
@@ -0,0 +1,159 @@
+/*
+ * 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.drill.jdbc;
+
+import static org.junit.Assert.*;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import net.hydromatic.linq4j.Ord;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.logical.LogicalPlan;
+import org.apache.drill.common.logical.data.LogicalOperator;
+import org.apache.drill.common.util.Hook;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.test.DrillTest;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.anyOf;
+import static org.hamcrest.core.StringContains.containsString;
+
+import com.google.common.base.Function;
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.ImmutableSet.Builder;
+
+public class DrillResultSetTest extends DrillTest {
+
+ // TODO: Move Jetty status server disabling to DrillTest.
+ private static final String STATUS_SERVER_PROPERTY_NAME =
+ ExecConstants.HTTP_ENABLE;
+
+ private static final String origStatusServerPropValue =
+ System.getProperty( STATUS_SERVER_PROPERTY_NAME, "true" );
+
+ // Disable Jetty status server so unit tests run (outside Maven setup).
+ // (TODO: Move this to base test class and/or have Jetty try other ports.)
+ @BeforeClass
+ public static void setUpClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, "false" );
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ System.setProperty( STATUS_SERVER_PROPERTY_NAME, origStatusServerPropValue );
+ }
+
+
+ @Test
+ public void test_next_blocksFurtherAccessAfterEnd()
+ throws SQLException
+ {
+ Connection connection = new Driver().connect( "jdbc:drill:zk=local", null );
+ Statement statement = connection.createStatement();
+ ResultSet resultSet =
+ statement.executeQuery( "SELECT 1 AS x \n" +
+ "FROM cp.`donuts.json` \n" +
+ "LIMIT 2" );
+
+ // Advance to first row; confirm can access data.
+ assertThat( resultSet.next(), is( true ) );
+ assertThat( resultSet.getInt( 1 ), is ( 1 ) );
+
+ // Advance from first to second (last) row, confirming data access.
+ assertThat( resultSet.next(), is( true ) );
+ assertThat( resultSet.getInt( 1 ), is ( 1 ) );
+
+ // Now advance past last row.
+ assertThat( resultSet.next(), is( false ) );
+
+ // Main check: That row data access methods now throw SQLException.
+ try {
+ resultSet.getInt( 1 );
+ fail( "Did get expected SQLException." );
+ }
+ catch ( SQLException e ) {
+ // Expect something like current InvalidCursorStateSqlException saying
+ // "Result set cursor is already positioned past all rows."
+ assertThat( e, instanceOf( InvalidCursorStateSqlException.class ) );
+ assertThat( e.toString(), containsString( "past" ) );
+ }
+ // (Any other exception is unexpected result.)
+
+ assertThat( resultSet.next(), is( false ) );
+
+ // TODO: Ideally, test all other accessor methods.
+ }
+
+ @Test
+ public void test_next_blocksFurtherAccessWhenNoRows()
+ throws Exception
+ {
+ Connection connection = new Driver().connect( "jdbc:drill:zk=local", null );
+ Statement statement = connection.createStatement();
+ ResultSet resultSet =
+ statement.executeQuery( "SELECT 'Hi' AS x \n" +
+ "FROM cp.`donuts.json` \n" +
+ "WHERE false" );
+
+ // Do initial next(). (Advance from before results to next possible
+ // position (after the set of zero rows).
+ assertThat( resultSet.next(), is( false ) );
+
+ // Main check: That row data access methods throw SQLException.
+ try {
+ resultSet.getString( 1 );
+ fail( "Did get expected SQLException." );
+ }
+ catch ( SQLException e ) {
+ // Expect something like current InvalidRowSQLException saying
+ // "Result set cursor is still before all rows. Call next() first."
+ assertThat( e, instanceOf( InvalidCursorStateSqlException.class ) );
+ assertThat( e.toString(), containsString( "before" ) );
+ }
+ // (Any non-SQLException exception is unexpected result.)
+
+ assertThat( resultSet.next(), is( false ) );
+
+ // TODO: Ideally, test all other accessor methods.
+ }
+
+
+ // TODO: Ideally, test other methods.
+
+}