diff --git a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBConnection.java b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBConnection.java
index 2a068f1..ca61992 100644
--- a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBConnection.java
+++ b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBConnection.java
@@ -174,7 +174,7 @@
 
     private void checkClosed() throws SQLException {
         if (isClosed()) {
-            throw getErrorReporter().errorObjectClosed(Connection.class);
+            throw getErrorReporter().errorObjectClosed(Connection.class, ADBErrorReporter.SQLState.CONNECTION_CLOSED);
         }
     }
 
diff --git a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBErrorReporter.java b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBErrorReporter.java
index 07abd87..7009090 100644
--- a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBErrorReporter.java
+++ b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBErrorReporter.java
@@ -40,6 +40,10 @@
         return new SQLException(String.format("%s is closed", jdbcInterface.getSimpleName()));
     }
 
+    public SQLException errorObjectClosed(Class<?> jdbcInterface, SQLState sqlState) {
+        return new SQLException(String.format("%s is closed", jdbcInterface.getSimpleName()), sqlState.code);
+    }
+
     public SQLFeatureNotSupportedException errorMethodNotSupported(Class<?> jdbcInterface, String methodName) {
         return new SQLFeatureNotSupportedException(
                 String.format("Method %s.%s() is not supported", jdbcInterface.getName(), methodName));
@@ -213,6 +217,7 @@
 
     public enum SQLState {
         CONNECTION_FAILURE("08001"), // TODO:08006??
+        CONNECTION_CLOSED("08003"),
         INVALID_AUTH_SPEC("28000"),
         INVALID_DATE_TYPE("HY004"),
         INVALID_CURSOR_POSITION("HY108");
@@ -222,5 +227,10 @@
         SQLState(String code) {
             this.code = Objects.requireNonNull(code);
         }
+
+        @Override
+        public String toString() {
+            return code;
+        }
     }
 }
diff --git a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBParameterMetaData.java b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBParameterMetaData.java
index a42def1..1663db0 100644
--- a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBParameterMetaData.java
+++ b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBParameterMetaData.java
@@ -20,7 +20,6 @@
 package org.apache.asterix.jdbc.core;
 
 import java.sql.ParameterMetaData;
-import java.sql.Types;
 import java.util.Objects;
 
 public class ADBParameterMetaData extends ADBWrapperSupport implements ParameterMetaData {
@@ -46,12 +45,12 @@
 
     @Override
     public int getParameterType(int parameterIndex) {
-        return Types.OTHER; // any
+        return ADBDatatype.ANY.getJdbcType().getVendorTypeNumber();
     }
 
     @Override
     public String getParameterTypeName(int parameterIndex) {
-        return "";
+        return ADBDatatype.ANY.getTypeName();
     }
 
     @Override
diff --git a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBPreparedStatement.java b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBPreparedStatement.java
index f87eede..ff28790 100644
--- a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBPreparedStatement.java
+++ b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBPreparedStatement.java
@@ -336,22 +336,30 @@
 
     @Override
     public void setObject(int parameterIndex, Object v, int targetSqlType) throws SQLException {
-        setObject(parameterIndex, v); // TODO:revisit
+        setObject(parameterIndex, v);
     }
 
     @Override
     public void setObject(int parameterIndex, Object v, SQLType targetSqlType) throws SQLException {
-        setObject(parameterIndex, v, targetSqlType.getVendorTypeNumber()); // TODO:revisit
+        if (targetSqlType == null) {
+            setObject(parameterIndex, v);
+        } else {
+            setObject(parameterIndex, v, targetSqlType.getVendorTypeNumber());
+        }
     }
 
     @Override
     public void setObject(int parameterIndex, Object v, int targetSqlType, int scaleOrLength) throws SQLException {
-        setObject(parameterIndex, v, targetSqlType); // TODO:revisit
+        setObject(parameterIndex, v, targetSqlType);
     }
 
     @Override
     public void setObject(int parameterIndex, Object v, SQLType targetSqlType, int scaleOrLength) throws SQLException {
-        setObject(parameterIndex, v, targetSqlType.getVendorTypeNumber()); // TODO:revisit
+        if (targetSqlType == null) {
+            setObject(parameterIndex, v);
+        } else {
+            setObject(parameterIndex, v, targetSqlType.getVendorTypeNumber());
+        }
     }
 
     // Unsupported
diff --git a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBResultSet.java b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBResultSet.java
index 1d46b4e..2af4585 100644
--- a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBResultSet.java
+++ b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBResultSet.java
@@ -332,17 +332,20 @@
     }
 
     @Override
-    public boolean isBeforeFirst() {
+    public boolean isBeforeFirst() throws SQLException {
+        checkClosed();
         return state == ST_BEFORE_FIRST;
     }
 
     @Override
-    public boolean isAfterLast() {
+    public boolean isAfterLast() throws SQLException {
+        checkClosed();
         return state == ST_AFTER_LAST;
     }
 
     @Override
-    public boolean isFirst() {
+    public boolean isFirst() throws SQLException {
+        checkClosed();
         return state == ST_NEXT && rowNumber == 1;
     }
 
@@ -354,7 +357,7 @@
     @Override
     public int getRow() throws SQLException {
         checkClosed();
-        return (int) rowNumber;
+        return state == ST_NEXT ? (int) rowNumber : 0;
     }
 
     private void checkCursorPosition() throws SQLException {
@@ -862,7 +865,7 @@
     }
 
     private InputStream getAsciiStreamImpl(int columnIndex) throws SQLException {
-        String value = getString(columnIndex);
+        String value = getStringImpl(columnIndex);
         return value != null ? new ByteArrayInputStream(value.getBytes(StandardCharsets.US_ASCII)) : null;
     }
 
@@ -881,7 +884,7 @@
     }
 
     private InputStream getUnicodeStreamImpl(int columnIndex) throws SQLException {
-        String value = getString(columnIndex);
+        String value = getStringImpl(columnIndex);
         return value != null ? new ByteArrayInputStream(value.getBytes(StandardCharsets.UTF_16)) : null;
     }
 
diff --git a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBRowStore.java b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBRowStore.java
index ed6995d..764d3b0 100644
--- a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBRowStore.java
+++ b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBRowStore.java
@@ -979,7 +979,8 @@
     }
 
     private LocalDate toLocalDateFromDatetimeChronon(long datetimeChrononInMillis) {
-        return LocalDate.ofEpochDay(TimeUnit.MILLISECONDS.toDays(datetimeChrononInMillis));
+        // TODO: use LocalDate.ofInstant() in JDK 9+
+        return toLocalDateTimeFromDatetimeChronon(datetimeChrononInMillis).toLocalDate();
     }
 
     private Time toTimeFromTimeChronon(long timeChrononInMillis, TimeZone tz) {
@@ -996,7 +997,8 @@
     }
 
     private LocalTime toLocalTimeFromDatetimeChronon(long datetimeChrononInMillis) {
-        return LocalTime.ofNanoOfDay(TimeUnit.MILLISECONDS.toNanos(datetimeChrononInMillis));
+        // TODO: use LocalTime.ofInstant() in JDK 9+
+        return toLocalDateTimeFromDatetimeChronon(datetimeChrononInMillis).toLocalTime();
     }
 
     private Timestamp toTimestampFromDatetimeChronon(long datetimeChrononInMillis, TimeZone tz) {
diff --git a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBStatement.java b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBStatement.java
index cdbece6..c024f21 100644
--- a/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBStatement.java
+++ b/asterixdb-jdbc/asterix-jdbc-core/src/main/java/org/apache/asterix/jdbc/core/ADBStatement.java
@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.lang.ref.Reference;
 import java.lang.ref.WeakReference;
+import java.math.BigDecimal;
 import java.sql.Connection;
 import java.sql.Date;
 import java.sql.ResultSet;
@@ -692,6 +693,7 @@
         // Long is serialized as JSON number by Jackson
         registerSerializer(serializerMap, createFloatSerializer());
         registerSerializer(serializerMap, createDoubleSerializer());
+        registerSerializer(serializerMap, createBigDecimalSerializer());
         registerSerializer(serializerMap, createStringSerializer());
         registerSerializer(serializerMap, createSqlDateSerializer());
         registerSerializer(serializerMap, createSqlDateWithCalendarSerializer());
@@ -755,6 +757,16 @@
         };
     }
 
+    protected static ATaggedValueSerializer createBigDecimalSerializer() {
+        return new ATaggedValueSerializer(BigDecimal.class, ADBDatatype.DOUBLE) {
+            @Override
+            protected void serializeNonTaggedValue(Object value, StringBuilder out) {
+                long bits = Double.doubleToLongBits(((BigDecimal) value).doubleValue());
+                out.append(bits);
+            }
+        };
+    }
+
     protected static ATaggedValueSerializer createSqlDateSerializer() {
         return new ATaggedValueSerializer(java.sql.Date.class, ADBDatatype.DATE) {
             @Override
diff --git a/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/ADBProtocol.java b/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/ADBProtocol.java
index bb477b0..53ab47c 100644
--- a/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/ADBProtocol.java
+++ b/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/ADBProtocol.java
@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -87,6 +88,8 @@
     private static final String QUERY_RESULT_ENDPOINT_PATH = "/query/service/result";
     private static final String ACTIVE_REQUESTS_ENDPOINT_PATH = "/admin/requests/running";
 
+    private static final int CONNECTION_REQUEST_TIMEOUT = 50; // ms
+
     static final List<Class<? extends IOException>> TIMEOUT_CONNECTION_ERRORS =
             Collections.singletonList(ConnectTimeoutException.class);
 
@@ -100,8 +103,8 @@
     final HttpClientContext httpClientContext;
     final CloseableHttpClient httpClient;
 
-    public ADBProtocol(String host, int port, Map<ADBDriverProperty, Object> params, ADBDriverContext driverContext)
-            throws SQLException {
+    public ADBProtocol(String host, int port, Map<ADBDriverProperty, Object> params, ADBDriverContext driverContext,
+            int loginTimeoutSeconds) throws SQLException {
         super(driverContext, params);
         boolean sslEnabled = (Boolean) ADBDriverProperty.Common.SSL.fetchPropertyValue(params);
         URI queryEndpoint = createEndpointUri(sslEnabled, host, port, QUERY_SERVICE_ENDPOINT_PATH,
@@ -126,13 +129,14 @@
         }
         RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
         Number connectTimeoutMillis = (Number) ADBDriverProperty.Common.CONNECT_TIMEOUT.fetchPropertyValue(params);
-        if (connectTimeoutMillis != null) {
-            requestConfigBuilder.setConnectionRequestTimeout(connectTimeoutMillis.intValue());
-            requestConfigBuilder.setConnectTimeout(connectTimeoutMillis.intValue());
-        }
+        int connectTimeout = Math.max(0, connectTimeoutMillis != null ? connectTimeoutMillis.intValue()
+                : (int) TimeUnit.SECONDS.toMillis(loginTimeoutSeconds));
+        requestConfigBuilder.setConnectTimeout(connectTimeout);
         if (socketTimeoutMillis != null) {
             requestConfigBuilder.setSocketTimeout(socketTimeoutMillis.intValue());
         }
+        requestConfigBuilder.setConnectionRequestTimeout(CONNECTION_REQUEST_TIMEOUT);
+
         RequestConfig requestConfig = requestConfigBuilder.build();
         HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
         httpClientBuilder.setConnectionManager(httpConnectionManager);
diff --git a/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/Driver.java b/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/Driver.java
index ac1701f..65e4af0 100644
--- a/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/Driver.java
+++ b/asterixdb-jdbc/asterix-jdbc-driver/src/main/java/org/apache/asterix/jdbc/Driver.java
@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Map;
@@ -53,7 +54,8 @@
     @Override
     protected ADBProtocol createProtocol(String host, int port, Map<ADBDriverProperty, Object> properties,
             ADBDriverContext driverContext) throws SQLException {
-        return new ADBProtocol(host, port, properties, driverContext);
+        int loginTimeoutSeconds = DriverManager.getLoginTimeout();
+        return new ADBProtocol(host, port, properties, driverContext, loginTimeoutSeconds);
     }
 
     @Override
diff --git a/asterixdb-jdbc/asterix-jdbc-test/pom.xml b/asterixdb-jdbc/asterix-jdbc-test/pom.xml
new file mode 100644
index 0000000..c294c48
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/pom.xml
@@ -0,0 +1,101 @@
+<!--
+ ! 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.
+ !-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <artifactId>asterix-jdbc-test</artifactId>
+  <version>0.9.8-SNAPSHOT</version>
+  <packaging>jar</packaging>
+  <parent>
+    <groupId>org.apache.asterix</groupId>
+    <artifactId>apache-asterixdb</artifactId>
+    <version>0.9.8-SNAPSHOT</version>
+    <relativePath>../../../../asterixdb/pom.xml</relativePath> <!-- asterixdb/pom.xml -->
+  </parent>
+
+  <properties>
+    <root.dir>${basedir}/..</root.dir>
+    <asterix-app.dir>${root.dir}/../../asterix-app</asterix-app.dir>
+    <testLog4jConfigFile>${asterix-app.dir}/src/test/resources/log4j2-asterixdb-test.xml</testLog4jConfigFile>
+  </properties>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <systemPropertyVariables combine.children="append">
+            <asterix-app.dir>${asterix-app.dir}</asterix-app.dir>
+          </systemPropertyVariables>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-failsafe-plugin</artifactId>
+        <configuration>
+          <systemPropertyVariables combine.children="append">
+            <asterix-app.dir>${asterix-app.dir}</asterix-app.dir>
+          </systemPropertyVariables>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.asterix</groupId>
+      <artifactId>asterix-jdbc-driver</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.hyracks</groupId>
+      <artifactId>hyracks-storage-am-lsm-btree-test</artifactId>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.asterix</groupId>
+      <artifactId>asterix-test-framework</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.asterix</groupId>
+      <artifactId>asterix-app</artifactId>
+      <version>${project.version}</version>
+      <type>jar</type>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.asterix</groupId>
+      <artifactId>asterix-app</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>
+
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcConnectionTester.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcConnectionTester.java
new file mode 100644
index 0000000..00cfaea
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcConnectionTester.java
@@ -0,0 +1,130 @@
+/*
+ * 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.asterix.test.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.asterix.jdbc.Driver;
+import org.apache.asterix.jdbc.core.ADBConnection;
+import org.junit.Assert;
+
+class JdbcConnectionTester extends JdbcTester {
+
+    public void testGetConnectionViaDriverManager() throws SQLException {
+        DriverManager.getConnection(testContext.getJdbcUrl()).close();
+        DriverManager.getConnection(testContext.getJdbcUrl(), null).close();
+        DriverManager.getConnection(testContext.getJdbcUrl(), new Properties()).close();
+        DriverManager.getConnection(testContext.getJdbcUrl(), null, null).close();
+    }
+
+    public void testGetConnectionDirect() throws SQLException {
+        Driver driver = new Driver();
+        driver.connect(testContext.getJdbcUrl(), null).close();
+        driver.connect(testContext.getJdbcUrl(), new Properties()).close();
+    }
+
+    public void testLifecycle() throws SQLException {
+        Connection c = createConnection();
+        Assert.assertNull(c.getWarnings());
+        Assert.assertTrue(c.isValid( /*timeout in seconds*/ 30));
+        Assert.assertFalse(c.isClosed());
+
+        c.close();
+        Assert.assertTrue(c.isClosed());
+
+        // ok to call close() on a closed connection
+        c.close();
+        Assert.assertTrue(c.isClosed());
+
+        // ok to call isValid() on a closed connection
+        Assert.assertFalse(c.isValid(0));
+
+        // errors on a closed connection
+        assertErrorOnClosed(c, Connection::clearWarnings, "clearWarnings");
+        assertErrorOnClosed(c, Connection::createStatement, "createStatement");
+        assertErrorOnClosed(c, Connection::getAutoCommit, "getAutoCommit");
+        assertErrorOnClosed(c, Connection::getCatalog, "getCatalog");
+        assertErrorOnClosed(c, Connection::getClientInfo, "getClientInfo");
+        assertErrorOnClosed(c, Connection::getHoldability, "getHoldability");
+        assertErrorOnClosed(c, Connection::getMetaData, "getMetadata");
+        assertErrorOnClosed(c, Connection::getSchema, "getSchema");
+        assertErrorOnClosed(c, Connection::getTransactionIsolation, "getTransactionIsolation");
+        assertErrorOnClosed(c, Connection::getWarnings, "getWarnings");
+        assertErrorOnClosed(c, Connection::getTypeMap, "getTypeMap");
+        assertErrorOnClosed(c, Connection::isReadOnly, "isReadOnly");
+        assertErrorOnClosed(c, ci -> ci.prepareStatement("select 1"), "prepareStatement");
+    }
+
+    public void testCatalogSchema() throws SQLException {
+        try (Connection c = createConnection()) {
+            Assert.assertEquals(DEFAULT_DATAVERSE_NAME, c.getCatalog());
+            Assert.assertNull(c.getSchema());
+        }
+
+        try (Connection c = createConnection(METADATA_DATAVERSE_NAME)) {
+            Assert.assertEquals(METADATA_DATAVERSE_NAME, c.getCatalog());
+            Assert.assertNull(c.getSchema());
+        }
+
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testCatalogSchema");
+            String dvCanon = getCanonicalDataverseName(dataverse);
+            String dataset = "ds1";
+            s.execute(printCreateDataverse(dataverse));
+            s.execute(printCreateDataset(dataverse, dataset));
+            s.execute(printInsert(dataverse, dataset, dataGen("x", 1, 2, 3)));
+            try (Connection c2 = createConnection(dvCanon); Statement s2 = c2.createStatement()) {
+                Assert.assertEquals(dvCanon, c2.getCatalog());
+                Assert.assertNull(c.getSchema());
+                try (ResultSet rs2 =
+                        s2.executeQuery(String.format("select count(*) from %s", printIdentifier(dataset)))) {
+                    Assert.assertTrue(rs2.next());
+                    Assert.assertEquals(3, rs2.getInt(1));
+                }
+            } finally {
+                s.execute(printDropDataverse(dataverse));
+            }
+        }
+    }
+
+    // Connection.setReadOnly() hint is currently ignored
+    // Connection.isReadOnly() always returns 'false'
+    public void testReadOnlyMode() throws SQLException {
+        try (Connection c = createConnection()) {
+            Assert.assertFalse(c.isReadOnly());
+            c.setReadOnly(true);
+            Assert.assertFalse(c.isReadOnly());
+        }
+    }
+
+    public void testWrapper() throws SQLException {
+        try (Connection c = createConnection()) {
+            Assert.assertTrue(c.isWrapperFor(ADBConnection.class));
+            Assert.assertNotNull(c.unwrap(ADBConnection.class));
+        }
+    }
+}
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcDriverTest.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcDriverTest.java
new file mode 100644
index 0000000..b763de2
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcDriverTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.asterix.test.jdbc;
+
+import java.lang.reflect.Method;
+import java.net.InetAddress;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.asterix.test.runtime.ExecutionTestUtil;
+import org.apache.hyracks.control.nc.NodeControllerService;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class JdbcDriverTest {
+
+    static final String ASTERIX_APP_PATH_PROPERTY = "asterix-app.dir";
+
+    static final String ASTERIX_APP_PATH = System.getProperty(ASTERIX_APP_PATH_PROPERTY);
+
+    static final List<Class<? extends JdbcTester>> TESTER_CLASSES =
+            Arrays.asList(JdbcMetadataTester.class, JdbcConnectionTester.class, JdbcStatementTester.class,
+                    JdbcPreparedStatementTester.class, JdbcResultSetTester.JdbcStatementResultSetTester.class,
+                    JdbcResultSetTester.JdbcPreparedStatementResultSetTester.class, JdbcStatementParameterTester.class);
+
+    public static final String TEST_METHOD_PREFIX = "test";
+
+    private static JdbcTester.JdbcTestContext testContext;
+
+    private final Class<? extends JdbcTester> testerClass;
+
+    private final Method testMethod;
+
+    public JdbcDriverTest(String simpleClassName, String methodName) throws Exception {
+        Optional<Class<? extends JdbcTester>> testerClassRef =
+                TESTER_CLASSES.stream().filter(c -> c.getSimpleName().equals(simpleClassName)).findFirst();
+        if (testerClassRef.isEmpty()) {
+            throw new Exception("Cannot find class: " + simpleClassName);
+        }
+        testerClass = testerClassRef.get();
+        Optional<Method> testMethodRef = Arrays.stream(testerClassRef.get().getMethods())
+                .filter(m -> m.getName().equals(methodName)).findFirst();
+        if (testMethodRef.isEmpty()) {
+            throw new Exception("Cannot find method: " + methodName + " in class " + testerClass.getName());
+        }
+        testMethod = testMethodRef.get();
+    }
+
+    @Parameterized.Parameters(name = "JdbcDriverTest {index}: {0}.{1}")
+    public static Collection<Object[]> tests() {
+        List<Object[]> testsuite = new ArrayList<>();
+        for (Class<? extends JdbcTester> testerClass : TESTER_CLASSES) {
+            Arrays.stream(testerClass.getMethods()).map(Method::getName).filter(n -> n.startsWith(TEST_METHOD_PREFIX))
+                    .sorted().forEach(n -> testsuite.add(new Object[] { testerClass.getSimpleName(), n }));
+        }
+        return testsuite;
+    }
+
+    @BeforeClass
+    public static void setUp() throws Exception {
+        if (ASTERIX_APP_PATH == null) {
+            throw new Exception(String.format("Property %s is not set", ASTERIX_APP_PATH_PROPERTY));
+        }
+
+        Path ccConfigFile = Path.of(ASTERIX_APP_PATH, "src", TEST_METHOD_PREFIX, "resources", "cc.conf");
+
+        ExecutionTestUtil.setUp(true, ccConfigFile.toString(), ExecutionTestUtil.integrationUtil, false,
+                Collections.emptyList());
+
+        NodeControllerService nc = ExecutionTestUtil.integrationUtil.ncs[0];
+        String host = InetAddress.getLoopbackAddress().getHostAddress();
+        INcApplicationContext appCtx = (INcApplicationContext) nc.getApplicationContext();
+        int apiPort = appCtx.getExternalProperties().getNcApiPort();
+
+        testContext = JdbcTester.createTestContext(host, apiPort);
+    }
+
+    @AfterClass
+    public static void tearDown() throws Exception {
+        ExecutionTestUtil.tearDown(true, false);
+    }
+
+    @Test
+    public void test() throws Exception {
+        JdbcTester tester = testerClass.getDeclaredConstructor().newInstance();
+        tester.setTestContext(testContext);
+        testMethod.invoke(tester);
+    }
+}
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcMetadataTester.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcMetadataTester.java
new file mode 100644
index 0000000..80199fe
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcMetadataTester.java
@@ -0,0 +1,1092 @@
+/*
+ * 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.asterix.test.jdbc;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.JDBCType;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLType;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.asterix.jdbc.core.ADBDatabaseMetaData;
+import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.junit.Assert;
+
+class JdbcMetadataTester extends JdbcTester {
+
+    static final String TABLE_CAT = "TABLE_CAT";
+    static final String TABLE_CATALOG = "TABLE_CATALOG";
+    static final String TABLE_SCHEM = "TABLE_SCHEM";
+    static final String TABLE_NAME = "TABLE_NAME";
+    static final String TABLE_TYPE = "TABLE_TYPE";
+    static final String TABLE = "TABLE";
+    static final String VIEW = "VIEW";
+    static final String COLUMN_NAME = "COLUMN_NAME";
+    static final String DATA_TYPE = "DATA_TYPE";
+    static final String TYPE_NAME = "TYPE_NAME";
+    static final String ORDINAL_POSITION = "ORDINAL_POSITION";
+    static final String NULLABLE = "NULLABLE";
+    static final String KEY_SEQ = "KEY_SEQ";
+    static final String PKTABLE_CAT = "PKTABLE_CAT";
+    static final String PKTABLE_SCHEM = "PKTABLE_SCHEM";
+    static final String PKTABLE_NAME = "PKTABLE_NAME";
+    static final String PKCOLUMN_NAME = "PKCOLUMN_NAME";
+    static final String FKTABLE_CAT = "FKTABLE_CAT";
+    static final String FKTABLE_SCHEM = "FKTABLE_SCHEM";
+    static final String FKTABLE_NAME = "FKTABLE_NAME";
+    static final String FKCOLUMN_NAME = "FKCOLUMN_NAME";
+
+    static final String STRING = "string";
+    static final String BIGINT = "int64";
+    static final String DOUBLE = "double";
+
+    static final List<String> DATASET_COLUMN_NAMES = Arrays.asList("tc", "ta", "tb");
+    static final List<String> VIEW_COLUMN_NAMES = Arrays.asList("vb", "vc", "va");
+    static final List<String> DATASET_COLUMN_TYPES = Arrays.asList(STRING, BIGINT, DOUBLE);
+    static final List<SQLType> DATASET_COLUMN_JDBC_TYPES =
+            Arrays.asList(JDBCType.VARCHAR, JDBCType.BIGINT, JDBCType.DOUBLE);
+    static final int DATASET_PK_LEN = 2;
+
+    public void testLifecycle() throws SQLException {
+        Connection c = createConnection();
+        Assert.assertSame(c, c.getMetaData().getConnection());
+        c.close();
+        try {
+            c.getMetaData();
+            Assert.fail("Got metadata on a closed connection");
+        } catch (SQLException e) {
+            Assert.assertEquals(SQL_STATE_CONNECTION_CLOSED, e.getSQLState());
+        }
+    }
+
+    public void testProperties() throws SQLException {
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            Assert.assertEquals(testContext.getJdbcUrl(), md.getURL());
+            Assert.assertNotNull(md.getDriverName());
+            Assert.assertNotNull(md.getDriverVersion());
+            Assert.assertNotNull(md.getDatabaseProductName());
+            Assert.assertNotNull(md.getDatabaseProductVersion());
+            Assert.assertEquals(4, md.getJDBCMajorVersion());
+            Assert.assertEquals(2, md.getJDBCMinorVersion());
+            Assert.assertTrue(md.isCatalogAtStart());
+            Assert.assertEquals(".", md.getCatalogSeparator());
+            Assert.assertEquals("`", md.getIdentifierQuoteString());
+            Assert.assertTrue(md.allTablesAreSelectable());
+            Assert.assertTrue(md.nullsAreSortedLow());
+            Assert.assertFalse(md.nullsAreSortedHigh());
+            Assert.assertFalse(md.nullsAreSortedAtStart());
+            Assert.assertFalse(md.nullsAreSortedAtEnd());
+            Assert.assertFalse(md.supportsCatalogsInTableDefinitions());
+            Assert.assertFalse(md.supportsCatalogsInIndexDefinitions());
+            Assert.assertFalse(md.supportsCatalogsInDataManipulation());
+            Assert.assertFalse(md.supportsSchemasInTableDefinitions());
+            Assert.assertFalse(md.supportsSchemasInIndexDefinitions());
+            Assert.assertFalse(md.supportsSchemasInDataManipulation());
+            Assert.assertTrue(md.supportsSubqueriesInComparisons());
+            Assert.assertTrue(md.supportsSubqueriesInExists());
+            Assert.assertTrue(md.supportsSubqueriesInIns());
+            Assert.assertTrue(md.supportsCorrelatedSubqueries());
+            Assert.assertTrue(md.supportsOrderByUnrelated());
+            Assert.assertTrue(md.supportsExpressionsInOrderBy());
+            Assert.assertTrue(md.supportsGroupBy());
+            Assert.assertTrue(md.supportsGroupByUnrelated());
+            Assert.assertTrue(md.supportsGroupByBeyondSelect());
+            Assert.assertTrue(md.supportsOuterJoins());
+            Assert.assertTrue(md.supportsMinimumSQLGrammar());
+            Assert.assertTrue(md.supportsTableCorrelationNames());
+            Assert.assertTrue(md.supportsUnionAll());
+        }
+    }
+
+    public void testGetCatalogs() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getCatalogs()) {
+                assertColumnValues(rs, TABLE_CAT, BUILT_IN_DATAVERSE_NAMES);
+            }
+            List<List<String>> newDataverseList = new ArrayList<>();
+            try {
+                createDataverses(s, newDataverseList);
+
+                List<String> allCatalogs = new ArrayList<>(BUILT_IN_DATAVERSE_NAMES);
+                for (List<String> n : newDataverseList) {
+                    allCatalogs.add(getCanonicalDataverseName(n));
+                }
+                try (ResultSet rs = md.getCatalogs()) {
+                    assertColumnValues(rs, TABLE_CAT, allCatalogs);
+                }
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    public void testGetCatalogsResultSetLifecycle() throws SQLException {
+        // check that Connection.close() closes metadata ResultSet
+        Connection c = createConnection();
+        DatabaseMetaData md = c.getMetaData();
+        ResultSet rs = md.getCatalogs();
+        Assert.assertFalse(rs.isClosed());
+        c.close();
+        Assert.assertTrue(rs.isClosed());
+    }
+
+    public void testGetSchemas() throws SQLException {
+        // get schemas in the default dataverse
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getSchemas()) {
+                assertColumnValues(rs, Arrays.asList(TABLE_SCHEM, TABLE_CATALOG), Arrays
+                        .asList(Collections.singletonList(null), Collections.singletonList(DEFAULT_DATAVERSE_NAME)));
+            }
+        }
+
+        // get schemas in the connection's dataverse
+        try (Connection c = createConnection(METADATA_DATAVERSE_NAME)) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getSchemas()) {
+                assertColumnValues(rs, Arrays.asList(TABLE_SCHEM, TABLE_CATALOG), Arrays
+                        .asList(Collections.singletonList(null), Collections.singletonList(METADATA_DATAVERSE_NAME)));
+            }
+        }
+
+        // get schemas in the connection's dataverse #2
+        try (Connection c = createConnection()) {
+            c.setCatalog(METADATA_DATAVERSE_NAME);
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getSchemas()) {
+                assertColumnValues(rs, Arrays.asList(TABLE_SCHEM, TABLE_CATALOG), Arrays
+                        .asList(Collections.singletonList(null), Collections.singletonList(METADATA_DATAVERSE_NAME)));
+            }
+        }
+
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            // we don't have any schemas without catalogs
+            try (ResultSet rs = md.getSchemas("", null)) {
+                Assert.assertEquals(0, countRows(rs));
+            }
+        }
+
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<List<String>> newDataverseList = new ArrayList<>();
+            try {
+                createDataverses(s, newDataverseList);
+
+                List<String> allCatalogs = new ArrayList<>(BUILT_IN_DATAVERSE_NAMES);
+                for (List<String> n : newDataverseList) {
+                    allCatalogs.add(getCanonicalDataverseName(n));
+                }
+                DatabaseMetaData md = c.getMetaData();
+                try (ResultSet rs = md.getSchemas("", null)) {
+                    Assert.assertFalse(rs.next());
+                }
+                try (ResultSet rs = md.getSchemas(null, null)) {
+                    assertColumnValues(rs, Arrays.asList(TABLE_SCHEM, TABLE_CATALOG),
+                            Arrays.asList(Collections.nCopies(allCatalogs.size(), null), allCatalogs));
+                }
+                try (ResultSet rs = md.getSchemas("x", null)) {
+                    assertColumnValues(rs, Arrays.asList(TABLE_SCHEM, TABLE_CATALOG),
+                            Arrays.asList(Collections.singletonList(null), Collections.singletonList("x")));
+                }
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    public void testGetTableTypes() throws SQLException {
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getTableTypes()) {
+                assertColumnValues(rs, TABLE_TYPE, Arrays.asList(TABLE, VIEW));
+            }
+        }
+    }
+
+    public void testGetTables() throws SQLException {
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getTables(METADATA_DATAVERSE_NAME, null, null, null)) {
+                int n = countRows(rs);
+                Assert.assertTrue(String.valueOf(n), n > 10);
+            }
+            try (ResultSet rs = md.getTables(METADATA_DATAVERSE_NAME, null, "Data%", null)) {
+                int n = countRows(rs);
+                Assert.assertTrue(String.valueOf(n), n > 2);
+            }
+            // we don't have any tables without catalogs
+            try (ResultSet rs = md.getTables("", null, null, null)) {
+                int n = countRows(rs);
+                Assert.assertEquals(0, n);
+            }
+        }
+
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<List<String>> newDataverseList = new ArrayList<>();
+            List<Pair<List<String>, String>> newDatasetList = new ArrayList<>();
+            List<Pair<List<String>, String>> newViewList = new ArrayList<>();
+
+            try {
+                createDataversesDatasetsViews(s, newDataverseList, newDatasetList, newViewList);
+
+                DatabaseMetaData md = c.getMetaData();
+                List<String> expectedColumns = Arrays.asList(TABLE_CAT, TABLE_SCHEM, TABLE_NAME, TABLE_TYPE);
+                List<String> expectedTableCat = new ArrayList<>();
+                List<String> expectedTableSchem = new ArrayList<>();
+                List<String> expectedTableName = new ArrayList<>();
+                List<String> expectedTableType = new ArrayList<>();
+
+                // Test getTables() in all catalogs
+                for (Pair<List<String>, String> p : newDatasetList) {
+                    expectedTableCat.add(getCanonicalDataverseName(p.first));
+                    expectedTableSchem.add(null);
+                    expectedTableName.add(p.second);
+                    expectedTableType.add(TABLE);
+                }
+                // using table name pattern
+                try (ResultSet rs = md.getTables(null, null, "t%", null)) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedTableType));
+                }
+                // using table type
+                try (ResultSet rs = md.getTables(null, null, null, new String[] { TABLE })) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedTableType),
+                            JdbcMetadataTester::isMetadataCatalog);
+                }
+                // all tables
+                for (Pair<List<String>, String> p : newViewList) {
+                    expectedTableCat.add(getCanonicalDataverseName(p.first));
+                    expectedTableSchem.add(null);
+                    expectedTableName.add(p.second);
+                    expectedTableType.add(VIEW);
+                }
+                try (ResultSet rs = md.getTables(null, null, null, null)) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedTableType),
+                            JdbcMetadataTester::isMetadataCatalog);
+                }
+                try (ResultSet rs = md.getTables(null, "", null, null)) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedTableType),
+                            JdbcMetadataTester::isMetadataCatalog);
+                }
+                try (ResultSet rs = md.getTables(null, null, null, new String[] { TABLE, VIEW })) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedTableType),
+                            JdbcMetadataTester::isMetadataCatalog);
+                }
+
+                // Test getTables() in a particular catalog
+                for (List<String> dvi : newDataverseList) {
+                    expectedTableCat.clear();
+                    expectedTableSchem.clear();
+                    expectedTableName.clear();
+                    expectedTableType.clear();
+                    String dvic = getCanonicalDataverseName(dvi);
+                    for (Pair<List<String>, String> p : newDatasetList) {
+                        String dv = getCanonicalDataverseName(p.first);
+                        if (dv.equals(dvic)) {
+                            expectedTableCat.add(dv);
+                            expectedTableSchem.add(null);
+                            expectedTableName.add(p.second);
+                            expectedTableType.add(TABLE);
+                        }
+                    }
+                    // using table name pattern
+                    try (ResultSet rs = md.getTables(dvic, null, "t%", null)) {
+                        assertColumnValues(rs, expectedColumns, Arrays.asList(expectedTableCat, expectedTableSchem,
+                                expectedTableName, expectedTableType));
+                    }
+                    // using table type
+                    try (ResultSet rs = md.getTables(dvic, null, null, new String[] { TABLE })) {
+                        assertColumnValues(rs, expectedColumns, Arrays.asList(expectedTableCat, expectedTableSchem,
+                                expectedTableName, expectedTableType));
+                    }
+                    for (Pair<List<String>, String> p : newViewList) {
+                        String dv = getCanonicalDataverseName(p.first);
+                        if (dv.equals(dvic)) {
+                            expectedTableCat.add(dv);
+                            expectedTableSchem.add(null);
+                            expectedTableName.add(p.second);
+                            expectedTableType.add(VIEW);
+                        }
+                    }
+                    try (ResultSet rs = md.getTables(dvic, null, null, null)) {
+                        assertColumnValues(rs, expectedColumns, Arrays.asList(expectedTableCat, expectedTableSchem,
+                                expectedTableName, expectedTableType));
+                    }
+                    try (ResultSet rs = md.getTables(dvic, "", null, null)) {
+                        assertColumnValues(rs, expectedColumns, Arrays.asList(expectedTableCat, expectedTableSchem,
+                                expectedTableName, expectedTableType));
+                    }
+                    try (ResultSet rs = md.getTables(dvic, null, null, new String[] { TABLE, VIEW })) {
+                        assertColumnValues(rs, expectedColumns, Arrays.asList(expectedTableCat, expectedTableSchem,
+                                expectedTableName, expectedTableType));
+                    }
+                }
+
+                // non-existent catalog
+                try (ResultSet rs = md.getTables("UNKNOWN", null, null, null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent schema
+                try (ResultSet rs = md.getTables(null, "UNKNOWN", null, null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent table name
+                try (ResultSet rs = md.getTables(null, null, "UNKNOWN", null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent table type
+                try (ResultSet rs = md.getTables(null, null, null, new String[] { "UNKNOWN" })) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    public void testGetColumns() throws SQLException {
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getColumns(METADATA_DATAVERSE_NAME, null, null, null)) {
+                int n = countRows(rs);
+                Assert.assertTrue(String.valueOf(n), n > 50);
+            }
+            try (ResultSet rs = md.getColumns(METADATA_DATAVERSE_NAME, null, "Data%", null)) {
+                int n = countRows(rs);
+                Assert.assertTrue(String.valueOf(n), n > 20);
+            }
+            // we don't have any columns without catalogs
+            try (ResultSet rs = md.getColumns("", null, null, null)) {
+                int n = countRows(rs);
+                Assert.assertEquals(0, n);
+            }
+        }
+
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<List<String>> newDataverseList = new ArrayList<>();
+            List<Pair<List<String>, String>> newDatasetList = new ArrayList<>();
+            List<Pair<List<String>, String>> newViewList = new ArrayList<>();
+
+            try {
+                createDataversesDatasetsViews(s, newDataverseList, newDatasetList, newViewList);
+
+                DatabaseMetaData md = c.getMetaData();
+
+                List<String> expectedColumns = Arrays.asList(TABLE_CAT, TABLE_SCHEM, TABLE_NAME, COLUMN_NAME, DATA_TYPE,
+                        TYPE_NAME, ORDINAL_POSITION, NULLABLE);
+                List<String> expectedTableCat = new ArrayList<>();
+                List<String> expectedTableSchem = new ArrayList<>();
+                List<String> expectedTableName = new ArrayList<>();
+                List<String> expectedColumnName = new ArrayList<>();
+                List<Integer> expectedDataType = new ArrayList<>();
+                List<String> expectedTypeName = new ArrayList<>();
+                List<Integer> expectedOrdinalPosition = new ArrayList<>();
+                List<Integer> expectedNullable = new ArrayList<>();
+
+                // Test getColumns() in all catalogs
+
+                // datasets only
+                for (Pair<List<String>, String> p : newDatasetList) {
+                    for (int i = 0, n = DATASET_COLUMN_NAMES.size(); i < n; i++) {
+                        String columnName = DATASET_COLUMN_NAMES.get(i);
+                        String columnType = DATASET_COLUMN_TYPES.get(i);
+                        SQLType columnJdbcType = DATASET_COLUMN_JDBC_TYPES.get(i);
+                        expectedTableCat.add(getCanonicalDataverseName(p.first));
+                        expectedTableSchem.add(null);
+                        expectedTableName.add(p.second);
+                        expectedColumnName.add(columnName);
+                        expectedDataType.add(columnJdbcType.getVendorTypeNumber());
+                        expectedTypeName.add(columnType);
+                        expectedOrdinalPosition.add(i + 1);
+                        expectedNullable.add(
+                                i < DATASET_PK_LEN ? DatabaseMetaData.columnNoNulls : DatabaseMetaData.columnNullable);
+                    }
+                }
+                // using column name pattern
+                try (ResultSet rs = md.getColumns(null, null, null, "t%")) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedColumnName,
+                                    expectedDataType, expectedTypeName, expectedOrdinalPosition, expectedNullable));
+                }
+                // using table name pattern
+                try (ResultSet rs = md.getColumns(null, null, "t%", null)) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedColumnName,
+                                    expectedDataType, expectedTypeName, expectedOrdinalPosition, expectedNullable));
+                }
+                // all columns
+                expectedTableCat.clear();
+                expectedTableSchem.clear();
+                expectedTableName.clear();
+                expectedColumnName.clear();
+                expectedDataType.clear();
+                expectedTypeName.clear();
+                expectedOrdinalPosition.clear();
+                expectedNullable.clear();
+
+                int dsIdx = 0, vIdx = 0;
+                for (List<String> dvName : newDataverseList) {
+                    String dvNameCanonical = getCanonicalDataverseName(dvName);
+                    for (; dsIdx < newDatasetList.size() && newDatasetList.get(dsIdx).first.equals(dvName); dsIdx++) {
+                        String dsName = newDatasetList.get(dsIdx).second;
+                        addExpectedColumnNamesForGetColumns(dvNameCanonical, dsName, DATASET_COLUMN_NAMES,
+                                expectedTableCat, expectedTableSchem, expectedTableName, expectedColumnName,
+                                expectedDataType, expectedTypeName, expectedOrdinalPosition, expectedNullable);
+                    }
+                    for (; vIdx < newViewList.size() && newViewList.get(vIdx).first.equals(dvName); vIdx++) {
+                        String vName = newViewList.get(vIdx).second;
+                        addExpectedColumnNamesForGetColumns(dvNameCanonical, vName, VIEW_COLUMN_NAMES, expectedTableCat,
+                                expectedTableSchem, expectedTableName, expectedColumnName, expectedDataType,
+                                expectedTypeName, expectedOrdinalPosition, expectedNullable);
+                    }
+                }
+
+                try (ResultSet rs = md.getColumns(null, null, null, null)) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedColumnName,
+                                    expectedDataType, expectedTypeName, expectedOrdinalPosition, expectedNullable),
+                            JdbcMetadataTester::isMetadataCatalog);
+                }
+                try (ResultSet rs = md.getColumns(null, "", null, null)) {
+                    assertColumnValues(rs, expectedColumns,
+                            Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName, expectedColumnName,
+                                    expectedDataType, expectedTypeName, expectedOrdinalPosition, expectedNullable),
+                            JdbcMetadataTester::isMetadataCatalog);
+                }
+
+                // Test getColumns() in a particular catalog
+                for (List<String> dvName : newDataverseList) {
+                    expectedTableCat.clear();
+                    expectedTableSchem.clear();
+                    expectedTableName.clear();
+                    expectedColumnName.clear();
+                    expectedDataType.clear();
+                    expectedTypeName.clear();
+                    expectedOrdinalPosition.clear();
+                    expectedNullable.clear();
+
+                    String dvNameCanonical = getCanonicalDataverseName(dvName);
+                    for (Pair<List<String>, String> p : newDatasetList) {
+                        if (dvName.equals(p.first)) {
+                            addExpectedColumnNamesForGetColumns(dvNameCanonical, p.second, DATASET_COLUMN_NAMES,
+                                    expectedTableCat, expectedTableSchem, expectedTableName, expectedColumnName,
+                                    expectedDataType, expectedTypeName, expectedOrdinalPosition, expectedNullable);
+                        }
+                    }
+                    try (ResultSet rs = md.getColumns(dvNameCanonical, null, "t%", null)) {
+                        assertColumnValues(rs, expectedColumns,
+                                Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName,
+                                        expectedColumnName, expectedDataType, expectedTypeName, expectedOrdinalPosition,
+                                        expectedNullable),
+                                JdbcMetadataTester::isMetadataCatalog);
+                    }
+                    try (ResultSet rs = md.getColumns(dvNameCanonical, null, null, "t%")) {
+                        assertColumnValues(rs, expectedColumns,
+                                Arrays.asList(expectedTableCat, expectedTableSchem, expectedTableName,
+                                        expectedColumnName, expectedDataType, expectedTypeName, expectedOrdinalPosition,
+                                        expectedNullable),
+                                JdbcMetadataTester::isMetadataCatalog);
+                    }
+                }
+
+                // non-existent catalog
+                try (ResultSet rs = md.getColumns("UNKNOWN", null, null, null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent schema
+                try (ResultSet rs = md.getColumns(null, "UNKNOWN", null, null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent table name
+                try (ResultSet rs = md.getColumns(null, null, "UNKNOWN", null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent column names
+                try (ResultSet rs = md.getColumns(null, null, null, "UNKNOWN")) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    private void addExpectedColumnNamesForGetColumns(String dvNameCanonical, String dsName, List<String> columnNames,
+            List<String> outTableCat, List<String> outTableSchem, List<String> outTableName, List<String> outColumnName,
+            List<Integer> outDataType, List<String> outTypeName, List<Integer> outOrdinalPosition,
+            List<Integer> outNullable) {
+        for (int i = 0; i < columnNames.size(); i++) {
+            String columnName = columnNames.get(i);
+            String columnType = DATASET_COLUMN_TYPES.get(i);
+            SQLType columnJdbcType = DATASET_COLUMN_JDBC_TYPES.get(i);
+            outTableCat.add(dvNameCanonical);
+            outTableSchem.add(null);
+            outTableName.add(dsName);
+            outColumnName.add(columnName);
+            outDataType.add(columnJdbcType.getVendorTypeNumber());
+            outTypeName.add(columnType);
+            outOrdinalPosition.add(i + 1);
+            outNullable.add(i < JdbcMetadataTester.DATASET_PK_LEN ? DatabaseMetaData.columnNoNulls
+                    : DatabaseMetaData.columnNullable);
+        }
+    }
+
+    public void testGetPrimaryKeys() throws SQLException {
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getPrimaryKeys(METADATA_DATAVERSE_NAME, null, null)) {
+                int n = countRows(rs);
+                Assert.assertTrue(String.valueOf(n), n > 20);
+            }
+            try (ResultSet rs = md.getPrimaryKeys(METADATA_DATAVERSE_NAME, null, "Data%")) {
+                int n = countRows(rs);
+                Assert.assertTrue(String.valueOf(n), n > 4);
+            }
+            // we don't have any tables without catalogs
+            try (ResultSet rs = md.getPrimaryKeys("", null, null)) {
+                int n = countRows(rs);
+                Assert.assertEquals(0, n);
+            }
+        }
+
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<List<String>> newDataverseList = new ArrayList<>();
+            List<Pair<List<String>, String>> newDatasetList = new ArrayList<>();
+            List<Pair<List<String>, String>> newViewList = new ArrayList<>();
+
+            try {
+                createDataversesDatasetsViews(s, newDataverseList, newDatasetList, newViewList);
+
+                DatabaseMetaData md = c.getMetaData();
+
+                List<String> expectedColumns = Arrays.asList(TABLE_CAT, TABLE_SCHEM, TABLE_NAME, COLUMN_NAME, KEY_SEQ);
+                List<String> expectedTableCat = new ArrayList<>();
+                List<String> expectedTableSchem = new ArrayList<>();
+                List<String> expectedTableName = new ArrayList<>();
+                List<String> expectedColumnName = new ArrayList<>();
+                List<Integer> expectedKeySeq = new ArrayList<>();
+
+                // Test getPrimaryKeys() for a particular dataset/view
+                for (int i = 0, n = newDatasetList.size(); i < n; i++) {
+                    for (int j = 0; j < 2; j++) {
+                        Pair<List<String>, String> p = j == 0 ? newDatasetList.get(i) : newViewList.get(i);
+                        List<String> columnNames = j == 0 ? DATASET_COLUMN_NAMES : VIEW_COLUMN_NAMES;
+                        String dvNameCanonical = getCanonicalDataverseName(p.first);
+                        String dsName = p.second;
+
+                        expectedTableCat.clear();
+                        expectedTableSchem.clear();
+                        expectedTableName.clear();
+                        expectedColumnName.clear();
+                        expectedKeySeq.clear();
+
+                        List<String> pkColumnNames = columnNames.subList(0, DATASET_PK_LEN);
+                        addExpectedColumnNamesForGetPrimaryKeys(dvNameCanonical, dsName, pkColumnNames,
+                                expectedTableCat, expectedTableSchem, expectedTableName, expectedColumnName,
+                                expectedKeySeq);
+
+                        try (ResultSet rs = md.getPrimaryKeys(dvNameCanonical, null, dsName)) {
+                            assertColumnValues(rs, expectedColumns, Arrays.asList(expectedTableCat, expectedTableSchem,
+                                    expectedTableName, expectedColumnName, expectedKeySeq));
+                        }
+                    }
+                }
+
+                // non-existent catalog
+                try (ResultSet rs = md.getPrimaryKeys("UNKNOWN", null, null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent schema
+                try (ResultSet rs = md.getPrimaryKeys(null, "UNKNOWN", null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent table name
+                try (ResultSet rs = md.getPrimaryKeys(null, null, "UNKNOWN")) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    private void addExpectedColumnNamesForGetPrimaryKeys(String dvNameCanonical, String dsName,
+            List<String> pkColumnNames, List<String> outTableCat, List<String> outTableSchem, List<String> outTableName,
+            List<String> outColumnName, List<Integer> outKeySeq) {
+        List<String> pkColumnNamesSorted = new ArrayList<>(pkColumnNames);
+        Collections.sort(pkColumnNamesSorted);
+        for (int i = 0; i < pkColumnNames.size(); i++) {
+            String pkColumnName = pkColumnNamesSorted.get(i);
+            outTableCat.add(dvNameCanonical);
+            outTableSchem.add(null);
+            outTableName.add(dsName);
+            outColumnName.add(pkColumnName);
+            outKeySeq.add(pkColumnNames.indexOf(pkColumnName) + 1);
+        }
+    }
+
+    public void testGetImportedKeys() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<List<String>> newDataverseList = new ArrayList<>();
+            List<Pair<List<String>, String>> newDatasetList = new ArrayList<>();
+            List<Pair<List<String>, String>> newViewList = new ArrayList<>();
+
+            try {
+                createDataversesDatasetsViews(s, newDataverseList, newDatasetList, newViewList);
+
+                DatabaseMetaData md = c.getMetaData();
+
+                List<String> expectedColumns = Arrays.asList(PKTABLE_CAT, PKTABLE_SCHEM, PKTABLE_NAME, PKCOLUMN_NAME,
+                        FKTABLE_CAT, FKTABLE_SCHEM, FKTABLE_NAME, FKCOLUMN_NAME, KEY_SEQ);
+                List<String> expectedPKTableCat = new ArrayList<>();
+                List<String> expectedPKTableSchem = new ArrayList<>();
+                List<String> expectedPKTableName = new ArrayList<>();
+                List<String> expectedPKColumnName = new ArrayList<>();
+                List<String> expectedFKTableCat = new ArrayList<>();
+                List<String> expectedFKTableSchem = new ArrayList<>();
+                List<String> expectedFKTableName = new ArrayList<>();
+                List<String> expectedFKColumnName = new ArrayList<>();
+                List<Integer> expectedKeySeq = new ArrayList<>();
+
+                // Test getImportedKeys() for a particular view
+                for (int i = 0, n = newViewList.size(); i < n; i++) {
+                    Pair<List<String>, String> p = newViewList.get(i);
+                    List<String> dvName = p.first;
+                    String dvNameCanonical = getCanonicalDataverseName(dvName);
+                    String viewName = p.second;
+
+                    expectedPKTableCat.clear();
+                    expectedPKTableSchem.clear();
+                    expectedPKTableName.clear();
+                    expectedPKColumnName.clear();
+                    expectedFKTableCat.clear();
+                    expectedFKTableSchem.clear();
+                    expectedFKTableName.clear();
+                    expectedFKColumnName.clear();
+
+                    expectedKeySeq.clear();
+
+                    List<String> pkFkColumnNames = VIEW_COLUMN_NAMES.subList(0, DATASET_PK_LEN);
+                    List<String> fkRefs = IntStream.range(0, i).mapToObj(newViewList::get)
+                            .filter(p2 -> p2.first.equals(dvName)).map(p2 -> p2.second).collect(Collectors.toList());
+
+                    addExpectedColumnNamesForGetImportedKeys(dvNameCanonical, viewName, pkFkColumnNames, fkRefs,
+                            expectedPKTableCat, expectedPKTableSchem, expectedPKTableName, expectedPKColumnName,
+                            expectedFKTableCat, expectedFKTableSchem, expectedFKTableName, expectedFKColumnName,
+                            expectedKeySeq);
+
+                    try (ResultSet rs = md.getImportedKeys(dvNameCanonical, null, viewName)) {
+                        assertColumnValues(rs, expectedColumns,
+                                Arrays.asList(expectedPKTableCat, expectedPKTableSchem, expectedPKTableName,
+                                        expectedPKColumnName, expectedFKTableCat, expectedFKTableSchem,
+                                        expectedFKTableName, expectedFKColumnName, expectedKeySeq));
+                    }
+                }
+
+                // non-existent catalog
+                try (ResultSet rs = md.getImportedKeys("UNKNOWN", null, null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent schema
+                try (ResultSet rs = md.getImportedKeys(null, "UNKNOWN", null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent table name
+                try (ResultSet rs = md.getImportedKeys(null, null, "UNKNOWN")) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    private void addExpectedColumnNamesForGetImportedKeys(String dvNameCanonical, String dsName,
+            List<String> pkFkColumnNames, List<String> fkRefs, List<String> outPKTableCat, List<String> outPKTableSchem,
+            List<String> outPKTableName, List<String> outPKColumnName, List<String> outFKTableCat,
+            List<String> outFKTableSchem, List<String> outFKTableName, List<String> outFKColumnName,
+            List<Integer> outKeySeq) {
+        for (String fkRef : fkRefs) {
+            for (int i = 0; i < pkFkColumnNames.size(); i++) {
+                String pkFkColumn = pkFkColumnNames.get(i);
+                outPKTableCat.add(dvNameCanonical);
+                outPKTableSchem.add(null);
+                outPKTableName.add(fkRef);
+                outPKColumnName.add(pkFkColumn);
+                outFKTableCat.add(dvNameCanonical);
+                outFKTableSchem.add(null);
+                outFKTableName.add(dsName);
+                outFKColumnName.add(pkFkColumn);
+                outKeySeq.add(i + 1);
+            }
+        }
+    }
+
+    public void testGetExportedKeys() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<List<String>> newDataverseList = new ArrayList<>();
+            List<Pair<List<String>, String>> newDatasetList = new ArrayList<>();
+            List<Pair<List<String>, String>> newViewList = new ArrayList<>();
+
+            try {
+                createDataversesDatasetsViews(s, newDataverseList, newDatasetList, newViewList);
+
+                DatabaseMetaData md = c.getMetaData();
+
+                List<String> expectedColumns = Arrays.asList(PKTABLE_CAT, PKTABLE_SCHEM, PKTABLE_NAME, PKCOLUMN_NAME,
+                        FKTABLE_CAT, FKTABLE_SCHEM, FKTABLE_NAME, FKCOLUMN_NAME, KEY_SEQ);
+                List<String> expectedPKTableCat = new ArrayList<>();
+                List<String> expectedPKTableSchem = new ArrayList<>();
+                List<String> expectedPKTableName = new ArrayList<>();
+                List<String> expectedPKColumnName = new ArrayList<>();
+                List<String> expectedFKTableCat = new ArrayList<>();
+                List<String> expectedFKTableSchem = new ArrayList<>();
+                List<String> expectedFKTableName = new ArrayList<>();
+                List<String> expectedFKColumnName = new ArrayList<>();
+                List<Integer> expectedKeySeq = new ArrayList<>();
+
+                // Test getExportedKeys() for a particular view
+                for (int i = 0, n = newViewList.size(); i < n; i++) {
+                    Pair<List<String>, String> p = newViewList.get(i);
+                    List<String> dvName = p.first;
+                    String dvNameCanonical = getCanonicalDataverseName(dvName);
+                    String viewName = p.second;
+
+                    expectedPKTableCat.clear();
+                    expectedPKTableSchem.clear();
+                    expectedPKTableName.clear();
+                    expectedPKColumnName.clear();
+                    expectedFKTableCat.clear();
+                    expectedFKTableSchem.clear();
+                    expectedFKTableName.clear();
+                    expectedFKColumnName.clear();
+                    expectedKeySeq.clear();
+
+                    List<String> pkFkColumnNames = VIEW_COLUMN_NAMES.subList(0, DATASET_PK_LEN);
+                    List<String> fkRefs = IntStream.range(i + 1, newViewList.size()).mapToObj(newViewList::get)
+                            .filter(p2 -> p2.first.equals(dvName)).map(p2 -> p2.second).collect(Collectors.toList());
+
+                    addExpectedColumnNamesForGetExportedKeys(dvNameCanonical, viewName, pkFkColumnNames, fkRefs,
+                            expectedPKTableCat, expectedPKTableSchem, expectedPKTableName, expectedPKColumnName,
+                            expectedFKTableCat, expectedFKTableSchem, expectedFKTableName, expectedFKColumnName,
+                            expectedKeySeq);
+
+                    try (ResultSet rs = md.getExportedKeys(dvNameCanonical, null, viewName)) {
+                        assertColumnValues(rs, expectedColumns,
+                                Arrays.asList(expectedPKTableCat, expectedPKTableSchem, expectedPKTableName,
+                                        expectedPKColumnName, expectedFKTableCat, expectedFKTableSchem,
+                                        expectedFKTableName, expectedFKColumnName, expectedKeySeq));
+                    }
+                }
+
+                // non-existent catalog
+                try (ResultSet rs = md.getExportedKeys("UNKNOWN", null, null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent schema
+                try (ResultSet rs = md.getExportedKeys(null, "UNKNOWN", null)) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+                // non-existent table name
+                try (ResultSet rs = md.getExportedKeys(null, null, "UNKNOWN")) {
+                    Assert.assertEquals(0, countRows(rs));
+                }
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    private void addExpectedColumnNamesForGetExportedKeys(String dvNameCanonical, String dsName,
+            List<String> pkFkColumnNames, List<String> fkRefs, List<String> outPKTableCat, List<String> outPKTableSchem,
+            List<String> outPKTableName, List<String> outPKColumnName, List<String> outFKTableCat,
+            List<String> outFKTableSchem, List<String> outFKTableName, List<String> outFKColumnName,
+            List<Integer> outKeySeq) {
+        for (String fkRef : fkRefs) {
+            for (int i = 0; i < pkFkColumnNames.size(); i++) {
+                String pkFkColumn = pkFkColumnNames.get(i);
+                outPKTableCat.add(dvNameCanonical);
+                outPKTableSchem.add(null);
+                outPKTableName.add(dsName);
+                outPKColumnName.add(pkFkColumn);
+                outFKTableCat.add(dvNameCanonical);
+                outFKTableSchem.add(null);
+                outFKTableName.add(fkRef);
+                outFKColumnName.add(pkFkColumn);
+                outKeySeq.add(i + 1);
+            }
+        }
+    }
+
+    public void testGetCrossReference() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<List<String>> newDataverseList = new ArrayList<>();
+            List<Pair<List<String>, String>> newDatasetList = new ArrayList<>();
+            List<Pair<List<String>, String>> newViewList = new ArrayList<>();
+
+            try {
+                createDataversesDatasetsViews(s, newDataverseList, newDatasetList, newViewList);
+
+                DatabaseMetaData md = c.getMetaData();
+
+                List<String> expectedColumns = Arrays.asList(PKTABLE_CAT, PKTABLE_SCHEM, PKTABLE_NAME, PKCOLUMN_NAME,
+                        FKTABLE_CAT, FKTABLE_SCHEM, FKTABLE_NAME, FKCOLUMN_NAME, KEY_SEQ);
+                List<String> expectedPKTableCat = new ArrayList<>();
+                List<String> expectedPKTableSchem = new ArrayList<>();
+                List<String> expectedPKTableName = new ArrayList<>();
+                List<String> expectedPKColumnName = new ArrayList<>();
+                List<String> expectedFKTableCat = new ArrayList<>();
+                List<String> expectedFKTableSchem = new ArrayList<>();
+                List<String> expectedFKTableName = new ArrayList<>();
+                List<String> expectedFKColumnName = new ArrayList<>();
+                List<Integer> expectedKeySeq = new ArrayList<>();
+
+                boolean testUnknown = true;
+                // Test getCrossReference() for a particular view
+                for (int i = 0, n = newViewList.size(); i < n; i++) {
+                    Pair<List<String>, String> p = newViewList.get(i);
+                    List<String> dvName = p.first;
+                    String dvNameCanonical = getCanonicalDataverseName(dvName);
+                    String viewName = p.second;
+
+                    List<String> pkFkColumnNames = VIEW_COLUMN_NAMES.subList(0, DATASET_PK_LEN);
+                    Iterator<String> fkRefIter = IntStream.range(i + 1, newViewList.size()).mapToObj(newViewList::get)
+                            .filter(p2 -> p2.first.equals(dvName)).map(p2 -> p2.second).iterator();
+                    boolean hasFkRefs = fkRefIter.hasNext();
+                    while (fkRefIter.hasNext()) {
+                        String fkRef = fkRefIter.next();
+
+                        expectedPKTableCat.clear();
+                        expectedPKTableSchem.clear();
+                        expectedPKTableName.clear();
+                        expectedPKColumnName.clear();
+                        expectedFKTableCat.clear();
+                        expectedFKTableSchem.clear();
+                        expectedFKTableName.clear();
+                        expectedFKColumnName.clear();
+                        expectedKeySeq.clear();
+
+                        addExpectedColumnNamesForGetCrossReference(dvNameCanonical, viewName, pkFkColumnNames, fkRef,
+                                expectedPKTableCat, expectedPKTableSchem, expectedPKTableName, expectedPKColumnName,
+                                expectedFKTableCat, expectedFKTableSchem, expectedFKTableName, expectedFKColumnName,
+                                expectedKeySeq);
+
+                        try (ResultSet rs =
+                                md.getCrossReference(dvNameCanonical, null, viewName, dvNameCanonical, null, fkRef)) {
+                            assertColumnValues(rs, expectedColumns,
+                                    Arrays.asList(expectedPKTableCat, expectedPKTableSchem, expectedPKTableName,
+                                            expectedPKColumnName, expectedFKTableCat, expectedFKTableSchem,
+                                            expectedFKTableName, expectedFKColumnName, expectedKeySeq));
+                        }
+                    }
+
+                    if (testUnknown && hasFkRefs) {
+                        testUnknown = false;
+                        // non-existent catalog
+                        try (ResultSet rs =
+                                md.getCrossReference(dvNameCanonical, null, viewName, "UNKNOWN", null, "UNKNOWN")) {
+                            Assert.assertEquals(0, countRows(rs));
+                        }
+                        // non-existent schema
+                        try (ResultSet rs = md.getCrossReference(dvNameCanonical, null, viewName, dvNameCanonical,
+                                "UNKNOWN", "UNKNOWN")) {
+                            Assert.assertEquals(0, countRows(rs));
+                        }
+                        // non-existent table name
+                        try (ResultSet rs = md.getCrossReference(dvNameCanonical, null, viewName, dvNameCanonical, null,
+                                "UNKNOWN")) {
+                            Assert.assertEquals(0, countRows(rs));
+                        }
+                    }
+                }
+
+            } finally {
+                dropDataverses(s, newDataverseList);
+            }
+        }
+    }
+
+    private void addExpectedColumnNamesForGetCrossReference(String dvNameCanonical, String dsName,
+            List<String> pkFkColumnNames, String fkRef, List<String> outPKTableCat, List<String> outPKTableSchem,
+            List<String> outPKTableName, List<String> outPKColumnName, List<String> outFKTableCat,
+            List<String> outFKTableSchem, List<String> outFKTableName, List<String> outFKColumnName,
+            List<Integer> outKeySeq) {
+        for (int i = 0; i < pkFkColumnNames.size(); i++) {
+            String pkFkColumn = pkFkColumnNames.get(i);
+            outPKTableCat.add(dvNameCanonical);
+            outPKTableSchem.add(null);
+            outPKTableName.add(dsName);
+            outPKColumnName.add(pkFkColumn);
+            outFKTableCat.add(dvNameCanonical);
+            outFKTableSchem.add(null);
+            outFKTableName.add(fkRef);
+            outFKColumnName.add(pkFkColumn);
+            outKeySeq.add(i + 1);
+        }
+    }
+
+    public void testGetTypeInfo() throws SQLException {
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            try (ResultSet rs = md.getTypeInfo()) {
+                int n = countRows(rs);
+                Assert.assertTrue(String.valueOf(n), n > 10);
+            }
+        }
+    }
+
+    private static boolean isMetadataCatalog(ResultSet rs) throws SQLException {
+        return METADATA_DATAVERSE_NAME.equals(rs.getString(TABLE_CAT));
+    }
+
+    private void createDataverses(Statement stmt, List<List<String>> outDataverseList) throws SQLException {
+        for (String p1 : new String[] { "x", "y" }) {
+            List<String> dv1 = Collections.singletonList(p1);
+            stmt.execute(printCreateDataverse(dv1));
+            outDataverseList.add(dv1);
+            for (int p2i = 0; p2i <= 9; p2i++) {
+                String p2 = "z" + p2i;
+                List<String> dv2 = Arrays.asList(p1, p2);
+                stmt.execute(printCreateDataverse(dv2));
+                outDataverseList.add(dv2);
+            }
+        }
+    }
+
+    private void createDataversesDatasetsViews(Statement stmt, List<List<String>> outDataverseList,
+            List<Pair<List<String>, String>> outDatasetList, List<Pair<List<String>, String>> outViewList)
+            throws SQLException {
+        for (String p1 : new String[] { "x", "y" }) {
+            for (int p2i = 0; p2i < 2; p2i++) {
+                String p2 = "z" + p2i;
+                List<String> dv = Arrays.asList(p1, p2);
+                stmt.execute(printCreateDataverse(dv));
+                outDataverseList.add(dv);
+                for (int i = 0; i < 3; i++) {
+                    // create dataset
+                    String datasetName = createDatasetName(i);
+                    stmt.execute(printCreateDataset(dv, datasetName, DATASET_COLUMN_NAMES, DATASET_COLUMN_TYPES,
+                            DATASET_PK_LEN));
+                    outDatasetList.add(new Pair<>(dv, datasetName));
+                    // create tabular view
+                    String viewName = createViewName(i);
+                    String viewQuery = "select r va, r vb, r vc from range(1,2) r";
+                    List<String> fkRefs = IntStream.range(0, i).mapToObj(JdbcMetadataTester::createViewName)
+                            .collect(Collectors.toList());
+                    stmt.execute(printCreateView(dv, viewName, VIEW_COLUMN_NAMES, DATASET_COLUMN_TYPES, DATASET_PK_LEN,
+                            fkRefs, viewQuery));
+                    outViewList.add(new Pair<>(dv, viewName));
+                }
+            }
+        }
+    }
+
+    private static String createDatasetName(int id) {
+        return "t" + id;
+    }
+
+    private static String createViewName(int id) {
+        return "v" + id;
+    }
+
+    private void dropDataverses(Statement stmt, List<List<String>> dataverseList) throws SQLException {
+        for (List<String> dv : dataverseList) {
+            stmt.execute(printDropDataverse(dv));
+        }
+    }
+
+    private void assertColumnValues(ResultSet rs, String column, List<?> values) throws SQLException {
+        assertColumnValues(rs, Collections.singletonList(column), Collections.singletonList(values));
+    }
+
+    private void assertColumnValues(ResultSet rs, List<String> columns, List<List<?>> values) throws SQLException {
+        assertColumnValues(rs, columns, values, null);
+    }
+
+    private void assertColumnValues(ResultSet rs, List<String> columns, List<List<?>> values,
+            JdbcPredicate<ResultSet> skipRowTest) throws SQLException {
+        int columnCount = columns.size();
+        Assert.assertEquals(columnCount, values.size());
+        List<Iterator<?>> valueIters = values.stream().map(List::iterator).collect(Collectors.toList());
+        while (rs.next()) {
+            if (skipRowTest != null && skipRowTest.test(rs)) {
+                continue;
+            }
+            for (int i = 0; i < columnCount; i++) {
+                String column = columns.get(i);
+                Object expectedValue = valueIters.get(i).next();
+                Object actualValue;
+                if (expectedValue instanceof String) {
+                    actualValue = rs.getString(column);
+                } else if (expectedValue instanceof Integer) {
+                    actualValue = rs.getInt(column);
+                } else if (expectedValue instanceof Long) {
+                    actualValue = rs.getLong(column);
+                } else {
+                    actualValue = rs.getObject(column);
+                }
+                if (rs.wasNull()) {
+                    Assert.assertNull(expectedValue);
+                } else {
+                    Assert.assertEquals(expectedValue, actualValue);
+                }
+            }
+        }
+        for (Iterator<?> i : valueIters) {
+            if (i.hasNext()) {
+                Assert.fail(String.valueOf(i.next()));
+            }
+        }
+    }
+
+    private int countRows(ResultSet rs) throws SQLException {
+        int n = 0;
+        while (rs.next()) {
+            n++;
+        }
+        return n;
+    }
+
+    public void testWrapper() throws SQLException {
+        try (Connection c = createConnection()) {
+            DatabaseMetaData md = c.getMetaData();
+            Assert.assertTrue(md.isWrapperFor(ADBDatabaseMetaData.class));
+            Assert.assertNotNull(md.unwrap(ADBDatabaseMetaData.class));
+        }
+    }
+}
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcPreparedStatementTester.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcPreparedStatementTester.java
new file mode 100644
index 0000000..4ff78d2
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcPreparedStatementTester.java
@@ -0,0 +1,291 @@
+/*
+ * 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.asterix.test.jdbc;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.jdbc.core.ADBPreparedStatement;
+import org.junit.Assert;
+
+class JdbcPreparedStatementTester extends JdbcTester {
+
+    public void testLifecycle() throws SQLException {
+        Connection c = createConnection();
+        PreparedStatement s = c.prepareStatement(Q1);
+        Assert.assertFalse(s.isClosed());
+        Assert.assertSame(c, s.getConnection());
+
+        s.close();
+        Assert.assertTrue(s.isClosed());
+
+        // ok to call close() on a closed statement
+        s.close();
+        Assert.assertTrue(s.isClosed());
+    }
+
+    public void testAutoCloseOnConnectionClose() throws SQLException {
+        Connection c = createConnection();
+        // check that a statement is automatically closed when the connection is closed
+        PreparedStatement s = c.prepareStatement(Q1);
+        Assert.assertFalse(s.isClosed());
+        c.close();
+        Assert.assertTrue(s.isClosed());
+    }
+
+    public void testCloseOnCompletion() throws SQLException {
+        try (Connection c = createConnection()) {
+            PreparedStatement s = c.prepareStatement(Q1);
+            Assert.assertFalse(s.isCloseOnCompletion());
+            s.closeOnCompletion();
+            Assert.assertTrue(s.isCloseOnCompletion());
+            Assert.assertFalse(s.isClosed());
+            ResultSet rs = s.executeQuery();
+            Assert.assertTrue(rs.next());
+            Assert.assertFalse(rs.next());
+            rs.close();
+            Assert.assertTrue(s.isClosed());
+        }
+    }
+
+    public void testExecuteQuery() throws SQLException {
+        try (Connection c = createConnection()) {
+            // Query -> ok
+            try (PreparedStatement s1 = c.prepareStatement(Q1); ResultSet rs1 = s1.executeQuery()) {
+                Assert.assertTrue(rs1.next());
+                Assert.assertEquals(1, rs1.getMetaData().getColumnCount());
+                Assert.assertEquals(V1, rs1.getInt(1));
+                Assert.assertFalse(rs1.next());
+                Assert.assertFalse(rs1.isClosed());
+            }
+
+            // DDL -> error
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testExecuteQuery");
+            try {
+                PreparedStatement s2 = c.prepareStatement(printCreateDataverse(dataverse));
+                s2.executeQuery();
+                Assert.fail("DDL did not fail in executeQuery()");
+            } catch (SQLException e) {
+                String msg = e.getMessage();
+                Assert.assertTrue(msg, msg.contains(ErrorCode.PROHIBITED_STATEMENT_CATEGORY.errorCode()));
+            }
+
+            // DML -> error
+            String dataset = "ds1";
+            PreparedStatement s3 = c.prepareStatement(printCreateDataverse(dataverse));
+            s3.execute();
+            PreparedStatement s4 = c.prepareStatement(printCreateDataset(dataverse, dataset));
+            s4.execute();
+            try {
+                PreparedStatement s5 = c.prepareStatement(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+                s5.executeQuery();
+                Assert.fail("DML did not fail in executeQuery()");
+            } catch (SQLException e) {
+                String msg = e.getMessage();
+                Assert.assertTrue(msg, msg.contains(ErrorCode.PROHIBITED_STATEMENT_CATEGORY.errorCode()));
+            }
+
+            // Cleanup
+            PreparedStatement s6 = c.prepareStatement(printDropDataverse(dataverse));
+            s6.execute();
+        }
+    }
+
+    public void testExecuteUpdate() throws SQLException {
+        try (Connection c = createConnection()) {
+            // DDL -> ok
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testExecuteUpdate");
+            PreparedStatement s1 = c.prepareStatement(printCreateDataverse(dataverse));
+            int res = s1.executeUpdate();
+            Assert.assertEquals(0, res);
+            String dataset = "ds1";
+            PreparedStatement s2 = c.prepareStatement(printCreateDataset(dataverse, dataset));
+            res = s2.executeUpdate();
+            Assert.assertEquals(0, res);
+
+            // DML -> ok
+            PreparedStatement s3 = c.prepareStatement(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+            res = s3.executeUpdate();
+            // currently, DML statements always return update count = 1
+            Assert.assertEquals(1, res);
+
+            // Query -> error
+            try {
+                PreparedStatement s4 = c.prepareStatement(Q1);
+                s4.executeUpdate();
+                Assert.fail("Query did not fail in executeUpdate()");
+            } catch (SQLException e) {
+                String msg = e.getMessage();
+                Assert.assertTrue(msg, msg.contains("Invalid statement category"));
+            }
+
+            // Cleanup
+            PreparedStatement s5 = c.prepareStatement(printDropDataverse(dataverse));
+            s5.executeUpdate();
+        }
+    }
+
+    public void testExecute() throws SQLException {
+        try (Connection c = createConnection()) {
+            // Query -> ok
+            PreparedStatement s1 = c.prepareStatement(Q1);
+            boolean res = s1.execute();
+            Assert.assertTrue(res);
+            Assert.assertEquals(-1, s1.getUpdateCount());
+            try (ResultSet rs = s1.getResultSet()) {
+                Assert.assertTrue(rs.next());
+                Assert.assertEquals(1, rs.getMetaData().getColumnCount());
+                Assert.assertEquals(V1, rs.getInt(1));
+                Assert.assertFalse(rs.next());
+                Assert.assertFalse(rs.isClosed());
+            }
+
+            // DDL -> ok
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testExecute");
+            PreparedStatement s2 = c.prepareStatement(printCreateDataverse(dataverse));
+            res = s2.execute();
+            Assert.assertFalse(res);
+            Assert.assertEquals(0, s2.getUpdateCount());
+            String dataset = "ds1";
+            PreparedStatement s3 = c.prepareStatement(printCreateDataset(dataverse, dataset));
+            res = s3.execute();
+            Assert.assertFalse(res);
+
+            // DML -> ok
+            PreparedStatement s4 = c.prepareStatement(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+            res = s4.execute();
+            Assert.assertFalse(res);
+            // currently, DML statements always return update count = 1
+            Assert.assertEquals(1, s4.getUpdateCount());
+
+            // Cleanup
+            PreparedStatement s5 = c.prepareStatement(printDropDataverse(dataverse));
+            s5.execute();
+        }
+    }
+
+    public void testGetResultSet() throws SQLException {
+        try (Connection c = createConnection()) {
+            // Query
+            PreparedStatement s1 = c.prepareStatement(Q1);
+            boolean res = s1.execute();
+            Assert.assertTrue(res);
+            ResultSet rs = s1.getResultSet();
+            Assert.assertFalse(rs.isClosed());
+            Assert.assertTrue(rs.next());
+            Assert.assertFalse(s1.getMoreResults()); // closes current ResultSet
+            Assert.assertTrue(rs.isClosed());
+
+            PreparedStatement s2 = c.prepareStatement(Q1);
+            res = s2.execute();
+            Assert.assertTrue(res);
+            rs = s2.getResultSet();
+            Assert.assertFalse(rs.isClosed());
+            Assert.assertTrue(rs.next());
+            Assert.assertFalse(s2.getMoreResults(Statement.KEEP_CURRENT_RESULT));
+            Assert.assertFalse(rs.isClosed());
+            rs.close();
+
+            // DDL
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testGetResultSet");
+            PreparedStatement s3 = c.prepareStatement(printCreateDataverse(dataverse));
+            res = s3.execute();
+            Assert.assertFalse(res);
+            Assert.assertNull(s3.getResultSet());
+            Assert.assertFalse(s3.getMoreResults());
+
+            String dataset = "ds1";
+            PreparedStatement s4 = c.prepareStatement(printCreateDataset(dataverse, dataset));
+            res = s4.execute();
+            Assert.assertFalse(res);
+
+            // DML
+            PreparedStatement s5 = c.prepareStatement(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+            res = s5.execute();
+            Assert.assertFalse(res);
+            Assert.assertNull(s5.getResultSet());
+            Assert.assertFalse(s5.getMoreResults());
+        }
+    }
+
+    public void testMaxRows() throws SQLException {
+        try (Connection c = createConnection()) {
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testMaxRows");
+            String dataset = "ds1";
+            String field = "x";
+            PreparedStatement s1 = c.prepareStatement(printCreateDataverse(dataverse));
+            s1.execute();
+            PreparedStatement s2 = c.prepareStatement(printCreateDataset(dataverse, dataset));
+            s2.execute();
+            PreparedStatement s3 = c.prepareStatement(printInsert(dataverse, dataset, dataGen(field, 1, 2, 3)));
+            s3.execute();
+
+            PreparedStatement s4 = c.prepareStatement(String.format("select %s from %s.%s", field,
+                    printDataverseName(dataverse), printIdentifier(dataset)));
+            s4.setMaxRows(2);
+            Assert.assertEquals(2, s4.getMaxRows());
+            try (ResultSet rs = s4.executeQuery()) {
+                Assert.assertTrue(rs.next());
+                Assert.assertTrue(rs.next());
+                Assert.assertFalse(rs.next());
+            }
+        }
+    }
+
+    public void testWarnings() throws SQLException {
+        try (Connection c = createConnection();
+                PreparedStatement s = c.prepareStatement("select double('x'), bigint('y')"); // --> NULL with warning
+                ResultSet rs = s.executeQuery()) {
+            Assert.assertTrue(rs.next());
+            rs.getDouble(1);
+            Assert.assertTrue(rs.wasNull());
+            rs.getLong(2);
+            Assert.assertTrue(rs.wasNull());
+
+            SQLWarning w = s.getWarnings();
+            Assert.assertNotNull(w);
+            String msg = w.getMessage();
+            Assert.assertTrue(msg, msg.contains(ErrorCode.INVALID_FORMAT.errorCode()));
+
+            SQLWarning w2 = w.getNextWarning();
+            Assert.assertNotNull(w2);
+            String msg2 = w.getMessage();
+            Assert.assertTrue(msg2, msg2.contains(ErrorCode.INVALID_FORMAT.errorCode()));
+
+            Assert.assertNull(w2.getNextWarning());
+            s.clearWarnings();
+            Assert.assertNull(s.getWarnings());
+        }
+    }
+
+    public void testWrapper() throws SQLException {
+        try (Connection c = createConnection(); PreparedStatement s = c.prepareStatement(Q1)) {
+            Assert.assertTrue(s.isWrapperFor(ADBPreparedStatement.class));
+            Assert.assertNotNull(s.unwrap(ADBPreparedStatement.class));
+        }
+    }
+}
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcResultSetTester.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcResultSetTester.java
new file mode 100644
index 0000000..9b1acd1
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcResultSetTester.java
@@ -0,0 +1,672 @@
+/*
+ * 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.asterix.test.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Date;
+import java.sql.JDBCType;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.Period;
+import java.time.ZoneOffset;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.asterix.jdbc.core.ADBResultSet;
+import org.apache.commons.io.IOUtils;
+import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.junit.Assert;
+
+abstract class JdbcResultSetTester extends JdbcTester {
+
+    protected abstract CloseablePair<Statement, ResultSet> executeQuery(Connection c, String query) throws SQLException;
+
+    public void testLifecycle() throws SQLException {
+        try (Connection c = createConnection()) {
+            Pair<Statement, ResultSet> p = executeQuery(c, Q2);
+            Statement s = p.getFirst();
+            ResultSet rs = p.getSecond();
+            Assert.assertFalse(rs.isClosed());
+            Assert.assertSame(s, rs.getStatement());
+            rs.close();
+            Assert.assertTrue(rs.isClosed());
+
+            // ok to call close() on a closed result set
+            rs.close();
+            Assert.assertTrue(rs.isClosed());
+        }
+    }
+
+    // test that Statement.close() closes its ResultSet
+    public void testAutoCloseOnStatementClose() throws SQLException {
+        try (Connection c = createConnection()) {
+            Pair<Statement, ResultSet> p = executeQuery(c, Q2);
+            Statement s = p.getFirst();
+            ResultSet rs = p.getSecond();
+            Assert.assertFalse(rs.isClosed());
+            s.close();
+            Assert.assertTrue(rs.isClosed());
+        }
+    }
+
+    // test that Connection.close() closes all Statements and their ResultSets
+    public void testAutoCloseOnConnectionClose() throws SQLException {
+        Connection c = createConnection();
+        Pair<Statement, ResultSet> p1 = executeQuery(c, Q2);
+        Statement s1 = p1.getFirst();
+        ResultSet rs1 = p1.getSecond();
+        Assert.assertFalse(rs1.isClosed());
+        Pair<Statement, ResultSet> p2 = executeQuery(c, Q2);
+        Statement s2 = p2.getFirst();
+        ResultSet rs2 = p2.getSecond();
+        Assert.assertFalse(rs2.isClosed());
+        c.close();
+        Assert.assertTrue(rs1.isClosed());
+        Assert.assertTrue(s1.isClosed());
+        Assert.assertTrue(rs2.isClosed());
+        Assert.assertTrue(s2.isClosed());
+    }
+
+    public void testNavigation() throws SQLException {
+        try (Connection c = createConnection()) {
+            Pair<Statement, ResultSet> p = executeQuery(c, Q2);
+            ResultSet rs = p.getSecond();
+            Assert.assertEquals(ResultSet.TYPE_FORWARD_ONLY, rs.getType());
+            Assert.assertEquals(ResultSet.FETCH_FORWARD, rs.getFetchDirection());
+            Assert.assertTrue(rs.isBeforeFirst());
+            Assert.assertFalse(rs.isFirst());
+            // Assert.assertFalse(rs.isLast()); -- Not supported
+            Assert.assertFalse(rs.isAfterLast());
+            Assert.assertEquals(0, rs.getRow());
+
+            for (int r = 1; r <= 9; r++) {
+                boolean next = rs.next();
+                Assert.assertTrue(next);
+                Assert.assertFalse(rs.isBeforeFirst());
+                Assert.assertEquals(r == 1, rs.isFirst());
+                Assert.assertFalse(rs.isAfterLast());
+                Assert.assertEquals(r, rs.getRow());
+            }
+
+            boolean next = rs.next();
+            Assert.assertFalse(next);
+            Assert.assertFalse(rs.isBeforeFirst());
+            Assert.assertFalse(rs.isFirst());
+            Assert.assertTrue(rs.isAfterLast());
+            Assert.assertEquals(0, rs.getRow());
+
+            next = rs.next();
+            Assert.assertFalse(next);
+
+            rs.close();
+            assertErrorOnClosed(rs, ResultSet::isBeforeFirst, "isBeforeFirst");
+            assertErrorOnClosed(rs, ResultSet::isFirst, "isFirst");
+            assertErrorOnClosed(rs, ResultSet::isAfterLast, "isAfterLast");
+            assertErrorOnClosed(rs, ResultSet::getRow, "getRow");
+            assertErrorOnClosed(rs, ResultSet::next, "next");
+        }
+    }
+
+    public void testColumReadBasic() throws SQLException {
+        String qProject = IntStream.range(1, 10).mapToObj(i -> String.format("r*10+%d as c%d", i, i))
+                .collect(Collectors.joining(","));
+        String q = String.format("select %s from range(1, 2) r order by r", qProject);
+        try (Connection c = createConnection(); CloseablePair<Statement, ResultSet> p = executeQuery(c, q)) {
+            ResultSet rs = p.getSecond();
+            for (int r = 1; rs.next(); r++) {
+                for (int col = 1; col < 10; col++) {
+                    int expected = r * 10 + col;
+                    Assert.assertEquals(expected, rs.getInt(col));
+                    Assert.assertEquals(expected, rs.getInt("c" + col));
+                    Assert.assertEquals(expected, rs.getInt(rs.findColumn("c" + col)));
+                }
+            }
+        }
+    }
+
+    public void testColumnRead() throws SQLException, IOException {
+        try (Connection c = createConnection(); CloseablePair<Statement, ResultSet> p = executeQuery(c, Q3)) {
+            ResultSet rs = p.getSecond();
+            for (int r = -1; rs.next(); r++) {
+                int v = r * 2;
+                verifyReadColumnOfNumericType(rs, 1, Q3_COLUMNS[0], v == 0 ? null : (byte) v);
+                verifyReadColumnOfNumericType(rs, 2, Q3_COLUMNS[1], v == 0 ? null : (short) v);
+                verifyReadColumnOfNumericType(rs, 3, Q3_COLUMNS[2], v == 0 ? null : v);
+                verifyReadColumnOfNumericType(rs, 4, Q3_COLUMNS[3], v == 0 ? null : (long) v);
+                verifyReadColumnOfNumericType(rs, 5, Q3_COLUMNS[4], v == 0 ? null : (float) v);
+                verifyReadColumnOfNumericType(rs, 6, Q3_COLUMNS[5], v == 0 ? null : (double) v);
+                verifyReadColumnOfStringType(rs, 7, Q3_COLUMNS[6], v == 0 ? null : "a" + v);
+                verifyReadColumnOfBooleanType(rs, 8, Q3_COLUMNS[7], v == 0 ? null : v > 0);
+                verifyReadColumnOfDateType(rs, 9, Q3_COLUMNS[8], v == 0 ? null : LocalDate.ofEpochDay(v));
+                verifyReadColumnOfTimeType(rs, 10, Q3_COLUMNS[9], v == 0 ? null : LocalTime.ofSecondOfDay(v + 3));
+                verifyReadColumnOfDatetimeType(rs, 11, Q3_COLUMNS[10],
+                        v == 0 ? null : LocalDateTime.ofEpochSecond(v, 0, ZoneOffset.UTC));
+                verifyReadColumnOfYearMonthDurationType(rs, 12, Q3_COLUMNS[11], v == 0 ? null : Period.ofMonths(v));
+                verifyReadColumnOfDayTimeDurationType(rs, 13, Q3_COLUMNS[12], v == 0 ? null : Duration.ofMillis(v));
+                verifyReadColumnOfDurationType(rs, 14, Q3_COLUMNS[13], v == 0 ? null : Period.ofMonths(v + 3),
+                        v == 0 ? null : Duration.ofMillis(TimeUnit.SECONDS.toMillis(v + 3)));
+                verifyReadColumnOfUuidType(rs, 15, Q3_COLUMNS[14],
+                        v == 0 ? null : UUID.fromString("5c848e5c-6b6a-498f-8452-8847a295742" + (v + 3)));
+            }
+        }
+    }
+
+    public void testColumnMetadata() throws SQLException {
+        try (Connection c = createConnection(); CloseablePair<Statement, ResultSet> p = executeQuery(c, Q3)) {
+            ResultSet rs = p.getSecond();
+            int expectedColumnCount = Q3_COLUMNS.length;
+            ResultSetMetaData rsmd = rs.getMetaData();
+            Assert.assertEquals(expectedColumnCount, rsmd.getColumnCount());
+            for (int i = 1; i <= expectedColumnCount; i++) {
+                String expectedColumnName = Q3_COLUMNS[i - 1];
+                JDBCType expectedColumnTypeJdbc = Q3_COLUMN_TYPES_JDBC[i - 1];
+                String expectedColumnTypeAdb = Q3_COLUMN_TYPES_ADB[i - 1];
+                Class<?> expectedColumnTypeJava = Q3_COLUMN_TYPES_JAVA[i - 1];
+                Assert.assertEquals(i, rs.findColumn(expectedColumnName));
+                Assert.assertEquals(expectedColumnName, rsmd.getColumnName(i));
+                Assert.assertEquals(expectedColumnTypeJdbc.getVendorTypeNumber().intValue(), rsmd.getColumnType(i));
+                Assert.assertEquals(expectedColumnTypeAdb, rsmd.getColumnTypeName(i));
+                Assert.assertEquals(expectedColumnTypeJava.getName(), rsmd.getColumnClassName(i));
+            }
+        }
+    }
+
+    private void verifyGetColumnAsByte(ResultSet rs, int columnIndex, String columnName, Number expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        byte expectedByte = expectedValue == null ? 0 : expectedValue.byteValue();
+        byte v1 = rs.getByte(columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedByte, v1);
+        byte v2 = rs.getByte(columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedByte, v2);
+    }
+
+    private void verifyGetColumnAsShort(ResultSet rs, int columnIndex, String columnName, Number expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        short expectedShort = expectedValue == null ? 0 : expectedValue.shortValue();
+        short v1 = rs.getShort(columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedShort, v1);
+        short v2 = rs.getShort(columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedShort, v2);
+    }
+
+    private void verifyGetColumnAsInt(ResultSet rs, int columnIndex, String columnName, Number expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        int expectedInt = expectedValue == null ? 0 : expectedValue.intValue();
+        int v1 = rs.getInt(columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedInt, v1);
+        int v2 = rs.getInt(columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedInt, v2);
+    }
+
+    private void verifyGetColumnAsLong(ResultSet rs, int columnIndex, String columnName, Number expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        long expectedLong = expectedValue == null ? 0 : expectedValue.longValue();
+        long v1 = rs.getLong(columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedLong, v1);
+        long v2 = rs.getLong(columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedLong, v2);
+    }
+
+    private void verifyGetColumnAsFloat(ResultSet rs, int columnIndex, String columnName, Number expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        float expectedFloat = expectedValue == null ? 0f : expectedValue.floatValue();
+        float v1 = rs.getFloat(columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedFloat, v1, 0);
+        float v2 = rs.getFloat(columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedFloat, v2, 0);
+    }
+
+    private void verifyGetColumnAsDouble(ResultSet rs, int columnIndex, String columnName, Number expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        double expectedDouble = expectedValue == null ? 0d : expectedValue.doubleValue();
+        double v1 = rs.getDouble(columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedDouble, v1, 0);
+        double v2 = rs.getDouble(columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedDouble, v2, 0);
+    }
+
+    private void verifyGetColumnAsDecimal(ResultSet rs, int columnIndex, String columnName, Number expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        BigDecimal expectedDecimal = expectedValue == null ? null : new BigDecimal(expectedValue.toString());
+        int expectedDecimalScale = expectedValue == null ? 0 : expectedDecimal.scale();
+        BigDecimal v1 = rs.getBigDecimal(columnIndex, expectedDecimalScale);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedDecimal, v1);
+        BigDecimal v2 = rs.getBigDecimal(columnName, expectedDecimalScale);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedDecimal, v2);
+    }
+
+    private void verifyGetColumnAsBoolean(ResultSet rs, int columnIndex, String columnName, Boolean expectedValue)
+            throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        boolean expectedBoolean = expectedNull ? false : expectedValue;
+        boolean v1 = rs.getBoolean(columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedBoolean, v1);
+        boolean v2 = rs.getBoolean(columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        Assert.assertEquals(expectedBoolean, v2);
+    }
+
+    private void verifyGetColumnAsString(ResultSet rs, int columnIndex, String columnName, String expectedValue)
+            throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, ResultSet::getString, ResultSet::getString);
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, ResultSet::getNString,
+                ResultSet::getNString);
+    }
+
+    private void verifyGetColumnAsObject(ResultSet rs, int columnIndex, String columnName, Object expectedValue)
+            throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, ResultSet::getObject, ResultSet::getObject);
+    }
+
+    private <V> void verifyGetColumnAsObject(ResultSet rs, int columnIndex, String columnName, V expectedValue,
+            Function<Object, V> valueConverter) throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, valueConverter, ResultSet::getObject,
+                ResultSet::getObject);
+    }
+
+    private <V> void verifyGetColumnAsObject(ResultSet rs, int columnIndex, String columnName, V expectedValue,
+            Class<V> type) throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, ResultSet::getObject, ResultSet::getObject,
+                type);
+    }
+
+    private <V, T> void verifyGetColumnAsObject(ResultSet rs, int columnIndex, String columnName, V expectedValue,
+            Class<T> type, Function<T, V> valueConverter) throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, valueConverter, ResultSet::getObject,
+                ResultSet::getObject, type);
+    }
+
+    private void verifyGetColumnAsSqlDate(ResultSet rs, int columnIndex, String columnName, LocalDate expectedValue)
+            throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, java.sql.Date::toLocalDate,
+                ResultSet::getDate, ResultSet::getDate);
+    }
+
+    private void verifyGetColumnAsSqlTime(ResultSet rs, int columnIndex, String columnName, LocalTime expectedValue)
+            throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, java.sql.Time::toLocalTime,
+                ResultSet::getTime, ResultSet::getTime);
+    }
+
+    private void verifyGetColumnAsSqlTimestamp(ResultSet rs, int columnIndex, String columnName,
+            LocalDateTime expectedValue) throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, java.sql.Timestamp::toLocalDateTime,
+                ResultSet::getTimestamp, ResultSet::getTimestamp);
+    }
+
+    private <V> void verifyGetColumnGeneric(ResultSet rs, int columnIndex, String columnName, V expectedValue,
+            GetColumnByIndex<V> columnByIndexAccessor, GetColumnByName<V> columnByNameAccessor) throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, Function.identity(), columnByIndexAccessor,
+                columnByNameAccessor);
+    }
+
+    private <V, T> void verifyGetColumnGeneric(ResultSet rs, int columnIndex, String columnName, V expectedValue,
+            Function<T, V> valueConverter, GetColumnByIndex<T> columnByIndexAccessor,
+            GetColumnByName<T> columnByNameAccessor) throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        T v1 = columnByIndexAccessor.get(rs, columnIndex);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        if (expectedNull) {
+            Assert.assertNull(v1);
+        } else {
+            Assert.assertEquals(expectedValue, valueConverter.apply(v1));
+        }
+        T v2 = columnByNameAccessor.get(rs, columnName);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        if (expectedNull) {
+            Assert.assertNull(v2);
+        } else {
+            Assert.assertEquals(expectedValue, valueConverter.apply(v2));
+        }
+    }
+
+    private <V, P> void verifyGetColumnGeneric(ResultSet rs, int columnIndex, String columnName, V expectedValue,
+            GetColumnByIndexWithParam<V, P> columnByIndexAccessor, GetColumnByNameWithParam<V, P> columnByNameAccessor,
+            P accessorParamValue) throws SQLException {
+        verifyGetColumnGeneric(rs, columnIndex, columnName, expectedValue, Function.identity(), columnByIndexAccessor,
+                columnByNameAccessor, accessorParamValue);
+    }
+
+    private <V, T, P> void verifyGetColumnGeneric(ResultSet rs, int columnIndex, String columnName, V expectedValue,
+            Function<T, V> valueConverter, GetColumnByIndexWithParam<T, P> columnByIndexAccessor,
+            GetColumnByNameWithParam<T, P> columnByNameAccessor, P accessorParamValue) throws SQLException {
+        boolean expectedNull = expectedValue == null;
+        T v1 = columnByIndexAccessor.get(rs, columnIndex, accessorParamValue);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        if (expectedNull) {
+            Assert.assertNull(v1);
+        } else {
+            Assert.assertEquals(expectedValue, valueConverter.apply(v1));
+        }
+        T v2 = columnByNameAccessor.get(rs, columnName, accessorParamValue);
+        Assert.assertEquals(expectedNull, rs.wasNull());
+        if (expectedNull) {
+            Assert.assertNull(v2);
+        } else {
+            Assert.assertEquals(expectedValue, valueConverter.apply(v2));
+        }
+    }
+
+    private void verifyGetColumnAsCharacterStream(ResultSet rs, int columnIndex, String columnName,
+            char[] expectedValue, GetColumnByIndex<Reader> columnByIndexAccessor,
+            GetColumnByName<Reader> columnByNameAccessor) throws SQLException, IOException {
+        boolean expectedNull = expectedValue == null;
+        try (Reader s1 = columnByIndexAccessor.get(rs, columnIndex)) {
+            Assert.assertEquals(expectedNull, rs.wasNull());
+            if (expectedNull) {
+                Assert.assertNull(s1);
+            } else {
+                Assert.assertArrayEquals(expectedValue, IOUtils.toCharArray(s1));
+            }
+        }
+        try (Reader s2 = columnByNameAccessor.get(rs, columnName)) {
+            Assert.assertEquals(expectedNull, rs.wasNull());
+            if (expectedNull) {
+                Assert.assertNull(s2);
+            } else {
+                Assert.assertArrayEquals(expectedValue, IOUtils.toCharArray(s2));
+            }
+        }
+    }
+
+    private void verifyGetColumnAsBinaryStream(ResultSet rs, int columnIndex, String columnName, byte[] expectedValue,
+            GetColumnByIndex<InputStream> columnByIndexAccessor, GetColumnByName<InputStream> columnByNameAccessor)
+            throws SQLException, IOException {
+        boolean expectedNull = expectedValue == null;
+        try (InputStream s1 = columnByIndexAccessor.get(rs, columnIndex)) {
+            Assert.assertEquals(expectedNull, rs.wasNull());
+            if (expectedNull) {
+                Assert.assertNull(s1);
+            } else {
+                Assert.assertArrayEquals(expectedValue, IOUtils.toByteArray(s1));
+            }
+        }
+        try (InputStream s2 = columnByNameAccessor.get(rs, columnName)) {
+            Assert.assertEquals(expectedNull, rs.wasNull());
+            if (expectedNull) {
+                Assert.assertNull(s2);
+            } else {
+                Assert.assertArrayEquals(expectedValue, IOUtils.toByteArray(s2));
+            }
+        }
+    }
+
+    private void verifyReadColumnOfNumericType(ResultSet rs, int columnIndex, String columnName,
+            Number expectedNumericValue) throws SQLException {
+        String expectedStringValue = expectedNumericValue == null ? null : expectedNumericValue.toString();
+        Byte expectedByteValue = expectedNumericValue == null ? null : expectedNumericValue.byteValue();
+        Short expectedShortValue = expectedNumericValue == null ? null : expectedNumericValue.shortValue();
+        Integer expectedIntValue = expectedNumericValue == null ? null : expectedNumericValue.intValue();
+        Long expectedLongValue = expectedNumericValue == null ? null : expectedNumericValue.longValue();
+        Float expectedFloatValue = expectedNumericValue == null ? null : expectedNumericValue.floatValue();
+        Double expectedDoubleValue = expectedNumericValue == null ? null : expectedNumericValue.doubleValue();
+        BigDecimal expectedDecimalValue =
+                expectedNumericValue == null ? null : new BigDecimal(expectedStringValue.replace(".0", ""));
+        Boolean expectedBooleanValue = toBoolean(expectedNumericValue);
+        verifyGetColumnAsByte(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsShort(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsInt(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsLong(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsFloat(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsDouble(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsDecimal(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsBoolean(rs, columnIndex, columnName, expectedBooleanValue);
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedNumericValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedByteValue, Byte.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedShortValue, Short.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedIntValue, Integer.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedLongValue, Long.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedFloatValue, Float.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDoubleValue, Double.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDecimalValue, BigDecimal.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedBooleanValue, Boolean.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue, String.class);
+    }
+
+    private void verifyReadColumnOfStringType(ResultSet rs, int columnIndex, String columnName,
+            String expectedStringValue) throws SQLException, IOException {
+        char[] expectedCharArray = expectedStringValue == null ? null : expectedStringValue.toCharArray();
+        byte[] expectedUtf8Array =
+                expectedStringValue == null ? null : expectedStringValue.getBytes(StandardCharsets.UTF_8);
+        byte[] expectedUtf16Array =
+                expectedStringValue == null ? null : expectedStringValue.getBytes(StandardCharsets.UTF_16);
+        byte[] expectedAsciiArray =
+                expectedStringValue == null ? null : expectedStringValue.getBytes(StandardCharsets.US_ASCII);
+        verifyGetColumnAsCharacterStream(rs, columnIndex, columnName, expectedCharArray, ResultSet::getCharacterStream,
+                ResultSet::getCharacterStream);
+        verifyGetColumnAsCharacterStream(rs, columnIndex, columnName, expectedCharArray, ResultSet::getNCharacterStream,
+                ResultSet::getNCharacterStream);
+        verifyGetColumnAsBinaryStream(rs, columnIndex, columnName, expectedUtf8Array, ResultSet::getBinaryStream,
+                ResultSet::getBinaryStream);
+        verifyGetColumnAsBinaryStream(rs, columnIndex, columnName, expectedUtf16Array, ResultSet::getUnicodeStream,
+                ResultSet::getUnicodeStream);
+        verifyGetColumnAsBinaryStream(rs, columnIndex, columnName, expectedAsciiArray, ResultSet::getAsciiStream,
+                ResultSet::getAsciiStream);
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue, String.class);
+    }
+
+    private void verifyReadColumnOfBooleanType(ResultSet rs, int columnIndex, String columnName,
+            Boolean expectedBooleanValue) throws SQLException {
+        Number expectedNumberValue = expectedBooleanValue == null ? null : expectedBooleanValue ? 1 : 0;
+        String expectedStringValue = expectedBooleanValue == null ? null : Boolean.toString(expectedBooleanValue);
+        verifyGetColumnAsBoolean(rs, columnIndex, columnName, expectedBooleanValue);
+        verifyGetColumnAsByte(rs, columnIndex, columnName, expectedNumberValue);
+        verifyGetColumnAsShort(rs, columnIndex, columnName, expectedNumberValue);
+        verifyGetColumnAsInt(rs, columnIndex, columnName, expectedNumberValue);
+        verifyGetColumnAsLong(rs, columnIndex, columnName, expectedNumberValue);
+        verifyGetColumnAsFloat(rs, columnIndex, columnName, expectedNumberValue);
+        verifyGetColumnAsDouble(rs, columnIndex, columnName, expectedNumberValue);
+        verifyGetColumnAsDecimal(rs, columnIndex, columnName, expectedNumberValue);
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedBooleanValue);
+    }
+
+    private void verifyReadColumnOfDateType(ResultSet rs, int columnIndex, String columnName,
+            LocalDate expectedDateValue) throws SQLException {
+        LocalDateTime expectedDateTimeValue = expectedDateValue == null ? null : expectedDateValue.atStartOfDay();
+        String expectedStringValue = expectedDateValue == null ? null : expectedDateValue.toString();
+        verifyGetColumnAsSqlDate(rs, columnIndex, columnName, expectedDateValue);
+        verifyGetColumnAsSqlTimestamp(rs, columnIndex, columnName, expectedDateTimeValue);
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateValue, v -> ((java.sql.Date) v).toLocalDate());
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateValue, java.sql.Date.class, Date::toLocalDate);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateTimeValue, java.sql.Timestamp.class,
+                Timestamp::toLocalDateTime);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateValue, LocalDate.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateTimeValue, LocalDateTime.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue, String.class);
+    }
+
+    private void verifyReadColumnOfTimeType(ResultSet rs, int columnIndex, String columnName,
+            LocalTime expectedTimeValue) throws SQLException {
+        String expectedStringValue = expectedTimeValue == null ? null : expectedTimeValue.toString();
+        verifyGetColumnAsSqlTime(rs, columnIndex, columnName, expectedTimeValue);
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedTimeValue, v -> ((java.sql.Time) v).toLocalTime());
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedTimeValue, java.sql.Time.class,
+                java.sql.Time::toLocalTime);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedTimeValue, LocalTime.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue, String.class);
+    }
+
+    private void verifyReadColumnOfDatetimeType(ResultSet rs, int columnIndex, String columnName,
+            LocalDateTime expectedDateTimeValue) throws SQLException {
+        LocalDate expectedDateValue = expectedDateTimeValue == null ? null : expectedDateTimeValue.toLocalDate();
+        LocalTime expectedTimeValue = expectedDateTimeValue == null ? null : expectedDateTimeValue.toLocalTime();
+        String expectedStringValue = expectedDateTimeValue == null ? null : expectedDateTimeValue.toString();
+        verifyGetColumnAsSqlTimestamp(rs, columnIndex, columnName, expectedDateTimeValue);
+        verifyGetColumnAsSqlDate(rs, columnIndex, columnName, expectedDateValue);
+        verifyGetColumnAsSqlTime(rs, columnIndex, columnName, expectedTimeValue);
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateTimeValue,
+                v -> ((java.sql.Timestamp) v).toLocalDateTime());
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateTimeValue, java.sql.Timestamp.class,
+                java.sql.Timestamp::toLocalDateTime);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateValue, java.sql.Date.class,
+                java.sql.Date::toLocalDate);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedTimeValue, java.sql.Time.class,
+                java.sql.Time::toLocalTime);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateTimeValue, LocalDateTime.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDateValue, LocalDate.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedTimeValue, LocalTime.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue, String.class);
+    }
+
+    private void verifyReadColumnOfYearMonthDurationType(ResultSet rs, int columnIndex, String columnName,
+            Period expectedPeriodValue) throws SQLException {
+        String expectedStringValue = expectedPeriodValue == null ? null : expectedPeriodValue.toString();
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedPeriodValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedPeriodValue, Period.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue, String.class);
+    }
+
+    private void verifyReadColumnOfDayTimeDurationType(ResultSet rs, int columnIndex, String columnName,
+            Duration expectedDurationValue) throws SQLException {
+        String expectedStringValue = expectedDurationValue == null ? null : expectedDurationValue.toString();
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDurationValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDurationValue, Duration.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue, String.class);
+    }
+
+    private void verifyReadColumnOfDurationType(ResultSet rs, int columnIndex, String columnName,
+            Period expectedPeriodValue, Duration expectedDurationValue) throws SQLException {
+        String expectedStringValue = expectedPeriodValue == null && expectedDurationValue == null ? null
+                : expectedPeriodValue + String.valueOf(expectedDurationValue).substring(1);
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedPeriodValue, Period.class);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedDurationValue, Duration.class);
+    }
+
+    private void verifyReadColumnOfUuidType(ResultSet rs, int columnIndex, String columnName, UUID expectedUuidValue)
+            throws SQLException {
+        String expectedStringValue = expectedUuidValue == null ? null : expectedUuidValue.toString();
+        verifyGetColumnAsString(rs, columnIndex, columnName, expectedStringValue);
+        verifyGetColumnAsObject(rs, columnIndex, columnName, expectedUuidValue);
+    }
+
+    public void testWrapper() throws SQLException {
+        try (Connection c = createConnection(); CloseablePair<Statement, ResultSet> p = executeQuery(c, Q2)) {
+            ResultSet rs = p.getSecond();
+            Assert.assertTrue(rs.isWrapperFor(ADBResultSet.class));
+            Assert.assertNotNull(rs.unwrap(ADBResultSet.class));
+        }
+    }
+
+    interface GetColumnByIndex<R> {
+        R get(ResultSet rs, int columnIndex) throws SQLException;
+    }
+
+    interface GetColumnByIndexWithParam<R, T> {
+        R get(ResultSet rs, int columnIndex, T param) throws SQLException;
+    }
+
+    interface GetColumnByName<R> {
+        R get(ResultSet rs, String columnName) throws SQLException;
+    }
+
+    interface GetColumnByNameWithParam<R, T> {
+        R get(ResultSet rs, String columnName, T param) throws SQLException;
+    }
+
+    static Boolean toBoolean(Number v) {
+        if (v == null) {
+            return null;
+        }
+        switch (v.toString()) {
+            case "0":
+            case "0.0":
+                return false;
+            default:
+                return true;
+        }
+    }
+
+    static class JdbcPreparedStatementResultSetTester extends JdbcResultSetTester {
+        @Override
+        protected CloseablePair<Statement, ResultSet> executeQuery(Connection c, String query) throws SQLException {
+            PreparedStatement s = c.prepareStatement(query);
+            try {
+                ResultSet rs = s.executeQuery();
+                return new CloseablePair<>(s, rs);
+            } catch (SQLException e) {
+                s.close();
+                throw e;
+            }
+        }
+    }
+
+    static class JdbcStatementResultSetTester extends JdbcResultSetTester {
+        @Override
+        protected CloseablePair<Statement, ResultSet> executeQuery(Connection c, String query) throws SQLException {
+            Statement s = c.createStatement();
+            try {
+                ResultSet rs = s.executeQuery(query);
+                return new CloseablePair<>(s, rs);
+            } catch (SQLException e) {
+                s.close();
+                throw e;
+            }
+        }
+    }
+}
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcStatementParameterTester.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcStatementParameterTester.java
new file mode 100644
index 0000000..b87527e
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcStatementParameterTester.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.asterix.test.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.JDBCType;
+import java.sql.ParameterMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.Period;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.function.BiPredicate;
+
+import org.junit.Assert;
+
+class JdbcStatementParameterTester extends JdbcTester {
+
+    public void testParameterBinding() throws SQLException {
+        String[] sqlppValues = new String[] { "int8('10')", "int16('20')", "int32('30')", "int64('40')", "float('1.5')",
+                "double('2.25')", "true", "'abc'", "date('2000-10-20')", "time('02:03:04')",
+                "datetime('2000-10-20T02:03:04')", "get_year_month_duration(duration_from_months(2))",
+                "get_day_time_duration(duration_from_ms(1234))", "uuid('5c848e5c-6b6a-498f-8452-8847a2957421')" };
+        try (Connection c = createConnection()) {
+            Byte i1 = (byte) 10;
+            verifyParameterBinding(c, sqlppValues, i1, PreparedStatement::setByte, ResultSet::getByte);
+            verifyParameterBinding(c, sqlppValues, i1, PreparedStatement::setObject, ResultSet::getByte);
+
+            Short i2 = (short) 20;
+            verifyParameterBinding(c, sqlppValues, i2, PreparedStatement::setShort, ResultSet::getShort);
+            verifyParameterBinding(c, sqlppValues, i2, PreparedStatement::setObject, ResultSet::getShort);
+
+            Integer i4 = 30;
+            verifyParameterBinding(c, sqlppValues, i4, PreparedStatement::setInt, ResultSet::getInt);
+            verifyParameterBinding(c, sqlppValues, i4, PreparedStatement::setObject, ResultSet::getInt);
+
+            Long i8 = 40L;
+            verifyParameterBinding(c, sqlppValues, i8, PreparedStatement::setLong, ResultSet::getLong);
+            verifyParameterBinding(c, sqlppValues, i8, PreparedStatement::setObject, ResultSet::getLong);
+
+            Float r4 = 1.5f;
+            verifyParameterBinding(c, sqlppValues, r4, PreparedStatement::setFloat, ResultSet::getFloat);
+            verifyParameterBinding(c, sqlppValues, r4, PreparedStatement::setObject, ResultSet::getFloat);
+
+            Double r8 = 2.25;
+            verifyParameterBinding(c, sqlppValues, r8, PreparedStatement::setDouble, ResultSet::getDouble);
+            verifyParameterBinding(c, sqlppValues, r8, PreparedStatement::setObject, ResultSet::getDouble);
+
+            BigDecimal dec = new BigDecimal("2.25");
+            verifyParameterBinding(c, sqlppValues, dec, PreparedStatement::setBigDecimal, ResultSet::getBigDecimal);
+            verifyParameterBinding(c, sqlppValues, dec, PreparedStatement::setObject, ResultSet::getBigDecimal);
+
+            Boolean b = true;
+            verifyParameterBinding(c, sqlppValues, b, PreparedStatement::setBoolean, ResultSet::getBoolean);
+            verifyParameterBinding(c, sqlppValues, b, PreparedStatement::setObject, ResultSet::getBoolean);
+
+            String s = "abc";
+            verifyParameterBinding(c, sqlppValues, s, PreparedStatement::setString, ResultSet::getString);
+            verifyParameterBinding(c, sqlppValues, s, PreparedStatement::setObject, ResultSet::getString);
+            verifyParameterBinding(c, sqlppValues, s, PreparedStatement::setNString, ResultSet::getString);
+
+            LocalDate date = LocalDate.of(2000, 10, 20);
+            verifyParameterBinding(c, sqlppValues, java.sql.Date.valueOf(date), PreparedStatement::setDate,
+                    ResultSet::getDate);
+            verifyParameterBinding(c, sqlppValues, java.sql.Date.valueOf(date), PreparedStatement::setObject,
+                    ResultSet::getDate);
+            verifyParameterBinding(c, sqlppValues, date, PreparedStatement::setObject,
+                    (rs, i) -> rs.getObject(i, LocalDate.class));
+
+            LocalTime time = LocalTime.of(2, 3, 4);
+            verifyParameterBinding(c, sqlppValues, java.sql.Time.valueOf(time), PreparedStatement::setTime,
+                    ResultSet::getTime, JdbcStatementParameterTester::sqlTimeEquals);
+            verifyParameterBinding(c, sqlppValues, java.sql.Time.valueOf(time), PreparedStatement::setObject,
+                    ResultSet::getTime, JdbcStatementParameterTester::sqlTimeEquals);
+            verifyParameterBinding(c, sqlppValues, time, PreparedStatement::setObject,
+                    (rs, i) -> rs.getObject(i, LocalTime.class));
+
+            LocalDateTime datetime = LocalDateTime.of(date, time);
+            verifyParameterBinding(c, sqlppValues, java.sql.Timestamp.valueOf(datetime),
+                    PreparedStatement::setTimestamp, ResultSet::getTimestamp);
+            verifyParameterBinding(c, sqlppValues, java.sql.Timestamp.valueOf(datetime), PreparedStatement::setObject,
+                    ResultSet::getTimestamp);
+            verifyParameterBinding(c, sqlppValues, datetime, PreparedStatement::setObject,
+                    (rs, i) -> rs.getObject(i, LocalDateTime.class));
+
+            Period ymDuration = Period.ofMonths(2);
+            verifyParameterBinding(c, sqlppValues, ymDuration, PreparedStatement::setObject,
+                    (rs, i) -> rs.getObject(i, Period.class));
+
+            Duration dtDuration = Duration.ofMillis(1234);
+            verifyParameterBinding(c, sqlppValues, dtDuration, PreparedStatement::setObject,
+                    (rs, i) -> rs.getObject(i, Duration.class));
+
+            UUID uuid = UUID.fromString("5c848e5c-6b6a-498f-8452-8847a2957421");
+            verifyParameterBinding(c, sqlppValues, uuid, PreparedStatement::setObject, ResultSet::getObject);
+        }
+    }
+
+    private <T> void verifyParameterBinding(Connection c, String[] sqlppValues, T value, SetParameterByIndex<T> setter,
+            JdbcResultSetTester.GetColumnByIndex<T> getter) throws SQLException {
+        verifyParameterBinding(c, sqlppValues, value, setter, getter, Objects::equals);
+    }
+
+    private <T> void verifyParameterBinding(Connection c, String[] sqlppValues, T value, SetParameterByIndex<T> setter,
+            JdbcResultSetTester.GetColumnByIndex<T> getter, BiPredicate<T, T> cmp) throws SQLException {
+        try (PreparedStatement s =
+                c.prepareStatement(String.format("select ? from [%s] v where v = ?", String.join(",", sqlppValues)))) {
+            for (int i = 1; i <= 2; i++) {
+                setter.set(s, i, value);
+            }
+            try (ResultSet rs = s.executeQuery()) {
+                if (rs.next()) {
+                    T outValue = getter.get(rs, 1);
+                    if (!cmp.test(value, outValue)) {
+                        Assert.fail(String.format("%s != %s", value, outValue));
+                    }
+                } else {
+                    Assert.fail(String.format("Empty result (expected value '%s' was not returned)", value));
+                }
+            }
+        }
+    }
+
+    public void testParameterMetadata() throws SQLException {
+        String q = "select r from range(1, 10) r where r = ? or r = ? or r = ?";
+        int paramCount = 3;
+        try (Connection c = createConnection(); PreparedStatement s = c.prepareStatement(q)) {
+            ParameterMetaData pmd = s.getParameterMetaData();
+            Assert.assertEquals(paramCount, pmd.getParameterCount());
+            for (int i = 1; i <= paramCount; i++) {
+                Assert.assertEquals(JDBCType.OTHER.getVendorTypeNumber().intValue(), pmd.getParameterType(i));
+                Assert.assertEquals("any", pmd.getParameterTypeName(i));
+                Assert.assertEquals(ParameterMetaData.parameterModeIn, pmd.getParameterMode(i));
+            }
+        }
+    }
+
+    interface SetParameterByIndex<T> {
+        void set(PreparedStatement s, int paramIndex, T paramValue) throws SQLException;
+    }
+
+    private static boolean sqlTimeEquals(java.sql.Time v1, java.sql.Time v2) {
+        // java.sql.Time.equals() compares millis since epoch,
+        // but we only want to compare time components
+        return v1.toLocalTime().equals(v2.toLocalTime());
+    }
+}
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcStatementTester.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcStatementTester.java
new file mode 100644
index 0000000..e2729ef
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcStatementTester.java
@@ -0,0 +1,266 @@
+/*
+ * 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.asterix.test.jdbc;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.jdbc.core.ADBStatement;
+import org.junit.Assert;
+
+class JdbcStatementTester extends JdbcTester {
+
+    public void testLifecycle() throws SQLException {
+        Connection c = createConnection();
+        Statement s = c.createStatement();
+        Assert.assertFalse(s.isClosed());
+        Assert.assertSame(c, s.getConnection());
+
+        s.close();
+        Assert.assertTrue(s.isClosed());
+
+        // ok to call close() on a closed statement
+        s.close();
+        Assert.assertTrue(s.isClosed());
+    }
+
+    public void testAutoCloseOnConnectionClose() throws SQLException {
+        Connection c = createConnection();
+        // check that a statement is automatically closed when the connection is closed
+        Statement s = c.createStatement();
+        Assert.assertFalse(s.isClosed());
+        c.close();
+        Assert.assertTrue(s.isClosed());
+    }
+
+    public void testCloseOnCompletion() throws SQLException {
+        try (Connection c = createConnection()) {
+            Statement s = c.createStatement();
+            Assert.assertFalse(s.isCloseOnCompletion());
+            s.closeOnCompletion();
+            Assert.assertTrue(s.isCloseOnCompletion());
+            Assert.assertFalse(s.isClosed());
+            ResultSet rs = s.executeQuery(Q1);
+            Assert.assertTrue(rs.next());
+            Assert.assertFalse(rs.next());
+            rs.close();
+            Assert.assertTrue(s.isClosed());
+        }
+    }
+
+    public void testExecuteQuery() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            // Query -> ok
+            try (ResultSet rs = s.executeQuery(Q1)) {
+                Assert.assertTrue(rs.next());
+                Assert.assertEquals(1, rs.getMetaData().getColumnCount());
+                Assert.assertEquals(V1, rs.getInt(1));
+                Assert.assertFalse(rs.next());
+                Assert.assertFalse(rs.isClosed());
+            }
+
+            // DDL -> error
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testExecuteQuery");
+            try {
+                s.executeQuery(printCreateDataverse(dataverse));
+                Assert.fail("DDL did not fail in executeQuery()");
+            } catch (SQLException e) {
+                String msg = e.getMessage();
+                Assert.assertTrue(msg, msg.contains(ErrorCode.PROHIBITED_STATEMENT_CATEGORY.errorCode()));
+            }
+
+            // DML -> error
+            String dataset = "ds1";
+            s.execute(printCreateDataverse(dataverse));
+            s.execute(printCreateDataset(dataverse, dataset));
+            try {
+                s.executeQuery(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+                Assert.fail("DML did not fail in executeQuery()");
+            } catch (SQLException e) {
+                String msg = e.getMessage();
+                Assert.assertTrue(msg, msg.contains(ErrorCode.PROHIBITED_STATEMENT_CATEGORY.errorCode()));
+            }
+
+            // Cleanup
+            s.execute(printDropDataverse(dataverse));
+        }
+    }
+
+    public void testExecuteUpdate() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            // DDL -> ok
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testExecuteUpdate");
+            int res = s.executeUpdate(printCreateDataverse(dataverse));
+            Assert.assertEquals(0, res);
+            String dataset = "ds1";
+            res = s.executeUpdate(printCreateDataset(dataverse, dataset));
+            Assert.assertEquals(0, res);
+
+            // DML -> ok
+            res = s.executeUpdate(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+            // currently, DML statements always return update count = 1
+            Assert.assertEquals(1, res);
+
+            // Query -> error
+            try {
+                s.executeUpdate(Q1);
+                Assert.fail("Query did not fail in executeUpdate()");
+            } catch (SQLException e) {
+                String msg = e.getMessage();
+                Assert.assertTrue(msg, msg.contains("Invalid statement category"));
+            }
+
+            // Cleanup
+            s.executeUpdate(printDropDataverse(dataverse));
+        }
+    }
+
+    public void testExecute() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            // Query -> ok
+            boolean res = s.execute(Q1);
+            Assert.assertTrue(res);
+            Assert.assertEquals(-1, s.getUpdateCount());
+            try (ResultSet rs = s.getResultSet()) {
+                Assert.assertTrue(rs.next());
+                Assert.assertEquals(1, rs.getMetaData().getColumnCount());
+                Assert.assertEquals(V1, rs.getInt(1));
+                Assert.assertFalse(rs.next());
+                Assert.assertFalse(rs.isClosed());
+            }
+
+            // DDL -> ok
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testExecute");
+            res = s.execute(printCreateDataverse(dataverse));
+            Assert.assertFalse(res);
+            Assert.assertEquals(0, s.getUpdateCount());
+            String dataset = "ds1";
+            res = s.execute(printCreateDataset(dataverse, dataset));
+            Assert.assertFalse(res);
+
+            // DML -> ok
+            res = s.execute(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+            Assert.assertFalse(res);
+            // currently, DML statements always return update count = 1
+            Assert.assertEquals(1, s.getUpdateCount());
+
+            // Cleanup
+            s.execute(printDropDataverse(dataverse));
+        }
+    }
+
+    public void testGetResultSet() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            // Query
+            boolean res = s.execute(Q1);
+            Assert.assertTrue(res);
+            ResultSet rs = s.getResultSet();
+            Assert.assertFalse(rs.isClosed());
+            Assert.assertTrue(rs.next());
+            Assert.assertFalse(s.getMoreResults()); // closes current ResultSet
+            Assert.assertTrue(rs.isClosed());
+
+            res = s.execute(Q1);
+            Assert.assertTrue(res);
+            rs = s.getResultSet();
+            Assert.assertFalse(rs.isClosed());
+            Assert.assertTrue(rs.next());
+            Assert.assertFalse(s.getMoreResults(Statement.KEEP_CURRENT_RESULT));
+            Assert.assertFalse(rs.isClosed());
+            rs.close();
+
+            // DDL
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testGetResultSet");
+            res = s.execute(printCreateDataverse(dataverse));
+            Assert.assertFalse(res);
+            Assert.assertNull(s.getResultSet());
+            Assert.assertFalse(s.getMoreResults());
+
+            String dataset = "ds1";
+            res = s.execute(printCreateDataset(dataverse, dataset));
+            Assert.assertFalse(res);
+
+            // DML
+            res = s.execute(printInsert(dataverse, dataset, dataGen("x", 1, 2)));
+            Assert.assertFalse(res);
+            Assert.assertNull(s.getResultSet());
+            Assert.assertFalse(s.getMoreResults());
+        }
+    }
+
+    public void testMaxRows() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            List<String> dataverse = Arrays.asList(getClass().getSimpleName(), "testMaxRows");
+            String dataset = "ds1";
+            String field = "x";
+            s.execute(printCreateDataverse(dataverse));
+            s.execute(printCreateDataset(dataverse, dataset));
+            s.execute(printInsert(dataverse, dataset, dataGen(field, 1, 2, 3)));
+
+            s.setMaxRows(2);
+            Assert.assertEquals(2, s.getMaxRows());
+            try (ResultSet rs = s.executeQuery(String.format("select %s from %s.%s", field,
+                    printDataverseName(dataverse), printIdentifier(dataset)))) {
+                Assert.assertTrue(rs.next());
+                Assert.assertTrue(rs.next());
+                Assert.assertFalse(rs.next());
+            }
+        }
+    }
+
+    public void testWarnings() throws SQLException {
+        try (Connection c = createConnection();
+                Statement s = c.createStatement();
+                ResultSet rs = s.executeQuery("select double('x'), bigint('y')")) { // --> NULL with warning
+            Assert.assertTrue(rs.next());
+            rs.getDouble(1);
+            Assert.assertTrue(rs.wasNull());
+            rs.getLong(2);
+            Assert.assertTrue(rs.wasNull());
+
+            SQLWarning w = s.getWarnings();
+            Assert.assertNotNull(w);
+            String msg = w.getMessage();
+            Assert.assertTrue(msg, msg.contains(ErrorCode.INVALID_FORMAT.errorCode()));
+
+            SQLWarning w2 = w.getNextWarning();
+            Assert.assertNotNull(w2);
+            String msg2 = w.getMessage();
+            Assert.assertTrue(msg2, msg2.contains(ErrorCode.INVALID_FORMAT.errorCode()));
+
+            Assert.assertNull(w2.getNextWarning());
+            s.clearWarnings();
+            Assert.assertNull(s.getWarnings());
+        }
+    }
+
+    public void testWrapper() throws SQLException {
+        try (Connection c = createConnection(); Statement s = c.createStatement()) {
+            Assert.assertTrue(s.isWrapperFor(ADBStatement.class));
+            Assert.assertNotNull(s.unwrap(ADBStatement.class));
+        }
+    }
+}
diff --git a/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcTester.java b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcTester.java
new file mode 100644
index 0000000..d631e0f
--- /dev/null
+++ b/asterixdb-jdbc/asterix-jdbc-test/src/test/java/org/apache/asterix/test/jdbc/JdbcTester.java
@@ -0,0 +1,259 @@
+/*
+ * 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.asterix.test.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.JDBCType;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.time.Period;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.junit.Assert;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+abstract class JdbcTester {
+
+    static final String DEFAULT_DATAVERSE_NAME = "Default";
+
+    static final String METADATA_DATAVERSE_NAME = "Metadata";
+
+    static final List<String> BUILT_IN_DATAVERSE_NAMES = Arrays.asList(DEFAULT_DATAVERSE_NAME, METADATA_DATAVERSE_NAME);
+
+    static final String SQL_STATE_CONNECTION_CLOSED = "08003";
+
+    static final char IDENTIFIER_QUOTE = '`';
+
+    static final int V1 = 42;
+
+    static final String Q1 = printSelect(V1);
+
+    static final String Q2 = "select r x, r * 11 y from range(1, 9) r order by r";
+
+    static final String Q3_PROJECT = "int8(v) c1_i1, int16(v) c2_i2, int32(v) c3_i4, int64(v) c4_i8, float(v) c5_r4, "
+            + "double(v) c6_r8, 'a' || string(v) c7_s, boolean(v+2) c8_b, date_from_unix_time_in_days(v) c9_d, "
+            + "time_from_unix_time_in_ms((v+3)*1000) c10_t, datetime_from_unix_time_in_secs(v) c11_dt,"
+            + "get_year_month_duration(duration_from_months(v)) c12_um, "
+            + "get_day_time_duration(duration_from_ms(v)) c13_ut, "
+            + "duration('P'||string(v+3)||'MT'||string(v+3)||'S') c14_uu, "
+            + "uuid('5c848e5c-6b6a-498f-8452-8847a295742' || string(v+3)) c15_id";
+
+    static final String Q3 = String.format("select %s from range(-1, 1) r let v=nullif(r,0)*2 order by r", Q3_PROJECT);
+
+    static String[] Q3_COLUMNS = new String[] { "c1_i1", "c2_i2", "c3_i4", "c4_i8", "c5_r4", "c6_r8", "c7_s", "c8_b",
+            "c9_d", "c10_t", "c11_dt", "c12_um", "c13_ut", "c14_uu", "c15_id" };
+
+    static JDBCType[] Q3_COLUMN_TYPES_JDBC = new JDBCType[] { JDBCType.TINYINT, JDBCType.SMALLINT, JDBCType.INTEGER,
+            JDBCType.BIGINT, JDBCType.REAL, JDBCType.DOUBLE, JDBCType.VARCHAR, JDBCType.BOOLEAN, JDBCType.DATE,
+            JDBCType.TIME, JDBCType.TIMESTAMP, JDBCType.OTHER, JDBCType.OTHER, JDBCType.OTHER, JDBCType.OTHER };
+
+    static String[] Q3_COLUMN_TYPES_ADB = new String[] { "int8", "int16", "int32", "int64", "float", "double", "string",
+            "boolean", "date", "time", "datetime", "year-month-duration", "day-time-duration", "duration", "uuid" };
+
+    static Class<?>[] Q3_COLUMN_TYPES_JAVA = new Class<?>[] { Byte.class, Short.class, Integer.class, Long.class,
+            Float.class, Double.class, String.class, Boolean.class, java.sql.Date.class, java.sql.Time.class,
+            java.sql.Timestamp.class, Period.class, Duration.class, String.class, UUID.class };
+
+    protected JdbcTestContext testContext;
+
+    protected JdbcTester() {
+    }
+
+    void setTestContext(JdbcTestContext testContext) {
+        this.testContext = Objects.requireNonNull(testContext);
+    }
+
+    static JdbcTestContext createTestContext(String host, int port) {
+        return new JdbcTestContext(host, port);
+    }
+
+    protected Connection createConnection() throws SQLException {
+        return DriverManager.getConnection(testContext.getJdbcUrl());
+    }
+
+    protected Connection createConnection(String dataverseName) throws SQLException {
+        return createConnection(Collections.singletonList(dataverseName));
+    }
+
+    protected Connection createConnection(List<String> dataverseName) throws SQLException {
+        return DriverManager.getConnection(testContext.getJdbcUrl(getCanonicalDataverseName(dataverseName)));
+    }
+
+    protected static String getCanonicalDataverseName(List<String> dataverseName) {
+        return String.join("/", dataverseName);
+    }
+
+    protected static String printDataverseName(List<String> dataverseName) {
+        return dataverseName.stream().map(JdbcTester::printIdentifier).collect(Collectors.joining("."));
+    }
+
+    protected static String printIdentifier(String ident) {
+        return IDENTIFIER_QUOTE + ident + IDENTIFIER_QUOTE;
+    }
+
+    protected static String printCreateDataverse(List<String> dataverseName) {
+        return String.format("create dataverse %s", printDataverseName(dataverseName));
+    }
+
+    protected static String printDropDataverse(List<String> dataverseName) {
+        return String.format("drop dataverse %s", printDataverseName(dataverseName));
+    }
+
+    protected static String printCreateDataset(List<String> dataverseName, String datasetName) {
+        return String.format("create dataset %s.%s(_id uuid) open type primary key _id autogenerated",
+                printDataverseName(dataverseName), printIdentifier(datasetName));
+    }
+
+    protected static String printCreateDataset(List<String> dataverseName, String datasetName, List<String> fieldNames,
+            List<String> fieldTypes, int pkLen) {
+        return String.format("create dataset %s.%s(%s) open type primary key %s", printDataverseName(dataverseName),
+                printIdentifier(datasetName), printSchema(fieldNames, fieldTypes),
+                printIdentifierList(fieldNames.subList(0, pkLen)));
+    }
+
+    protected static String printCreateView(List<String> dataverseName, String viewName, List<String> fieldNames,
+            List<String> fieldTypes, int pkLen, List<String> fkRefs, String viewQuery) {
+        List<String> pkFieldNames = fieldNames.subList(0, pkLen);
+        String pkDecl = String.format(" primary key (%s) not enforced", printIdentifierList(pkFieldNames));
+        String fkDecl =
+                fkRefs.stream()
+                        .map(fkRef -> String.format("foreign key (%s) references %s not enforced",
+                                printIdentifierList(pkFieldNames), printIdentifier(fkRef)))
+                        .collect(Collectors.joining(" "));
+        return String.format("create view %s.%s(%s) default null %s %s as %s", printDataverseName(dataverseName),
+                printIdentifier(viewName), printSchema(fieldNames, fieldTypes), pkDecl, fkDecl, viewQuery);
+    }
+
+    protected static String printSchema(List<String> fieldNames, List<String> fieldTypes) {
+        StringBuilder schema = new StringBuilder(128);
+        for (int i = 0, n = fieldNames.size(); i < n; i++) {
+            if (i > 0) {
+                schema.append(',');
+            }
+            schema.append(printIdentifier(fieldNames.get(i))).append(' ').append(fieldTypes.get(i));
+        }
+        return schema.toString();
+    }
+
+    protected static String printIdentifierList(List<String> fieldNames) {
+        return fieldNames.stream().map(JdbcTester::printIdentifier).collect(Collectors.joining(","));
+    }
+
+    protected static String printInsert(List<String> dataverseName, String datasetName, ArrayNode values) {
+        return String.format("insert into %s.%s (%s)", printDataverseName(dataverseName), printIdentifier(datasetName),
+                values);
+    }
+
+    protected static String printSelect(Object... values) {
+        return String.format("select %s", Arrays.stream(values).map(String::valueOf).collect(Collectors.joining(",")));
+    }
+
+    protected static ArrayNode dataGen(String fieldName1, Object... data1) {
+        ObjectMapper om = new ObjectMapper();
+        ArrayNode values = om.createArrayNode();
+        for (Object v : data1) {
+            ObjectNode obj = om.createObjectNode();
+            obj.putPOJO(fieldName1, v);
+            values.add(obj);
+        }
+        return values;
+    }
+
+    protected static <T> void assertErrorOnClosed(T param, JdbcConnectionTester.JdbcRunnable<T> cmd,
+            String description) {
+        try {
+            cmd.run(param);
+            Assert.fail(String.format("Unexpected: %s succeeded on a closed %s", description,
+                    param.getClass().getSimpleName()));
+        } catch (SQLException e) {
+            String msg = e.getMessage();
+            Assert.assertTrue(msg, msg.contains("closed"));
+        }
+    }
+
+    static class JdbcTestContext {
+
+        private static final String JDBC_URL_TEMPLATE = "jdbc:asterixdb://%s:%d";
+
+        private final String jdbcUrl;
+
+        private JdbcTestContext(String host, int port) {
+            jdbcUrl = String.format(JDBC_URL_TEMPLATE, host, port);
+        }
+
+        public String getJdbcUrl() {
+            return jdbcUrl;
+        }
+
+        public String getJdbcUrl(String dataverseName) {
+            return jdbcUrl + '/' + dataverseName;
+        }
+    }
+
+    interface JdbcRunnable<T> {
+        void run(T param) throws SQLException;
+    }
+
+    interface JdbcPredicate<T> {
+        boolean test(T param) throws SQLException;
+    }
+
+    static class CloseablePair<K extends AutoCloseable, V extends AutoCloseable> extends Pair<K, V>
+            implements AutoCloseable {
+        CloseablePair(K first, V second) {
+            super(first, second);
+        }
+
+        @Override
+        public void close() throws SQLException {
+            try {
+                if (second != null) {
+                    try {
+                        second.close();
+                    } catch (SQLException e) {
+                        throw e;
+                    } catch (Exception e) {
+                        throw new SQLException(e);
+                    }
+                }
+            } finally {
+                if (first != null) {
+                    try {
+                        first.close();
+                    } catch (SQLException e) {
+                        throw e;
+                    } catch (Exception e) {
+                        throw new SQLException(e);
+                    }
+                }
+            }
+        }
+    }
+}
diff --git a/asterixdb-jdbc/pom.xml b/asterixdb-jdbc/pom.xml
index 25d007d..a2c7d06 100644
--- a/asterixdb-jdbc/pom.xml
+++ b/asterixdb-jdbc/pom.xml
@@ -485,4 +485,19 @@
     <module>asterix-jdbc-driver</module>
     <module>asterix-jdbc-taco</module>
   </modules>
+
+  <profiles>
+    <profile>
+      <id>asterix-jdbc-test-enabled</id>
+      <activation>
+        <file>
+        <exists>${basedir}/../../asterix-app/pom.xml</exists>
+        </file>
+      </activation>
+      <modules>
+        <module>asterix-jdbc-test</module>
+      </modules>
+    </profile>
+  </profiles>
+
 </project>
\ No newline at end of file