propertiesFromUrl = getUrlsArgs(url);
+ properties.putAll(propertiesFromUrl);
+ }
+
+ try {
+ return ArrowFlightConnection.createNewConnection(
+ this,
+ factory,
+ url,
+ lowerCasePropertyKeys(properties),
+ new RootAllocator(Long.MAX_VALUE));
+ } catch (final FlightRuntimeException e) {
+ throw new SQLException("Failed to connect.", e);
+ }
+ }
+
+ @Override
+ protected String getFactoryClassName(final JdbcVersion jdbcVersion) {
+ return ArrowFlightJdbcFactory.class.getName();
+ }
+
+ @Override
+ protected DriverVersion createDriverVersion() {
+ if (version == null) {
+ final InputStream flightProperties = this.getClass().getResourceAsStream("/properties/flight.properties");
+ if (flightProperties == null) {
+ throw new RuntimeException("Flight Properties not found. Ensure the JAR was built properly.");
+ }
+ try (final Reader reader = new BufferedReader(new InputStreamReader(flightProperties, StandardCharsets.UTF_8))) {
+ final Properties properties = new Properties();
+ properties.load(reader);
+
+ final String parentName = properties.getProperty("org.apache.arrow.flight.name");
+ final String parentVersion = properties.getProperty("org.apache.arrow.flight.version");
+ final String[] pVersion = parentVersion.split("\\.");
+
+ final int parentMajorVersion = Integer.parseInt(pVersion[0]);
+ final int parentMinorVersion = Integer.parseInt(pVersion[1]);
+
+ final String childName = properties.getProperty("org.apache.arrow.flight.jdbc-driver.name");
+ final String childVersion = properties.getProperty("org.apache.arrow.flight.jdbc-driver.version");
+ final String[] cVersion = childVersion.split("\\.");
+
+ final int childMajorVersion = Integer.parseInt(cVersion[0]);
+ final int childMinorVersion = Integer.parseInt(cVersion[1]);
+
+ version = new DriverVersion(
+ childName,
+ childVersion,
+ parentName,
+ parentVersion,
+ true,
+ childMajorVersion,
+ childMinorVersion,
+ parentMajorVersion,
+ parentMinorVersion);
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to load driver version.", e);
+ }
+ }
+
+ return version;
+ }
+
+ @Override
+ public Meta createMeta(final AvaticaConnection connection) {
+ return new ArrowFlightMetaImpl(connection);
+ }
+
+ @Override
+ protected String getConnectStringPrefix() {
+ return CONNECT_STRING_PREFIX;
+ }
+
+ @Override
+ public boolean acceptsURL(final String url) {
+ Preconditions.checkNotNull(url);
+ return url.startsWith(CONNECT_STRING_PREFIX) || url.startsWith(CONNECT_STRING_PREFIX_DEPRECATED);
+ }
+
+ /**
+ * Parses the provided url based on the format this driver accepts, retrieving
+ * arguments after the {@link #CONNECT_STRING_PREFIX}.
+ *
+ * This method gets the args if the provided URL follows this pattern:
+ * {@code jdbc:arrow-flight-sql://:[/?key1=val1&key2=val2&(...)]}
+ *
+ *
+ *
+ * | Group |
+ * Definition |
+ * Value |
+ *
+ *
+ * | ? — inaccessible |
+ * {@link #getConnectStringPrefix} |
+ *
+ * the URL prefix accepted by this driver, i.e.,
+ * {@code "jdbc:arrow-flight-sql://"}
+ * |
+ *
+ *
+ * | 1 |
+ * IPv4 host name |
+ *
+ * first word after previous group and before "{@code :}"
+ * |
+ *
+ *
+ * | 2 |
+ * IPv4 port number |
+ *
+ * first number after previous group and before "{@code /?}"
+ * |
+ *
+ *
+ * | 3 |
+ * custom call parameters |
+ *
+ * all parameters provided after "{@code /?}" — must follow the
+ * pattern: "{@code key=value}" with "{@code &}" separating a
+ * parameter from another
+ * |
+ *
+ *
+ *
+ * @param url The url to parse.
+ * @return the parsed arguments.
+ * @throws SQLException If an error occurs while trying to parse the URL.
+ */
+ @VisibleForTesting // ArrowFlightJdbcDriverTest
+ Map getUrlsArgs(String url)
+ throws SQLException {
+
+ /*
+ *
+ * Perhaps this logic should be inside a utility class, separated from this
+ * one, so as to better delegate responsibilities and concerns throughout
+ * the code and increase maintainability.
+ *
+ * =====
+ *
+ * Keep in mind that the URL must ALWAYS follow the pattern:
+ * "jdbc:arrow-flight-sql://:[/?param1=value1¶m2=value2&(...)]."
+ *
+ */
+
+ final Properties resultMap = new Properties();
+ url = replaceSemiColons(url);
+
+ if (!url.startsWith("jdbc:")) {
+ throw new SQLException("Connection string must start with 'jdbc:'. Expected format: " +
+ CONNECTION_STRING_EXPECTED);
+ }
+
+ // It's necessary to use a string without "jdbc:" at the beginning to be parsed as a valid URL.
+ url = url.substring(5);
+
+ final URI uri;
+
+ try {
+ uri = URI.create(url);
+ } catch (final IllegalArgumentException e) {
+ throw new SQLException("Malformed/invalid URL!", e);
+ }
+
+ if (!Objects.equals(uri.getScheme(), "arrow-flight") &&
+ !Objects.equals(uri.getScheme(), "arrow-flight-sql")) {
+ throw new SQLException("URL Scheme must be 'arrow-flight'. Expected format: " +
+ CONNECTION_STRING_EXPECTED);
+ }
+
+ if (uri.getHost() == null) {
+ throw new SQLException("URL must have a host. Expected format: " + CONNECTION_STRING_EXPECTED);
+ } else if (uri.getPort() < 0) {
+ throw new SQLException("URL must have a port. Expected format: " + CONNECTION_STRING_EXPECTED);
+ }
+ resultMap.put(ArrowFlightConnectionProperty.HOST.camelName(), uri.getHost()); // host
+ resultMap.put(ArrowFlightConnectionProperty.PORT.camelName(), uri.getPort()); // port
+
+ final String extraParams = uri.getRawQuery(); // optional params
+ if (extraParams != null) {
+ final Map keyValuePairs = UrlParser.parse(extraParams, "&");
+ resultMap.putAll(keyValuePairs);
+ }
+
+ return resultMap;
+ }
+
+ static Properties lowerCasePropertyKeys(final Properties properties) {
+ final Properties resultProperty = new Properties();
+ properties.forEach((k, v) -> resultProperty.put(k.toString().toLowerCase(), v));
+ return resultProperty;
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFactory.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFactory.java
new file mode 100644
index 0000000000000..a54fbb9511b55
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFactory.java
@@ -0,0 +1,124 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.Properties;
+import java.util.TimeZone;
+
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.calcite.avatica.AvaticaConnection;
+import org.apache.calcite.avatica.AvaticaFactory;
+import org.apache.calcite.avatica.AvaticaResultSetMetaData;
+import org.apache.calcite.avatica.AvaticaSpecificDatabaseMetaData;
+import org.apache.calcite.avatica.AvaticaStatement;
+import org.apache.calcite.avatica.Meta;
+import org.apache.calcite.avatica.QueryState;
+import org.apache.calcite.avatica.UnregisteredDriver;
+
+/**
+ * Factory for the Arrow Flight JDBC Driver.
+ */
+public class ArrowFlightJdbcFactory implements AvaticaFactory {
+ private final int major;
+ private final int minor;
+
+ // This need to be public so Avatica can call this constructor
+ public ArrowFlightJdbcFactory() {
+ this(4, 1);
+ }
+
+ private ArrowFlightJdbcFactory(final int major, final int minor) {
+ this.major = major;
+ this.minor = minor;
+ }
+
+ @Override
+ public AvaticaConnection newConnection(final UnregisteredDriver driver,
+ final AvaticaFactory factory,
+ final String url,
+ final Properties info) throws SQLException {
+ return ArrowFlightConnection.createNewConnection(
+ (ArrowFlightJdbcDriver) driver,
+ factory,
+ url,
+ info,
+ new RootAllocator(Long.MAX_VALUE));
+ }
+
+ @Override
+ public AvaticaStatement newStatement(
+ final AvaticaConnection connection,
+ final Meta.StatementHandle handle,
+ final int resultType,
+ final int resultSetConcurrency,
+ final int resultSetHoldability) {
+ return new ArrowFlightStatement((ArrowFlightConnection) connection,
+ handle, resultType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public ArrowFlightPreparedStatement newPreparedStatement(
+ final AvaticaConnection connection,
+ final Meta.StatementHandle statementHandle,
+ final Meta.Signature signature,
+ final int resultType,
+ final int resultSetConcurrency,
+ final int resultSetHoldability) throws SQLException {
+ return ArrowFlightPreparedStatement.createNewPreparedStatement(
+ (ArrowFlightConnection) connection, statementHandle, signature,
+ resultType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public ArrowFlightJdbcVectorSchemaRootResultSet newResultSet(final AvaticaStatement statement,
+ final QueryState state,
+ final Meta.Signature signature,
+ final TimeZone timeZone,
+ final Meta.Frame frame)
+ throws SQLException {
+ final ResultSetMetaData metaData = newResultSetMetaData(statement, signature);
+
+ return new ArrowFlightJdbcFlightStreamResultSet(statement, state, signature, metaData, timeZone,
+ frame);
+ }
+
+ @Override
+ public AvaticaSpecificDatabaseMetaData newDatabaseMetaData(final AvaticaConnection connection) {
+ return new ArrowDatabaseMetadata(connection);
+ }
+
+ @Override
+ public ResultSetMetaData newResultSetMetaData(
+ final AvaticaStatement avaticaStatement,
+ final Meta.Signature signature) {
+ return new AvaticaResultSetMetaData(avaticaStatement,
+ null, signature);
+ }
+
+ @Override
+ public int getJdbcMajorVersion() {
+ return major;
+ }
+
+ @Override
+ public int getJdbcMinorVersion() {
+ return minor;
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java
new file mode 100644
index 0000000000000..4c01cb6e5813c
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java
@@ -0,0 +1,250 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import static org.apache.arrow.driver.jdbc.utils.FlightStreamQueue.createNewQueue;
+
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.Optional;
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.arrow.driver.jdbc.utils.FlightStreamQueue;
+import org.apache.arrow.driver.jdbc.utils.VectorSchemaRootTransformer;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.calcite.avatica.AvaticaResultSet;
+import org.apache.calcite.avatica.AvaticaResultSetMetaData;
+import org.apache.calcite.avatica.AvaticaStatement;
+import org.apache.calcite.avatica.Meta;
+import org.apache.calcite.avatica.QueryState;
+
+/**
+ * {@link ResultSet} implementation for Arrow Flight used to access the results of multiple {@link FlightStream}
+ * objects.
+ */
+public final class ArrowFlightJdbcFlightStreamResultSet
+ extends ArrowFlightJdbcVectorSchemaRootResultSet {
+
+ private final ArrowFlightConnection connection;
+ private FlightStream currentFlightStream;
+ private FlightStreamQueue flightStreamQueue;
+
+ private VectorSchemaRootTransformer transformer;
+ private VectorSchemaRoot currentVectorSchemaRoot;
+
+ private Schema schema;
+
+ ArrowFlightJdbcFlightStreamResultSet(final AvaticaStatement statement,
+ final QueryState state,
+ final Meta.Signature signature,
+ final ResultSetMetaData resultSetMetaData,
+ final TimeZone timeZone,
+ final Meta.Frame firstFrame) throws SQLException {
+ super(statement, state, signature, resultSetMetaData, timeZone, firstFrame);
+ this.connection = (ArrowFlightConnection) statement.connection;
+ }
+
+ ArrowFlightJdbcFlightStreamResultSet(final ArrowFlightConnection connection,
+ final QueryState state,
+ final Meta.Signature signature,
+ final ResultSetMetaData resultSetMetaData,
+ final TimeZone timeZone,
+ final Meta.Frame firstFrame) throws SQLException {
+ super(null, state, signature, resultSetMetaData, timeZone, firstFrame);
+ this.connection = connection;
+ }
+
+ /**
+ * Create a {@link ResultSet} which pulls data from given {@link FlightInfo}.
+ *
+ * @param connection The connection linked to the returned ResultSet.
+ * @param flightInfo The FlightInfo from which data will be iterated by the returned ResultSet.
+ * @param transformer Optional transformer for processing VectorSchemaRoot before access from ResultSet
+ * @return A ResultSet which pulls data from given FlightInfo.
+ */
+ static ArrowFlightJdbcFlightStreamResultSet fromFlightInfo(
+ final ArrowFlightConnection connection,
+ final FlightInfo flightInfo,
+ final VectorSchemaRootTransformer transformer) throws SQLException {
+ // Similar to how org.apache.calcite.avatica.util.ArrayFactoryImpl does
+
+ final TimeZone timeZone = TimeZone.getDefault();
+ final QueryState state = new QueryState();
+
+ final Meta.Signature signature = ArrowFlightMetaImpl.newSignature(null);
+
+ final AvaticaResultSetMetaData resultSetMetaData =
+ new AvaticaResultSetMetaData(null, null, signature);
+ final ArrowFlightJdbcFlightStreamResultSet resultSet =
+ new ArrowFlightJdbcFlightStreamResultSet(connection, state, signature, resultSetMetaData,
+ timeZone, null);
+
+ resultSet.transformer = transformer;
+
+ resultSet.execute(flightInfo);
+ return resultSet;
+ }
+
+ private void loadNewQueue() {
+ Optional.ofNullable(flightStreamQueue).ifPresent(AutoCloseables::closeNoChecked);
+ flightStreamQueue = createNewQueue(connection.getExecutorService());
+ }
+
+ private void loadNewFlightStream() throws SQLException {
+ if (currentFlightStream != null) {
+ AutoCloseables.closeNoChecked(currentFlightStream);
+ }
+ this.currentFlightStream = getNextFlightStream(true);
+ }
+
+ @Override
+ protected AvaticaResultSet execute() throws SQLException {
+ final FlightInfo flightInfo = ((ArrowFlightInfoStatement) statement).executeFlightInfoQuery();
+
+ if (flightInfo != null) {
+ schema = flightInfo.getSchema();
+ execute(flightInfo);
+ }
+ return this;
+ }
+
+ private void execute(final FlightInfo flightInfo) throws SQLException {
+ loadNewQueue();
+ flightStreamQueue.enqueue(connection.getClientHandler().getStreams(flightInfo));
+ loadNewFlightStream();
+
+ // Ownership of the root will be passed onto the cursor.
+ if (currentFlightStream != null) {
+ executeForCurrentFlightStream();
+ }
+ }
+
+ private void executeForCurrentFlightStream() throws SQLException {
+ final VectorSchemaRoot originalRoot = currentFlightStream.getRoot();
+
+ if (transformer != null) {
+ try {
+ currentVectorSchemaRoot = transformer.transform(originalRoot, currentVectorSchemaRoot);
+ } catch (final Exception e) {
+ throw new SQLException("Failed to transform VectorSchemaRoot.", e);
+ }
+ } else {
+ currentVectorSchemaRoot = originalRoot;
+ }
+
+ if (schema != null) {
+ execute(currentVectorSchemaRoot, schema);
+ } else {
+ execute(currentVectorSchemaRoot);
+ }
+ }
+
+ @Override
+ public boolean next() throws SQLException {
+ if (currentVectorSchemaRoot == null) {
+ return false;
+ }
+ while (true) {
+ final boolean hasNext = super.next();
+ final int maxRows = statement != null ? statement.getMaxRows() : 0;
+ if (maxRows != 0 && this.getRow() > maxRows) {
+ if (statement.isCloseOnCompletion()) {
+ statement.close();
+ }
+ return false;
+ }
+
+ if (hasNext) {
+ return true;
+ }
+
+ if (currentFlightStream != null) {
+ currentFlightStream.getRoot().clear();
+ if (currentFlightStream.next()) {
+ executeForCurrentFlightStream();
+ continue;
+ }
+
+ flightStreamQueue.enqueue(currentFlightStream);
+ }
+
+ currentFlightStream = getNextFlightStream(false);
+
+ if (currentFlightStream != null) {
+ executeForCurrentFlightStream();
+ continue;
+ }
+
+ if (statement != null && statement.isCloseOnCompletion()) {
+ statement.close();
+ }
+
+ return false;
+ }
+ }
+
+ @Override
+ protected void cancel() {
+ super.cancel();
+ final FlightStream currentFlightStream = this.currentFlightStream;
+ if (currentFlightStream != null) {
+ currentFlightStream.cancel("Cancel", null);
+ }
+
+ if (flightStreamQueue != null) {
+ try {
+ flightStreamQueue.close();
+ } catch (final Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public synchronized void close() {
+ try {
+ if (flightStreamQueue != null) {
+ // flightStreamQueue should close currentFlightStream internally
+ flightStreamQueue.close();
+ } else if (currentFlightStream != null) {
+ // close is only called for currentFlightStream if there's no queue
+ currentFlightStream.close();
+ }
+ } catch (final Exception e) {
+ throw new RuntimeException(e);
+ } finally {
+ super.close();
+ }
+ }
+
+ private FlightStream getNextFlightStream(final boolean isExecution) throws SQLException {
+ if (isExecution) {
+ final int statementTimeout = statement != null ? statement.getQueryTimeout() : 0;
+ return statementTimeout != 0 ?
+ flightStreamQueue.next(statementTimeout, TimeUnit.SECONDS) : flightStreamQueue.next();
+ } else {
+ return flightStreamQueue.next();
+ }
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcPooledConnection.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcPooledConnection.java
new file mode 100644
index 0000000000000..96a2d9dda1d18
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcPooledConnection.java
@@ -0,0 +1,112 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+
+import javax.sql.ConnectionEvent;
+import javax.sql.ConnectionEventListener;
+import javax.sql.PooledConnection;
+import javax.sql.StatementEventListener;
+
+import org.apache.arrow.driver.jdbc.utils.ConnectionWrapper;
+
+/**
+ * {@link PooledConnection} implementation for Arrow Flight JDBC Driver.
+ */
+public class ArrowFlightJdbcPooledConnection implements PooledConnection {
+
+ private final ArrowFlightConnection connection;
+ private final Set eventListeners;
+ private final Set statementEventListeners;
+
+ private final class ConnectionHandle extends ConnectionWrapper {
+ private boolean closed = false;
+
+ public ConnectionHandle() {
+ super(connection);
+ }
+
+ @Override
+ public void close() throws SQLException {
+ if (!closed) {
+ closed = true;
+ onConnectionClosed();
+ }
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return this.closed || super.isClosed();
+ }
+ }
+
+ ArrowFlightJdbcPooledConnection(ArrowFlightConnection connection) {
+ this.connection = connection;
+ this.eventListeners = Collections.synchronizedSet(new HashSet<>());
+ this.statementEventListeners = Collections.synchronizedSet(new HashSet<>());
+ }
+
+ public Properties getProperties() {
+ return connection.getClientInfo();
+ }
+
+ @Override
+ public Connection getConnection() throws SQLException {
+ return new ConnectionHandle();
+ }
+
+ @Override
+ public void close() throws SQLException {
+ this.connection.close();
+ }
+
+ void reset() throws SQLException {
+ this.connection.reset();
+ }
+
+ @Override
+ public void addConnectionEventListener(ConnectionEventListener listener) {
+ eventListeners.add(listener);
+ }
+
+ @Override
+ public void removeConnectionEventListener(ConnectionEventListener listener) {
+ this.eventListeners.remove(listener);
+ }
+
+ @Override
+ public void addStatementEventListener(StatementEventListener listener) {
+ statementEventListeners.add(listener);
+ }
+
+ @Override
+ public void removeStatementEventListener(StatementEventListener listener) {
+ this.statementEventListeners.remove(listener);
+ }
+
+ private void onConnectionClosed() {
+ ConnectionEvent connectionEvent = new ConnectionEvent(this);
+ eventListeners.forEach(listener -> listener.connectionClosed(connectionEvent));
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcTime.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcTime.java
new file mode 100644
index 0000000000000..109048bc05c91
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcTime.java
@@ -0,0 +1,106 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import static org.apache.calcite.avatica.util.DateTimeUtils.MILLIS_PER_DAY;
+
+import java.sql.Time;
+import java.time.LocalTime;
+import java.time.temporal.ChronoField;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.arrow.util.VisibleForTesting;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Wrapper class for Time objects to include the milliseconds part in ISO 8601 format in this#toString.
+ */
+public class ArrowFlightJdbcTime extends Time {
+ private static final List LEADING_ZEROES = ImmutableList.of("", "0", "00");
+
+ // Desired length of the millisecond portion should be 3
+ private static final int DESIRED_MILLIS_LENGTH = 3;
+
+ // Millis of the date time object.
+ private final int millisReprValue;
+
+ /**
+ * Constructs this object based on epoch millis.
+ *
+ * @param milliseconds milliseconds representing Time.
+ */
+ public ArrowFlightJdbcTime(final long milliseconds) {
+ super(milliseconds);
+ millisReprValue = getMillisReprValue(milliseconds);
+ }
+
+ @VisibleForTesting
+ ArrowFlightJdbcTime(final LocalTime time) {
+ // Although the constructor is deprecated, this is the exact same code as Time#valueOf(LocalTime)
+ super(time.getHour(), time.getMinute(), time.getSecond());
+ millisReprValue = time.get(ChronoField.MILLI_OF_SECOND);
+ }
+
+ private int getMillisReprValue(long milliseconds) {
+ // Extract the millisecond part from epoch nano day
+ if (milliseconds >= MILLIS_PER_DAY) {
+ // Convert to Epoch Day
+ milliseconds %= MILLIS_PER_DAY;
+ } else if (milliseconds < 0) {
+ // LocalTime#ofNanoDay only accepts positive values
+ milliseconds -= ((milliseconds / MILLIS_PER_DAY) - 1) * MILLIS_PER_DAY;
+ }
+ return LocalTime.ofNanoOfDay(TimeUnit.MILLISECONDS.toNanos(milliseconds))
+ .get(ChronoField.MILLI_OF_SECOND);
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder time = new StringBuilder().append(super.toString());
+
+ if (millisReprValue > 0) {
+ final String millisString = Integer.toString(millisReprValue);
+
+ // dot to separate the fractional seconds
+ time.append(".");
+
+ final int millisLength = millisString.length();
+ if (millisLength < DESIRED_MILLIS_LENGTH) {
+ // add necessary leading zeroes
+ time.append(LEADING_ZEROES.get(DESIRED_MILLIS_LENGTH - millisLength));
+ }
+ time.append(millisString);
+ }
+
+ return time.toString();
+ }
+
+ // Spotbugs requires these methods to be overridden
+ @Override
+ public boolean equals(Object obj) {
+ return super.equals(obj);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), this.millisReprValue);
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcVectorSchemaRootResultSet.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcVectorSchemaRootResultSet.java
new file mode 100644
index 0000000000000..9e377e51decc9
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcVectorSchemaRootResultSet.java
@@ -0,0 +1,153 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import static java.util.Objects.isNull;
+
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TimeZone;
+
+import org.apache.arrow.driver.jdbc.utils.ConvertUtils;
+import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.calcite.avatica.AvaticaResultSet;
+import org.apache.calcite.avatica.AvaticaResultSetMetaData;
+import org.apache.calcite.avatica.AvaticaStatement;
+import org.apache.calcite.avatica.ColumnMetaData;
+import org.apache.calcite.avatica.Meta;
+import org.apache.calcite.avatica.Meta.Frame;
+import org.apache.calcite.avatica.Meta.Signature;
+import org.apache.calcite.avatica.QueryState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link ResultSet} implementation used to access a {@link VectorSchemaRoot}.
+ */
+public class ArrowFlightJdbcVectorSchemaRootResultSet extends AvaticaResultSet {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ArrowFlightJdbcVectorSchemaRootResultSet.class);
+ VectorSchemaRoot vectorSchemaRoot;
+
+ ArrowFlightJdbcVectorSchemaRootResultSet(final AvaticaStatement statement, final QueryState state,
+ final Signature signature,
+ final ResultSetMetaData resultSetMetaData,
+ final TimeZone timeZone, final Frame firstFrame)
+ throws SQLException {
+ super(statement, state, signature, resultSetMetaData, timeZone, firstFrame);
+ }
+
+ /**
+ * Instantiate a ResultSet backed up by given VectorSchemaRoot.
+ *
+ * @param vectorSchemaRoot root from which the ResultSet will access.
+ * @return a ResultSet which accesses the given VectorSchemaRoot
+ */
+ public static ArrowFlightJdbcVectorSchemaRootResultSet fromVectorSchemaRoot(
+ final VectorSchemaRoot vectorSchemaRoot)
+ throws SQLException {
+ // Similar to how org.apache.calcite.avatica.util.ArrayFactoryImpl does
+
+ final TimeZone timeZone = TimeZone.getDefault();
+ final QueryState state = new QueryState();
+
+ final Meta.Signature signature = ArrowFlightMetaImpl.newSignature(null);
+
+ final AvaticaResultSetMetaData resultSetMetaData =
+ new AvaticaResultSetMetaData(null, null, signature);
+ final ArrowFlightJdbcVectorSchemaRootResultSet
+ resultSet =
+ new ArrowFlightJdbcVectorSchemaRootResultSet(null, state, signature, resultSetMetaData,
+ timeZone, null);
+
+ resultSet.execute(vectorSchemaRoot);
+ return resultSet;
+ }
+
+ @Override
+ protected AvaticaResultSet execute() throws SQLException {
+ throw new RuntimeException("Can only execute with execute(VectorSchemaRoot)");
+ }
+
+ void execute(final VectorSchemaRoot vectorSchemaRoot) {
+ final List fields = vectorSchemaRoot.getSchema().getFields();
+ final List columns = ConvertUtils.convertArrowFieldsToColumnMetaDataList(fields);
+ signature.columns.clear();
+ signature.columns.addAll(columns);
+
+ this.vectorSchemaRoot = vectorSchemaRoot;
+ execute2(new ArrowFlightJdbcCursor(vectorSchemaRoot), this.signature.columns);
+ }
+
+ void execute(final VectorSchemaRoot vectorSchemaRoot, final Schema schema) {
+ final List columns = ConvertUtils.convertArrowFieldsToColumnMetaDataList(schema.getFields());
+ signature.columns.clear();
+ signature.columns.addAll(columns);
+
+ this.vectorSchemaRoot = vectorSchemaRoot;
+ execute2(new ArrowFlightJdbcCursor(vectorSchemaRoot), this.signature.columns);
+ }
+
+ @Override
+ protected void cancel() {
+ signature.columns.clear();
+ super.cancel();
+ try {
+ AutoCloseables.close(vectorSchemaRoot);
+ } catch (final Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void close() {
+ final Set exceptions = new HashSet<>();
+ try {
+ if (isClosed()) {
+ return;
+ }
+ } catch (final SQLException e) {
+ exceptions.add(e);
+ }
+ try {
+ AutoCloseables.close(vectorSchemaRoot);
+ } catch (final Exception e) {
+ exceptions.add(e);
+ }
+ if (!isNull(statement)) {
+ try {
+ super.close();
+ } catch (final Exception e) {
+ exceptions.add(e);
+ }
+ }
+ exceptions.parallelStream().forEach(e -> LOGGER.error(e.getMessage(), e));
+ exceptions.stream().findAny().ifPresent(e -> {
+ throw new RuntimeException(e);
+ });
+ }
+
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java
new file mode 100644
index 0000000000000..cc7addc3a74d1
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java
@@ -0,0 +1,200 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import static java.lang.String.format;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.SQLTimeoutException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.arrow.driver.jdbc.client.ArrowFlightSqlClientHandler.PreparedStatement;
+import org.apache.arrow.util.Preconditions;
+import org.apache.calcite.avatica.AvaticaConnection;
+import org.apache.calcite.avatica.AvaticaParameter;
+import org.apache.calcite.avatica.ColumnMetaData;
+import org.apache.calcite.avatica.MetaImpl;
+import org.apache.calcite.avatica.NoSuchStatementException;
+import org.apache.calcite.avatica.QueryState;
+import org.apache.calcite.avatica.remote.TypedValue;
+
+/**
+ * Metadata handler for Arrow Flight.
+ */
+public class ArrowFlightMetaImpl extends MetaImpl {
+ private final Map statementHandlePreparedStatementMap;
+
+ /**
+ * Constructs a {@link MetaImpl} object specific for Arrow Flight.
+ * @param connection A {@link AvaticaConnection}.
+ */
+ public ArrowFlightMetaImpl(final AvaticaConnection connection) {
+ super(connection);
+ this.statementHandlePreparedStatementMap = new ConcurrentHashMap<>();
+ setDefaultConnectionProperties();
+ }
+
+ static Signature newSignature(final String sql) {
+ return new Signature(
+ new ArrayList(),
+ sql,
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ null, // unnecessary, as SQL requests use ArrowFlightJdbcCursor
+ StatementType.SELECT
+ );
+ }
+
+ @Override
+ public void closeStatement(final StatementHandle statementHandle) {
+ PreparedStatement preparedStatement = statementHandlePreparedStatementMap.remove(statementHandle);
+ // Testing if the prepared statement was created because the statement can be not created until this moment
+ if (preparedStatement != null) {
+ preparedStatement.close();
+ }
+ }
+
+ @Override
+ public void commit(final ConnectionHandle connectionHandle) {
+ // TODO Fill this stub.
+ }
+
+ @Override
+ public ExecuteResult execute(final StatementHandle statementHandle,
+ final List typedValues, final long maxRowCount) {
+ // TODO Why is maxRowCount ignored?
+ Preconditions.checkNotNull(statementHandle.signature, "Signature not found.");
+ return new ExecuteResult(
+ Collections.singletonList(MetaResultSet.create(
+ statementHandle.connectionId, statementHandle.id,
+ true, statementHandle.signature, null)));
+ }
+
+ @Override
+ public ExecuteResult execute(final StatementHandle statementHandle,
+ final List typedValues, final int maxRowsInFirstFrame) {
+ return execute(statementHandle, typedValues, (long) maxRowsInFirstFrame);
+ }
+
+ @Override
+ public ExecuteBatchResult executeBatch(final StatementHandle statementHandle,
+ final List> parameterValuesList)
+ throws IllegalStateException {
+ throw new IllegalStateException("executeBatch not implemented.");
+ }
+
+ @Override
+ public Frame fetch(final StatementHandle statementHandle, final long offset,
+ final int fetchMaxRowCount) {
+ /*
+ * ArrowFlightMetaImpl does not use frames.
+ * Instead, we have accessors that contain a VectorSchemaRoot with
+ * the results.
+ */
+ throw AvaticaConnection.HELPER.wrap(
+ format("%s does not use frames.", this),
+ AvaticaConnection.HELPER.unsupported());
+ }
+
+ @Override
+ public StatementHandle prepare(final ConnectionHandle connectionHandle,
+ final String query, final long maxRowCount) {
+ final StatementHandle handle = super.createStatement(connectionHandle);
+ handle.signature = newSignature(query);
+ return handle;
+ }
+
+ @Override
+ public ExecuteResult prepareAndExecute(final StatementHandle statementHandle,
+ final String query, final long maxRowCount,
+ final PrepareCallback prepareCallback)
+ throws NoSuchStatementException {
+ return prepareAndExecute(
+ statementHandle, query, maxRowCount, -1 /* Not used */, prepareCallback);
+ }
+
+ @Override
+ public ExecuteResult prepareAndExecute(final StatementHandle handle,
+ final String query, final long maxRowCount,
+ final int maxRowsInFirstFrame,
+ final PrepareCallback callback)
+ throws NoSuchStatementException {
+ try {
+ final PreparedStatement preparedStatement =
+ ((ArrowFlightConnection) connection).getClientHandler().prepare(query);
+ final StatementType statementType = preparedStatement.getType();
+ statementHandlePreparedStatementMap.put(handle, preparedStatement);
+ final Signature signature = newSignature(query);
+ final long updateCount =
+ statementType.equals(StatementType.UPDATE) ? preparedStatement.executeUpdate() : -1;
+ synchronized (callback.getMonitor()) {
+ callback.clear();
+ callback.assign(signature, null, updateCount);
+ }
+ callback.execute();
+ final MetaResultSet metaResultSet = MetaResultSet.create(handle.connectionId, handle.id,
+ false, signature, null);
+ return new ExecuteResult(Collections.singletonList(metaResultSet));
+ } catch (SQLTimeoutException e) {
+ // So far AvaticaStatement(executeInternal) only handles NoSuchStatement and Runtime Exceptions.
+ throw new RuntimeException(e);
+ } catch (SQLException e) {
+ throw new NoSuchStatementException(handle);
+ }
+ }
+
+ @Override
+ public ExecuteBatchResult prepareAndExecuteBatch(
+ final StatementHandle statementHandle, final List queries)
+ throws NoSuchStatementException {
+ // TODO Fill this stub.
+ return null;
+ }
+
+ @Override
+ public void rollback(final ConnectionHandle connectionHandle) {
+ // TODO Fill this stub.
+ }
+
+ @Override
+ public boolean syncResults(final StatementHandle statementHandle,
+ final QueryState queryState, final long offset)
+ throws NoSuchStatementException {
+ // TODO Fill this stub.
+ return false;
+ }
+
+ void setDefaultConnectionProperties() {
+ // TODO Double-check this.
+ connProps.setDirty(false)
+ .setAutoCommit(true)
+ .setReadOnly(true)
+ .setCatalog(null)
+ .setSchema(null)
+ .setTransactionIsolation(Connection.TRANSACTION_NONE);
+ }
+
+ PreparedStatement getPreparedStatement(StatementHandle statementHandle) {
+ return statementHandlePreparedStatementMap.get(statementHandle);
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java
new file mode 100644
index 0000000000000..80029f38f0958
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java
@@ -0,0 +1,98 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+import org.apache.arrow.driver.jdbc.client.ArrowFlightSqlClientHandler;
+import org.apache.arrow.driver.jdbc.utils.ConvertUtils;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.calcite.avatica.AvaticaPreparedStatement;
+import org.apache.calcite.avatica.Meta.Signature;
+import org.apache.calcite.avatica.Meta.StatementHandle;
+
+
+/**
+ * Arrow Flight JBCS's implementation {@link PreparedStatement}.
+ */
+public class ArrowFlightPreparedStatement extends AvaticaPreparedStatement
+ implements ArrowFlightInfoStatement {
+
+ private final ArrowFlightSqlClientHandler.PreparedStatement preparedStatement;
+
+ private ArrowFlightPreparedStatement(final ArrowFlightConnection connection,
+ final ArrowFlightSqlClientHandler.PreparedStatement preparedStatement,
+ final StatementHandle handle,
+ final Signature signature, final int resultSetType,
+ final int resultSetConcurrency,
+ final int resultSetHoldability)
+ throws SQLException {
+ super(connection, handle, signature, resultSetType, resultSetConcurrency, resultSetHoldability);
+ this.preparedStatement = Preconditions.checkNotNull(preparedStatement);
+ }
+
+ /**
+ * Creates a new {@link ArrowFlightPreparedStatement} from the provided information.
+ *
+ * @param connection the {@link Connection} to use.
+ * @param statementHandle the {@link StatementHandle} to use.
+ * @param signature the {@link Signature} to use.
+ * @param resultSetType the ResultSet type.
+ * @param resultSetConcurrency the ResultSet concurrency.
+ * @param resultSetHoldability the ResultSet holdability.
+ * @return a new {@link PreparedStatement}.
+ * @throws SQLException on error.
+ */
+ static ArrowFlightPreparedStatement createNewPreparedStatement(
+ final ArrowFlightConnection connection,
+ final StatementHandle statementHandle,
+ final Signature signature,
+ final int resultSetType,
+ final int resultSetConcurrency,
+ final int resultSetHoldability) throws SQLException {
+
+ final ArrowFlightSqlClientHandler.PreparedStatement prepare = connection.getClientHandler().prepare(signature.sql);
+ final Schema resultSetSchema = prepare.getDataSetSchema();
+
+ signature.columns.addAll(ConvertUtils.convertArrowFieldsToColumnMetaDataList(resultSetSchema.getFields()));
+
+ return new ArrowFlightPreparedStatement(
+ connection, prepare, statementHandle,
+ signature, resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public ArrowFlightConnection getConnection() throws SQLException {
+ return (ArrowFlightConnection) super.getConnection();
+ }
+
+ @Override
+ public synchronized void close() throws SQLException {
+ this.preparedStatement.close();
+ super.close();
+ }
+
+ @Override
+ public FlightInfo executeFlightInfoQuery() throws SQLException {
+ return preparedStatement.executeQuery();
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java
new file mode 100644
index 0000000000000..5bc7c2ab9b4f8
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java
@@ -0,0 +1,60 @@
+/*
+ * 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.arrow.driver.jdbc;
+
+import java.sql.SQLException;
+
+import org.apache.arrow.driver.jdbc.client.ArrowFlightSqlClientHandler.PreparedStatement;
+import org.apache.arrow.driver.jdbc.utils.ConvertUtils;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.calcite.avatica.AvaticaStatement;
+import org.apache.calcite.avatica.Meta;
+import org.apache.calcite.avatica.Meta.StatementHandle;
+
+/**
+ * A SQL statement for querying data from an Arrow Flight server.
+ */
+public class ArrowFlightStatement extends AvaticaStatement implements ArrowFlightInfoStatement {
+
+ ArrowFlightStatement(final ArrowFlightConnection connection,
+ final StatementHandle handle, final int resultSetType,
+ final int resultSetConcurrency, final int resultSetHoldability) {
+ super(connection, handle, resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public ArrowFlightConnection getConnection() throws SQLException {
+ return (ArrowFlightConnection) super.getConnection();
+ }
+
+ @Override
+ public FlightInfo executeFlightInfoQuery() throws SQLException {
+ final PreparedStatement preparedStatement = getConnection().getMeta().getPreparedStatement(handle);
+ final Meta.Signature signature = getSignature();
+ if (signature == null) {
+ return null;
+ }
+
+ final Schema resultSetSchema = preparedStatement.getDataSetSchema();
+ signature.columns.addAll(ConvertUtils.convertArrowFieldsToColumnMetaDataList(resultSetSchema.getFields()));
+ setSignature(signature);
+
+ return preparedStatement.executeQuery();
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/ArrowFlightJdbcAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/ArrowFlightJdbcAccessor.java
new file mode 100644
index 0000000000000..3821ee1dc8755
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/ArrowFlightJdbcAccessor.java
@@ -0,0 +1,256 @@
+/*
+ * 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.arrow.driver.jdbc.accessor;
+
+import static org.apache.calcite.avatica.util.Cursor.Accessor;
+
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.net.URL;
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.Clob;
+import java.sql.Date;
+import java.sql.NClob;
+import java.sql.Ref;
+import java.sql.SQLException;
+import java.sql.SQLXML;
+import java.sql.Struct;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Calendar;
+import java.util.Map;
+import java.util.function.IntSupplier;
+
+/**
+ * Base Jdbc Accessor.
+ */
+public abstract class ArrowFlightJdbcAccessor implements Accessor {
+ private final IntSupplier currentRowSupplier;
+
+ // All the derived accessor classes should alter this as they encounter null Values
+ protected boolean wasNull;
+ protected ArrowFlightJdbcAccessorFactory.WasNullConsumer wasNullConsumer;
+
+ protected ArrowFlightJdbcAccessor(final IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer wasNullConsumer) {
+ this.currentRowSupplier = currentRowSupplier;
+ this.wasNullConsumer = wasNullConsumer;
+ }
+
+ protected int getCurrentRow() {
+ return currentRowSupplier.getAsInt();
+ }
+
+ // It needs to be public so this method can be accessed when creating the complex types.
+ public abstract Class getObjectClass();
+
+ @Override
+ public boolean wasNull() {
+ return wasNull;
+ }
+
+ @Override
+ public String getString() throws SQLException {
+ final Object object = getObject();
+ if (object == null) {
+ return null;
+ }
+
+ return object.toString();
+ }
+
+ @Override
+ public boolean getBoolean() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public byte getByte() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public short getShort() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public int getInt() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public long getLong() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public float getFloat() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public double getDouble() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public BigDecimal getBigDecimal() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(final int i) throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public byte[] getBytes() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public InputStream getAsciiStream() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public InputStream getUnicodeStream() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public InputStream getBinaryStream() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Object getObject() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Reader getCharacterStream() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Object getObject(final Map> map) throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Ref getRef() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Blob getBlob() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Clob getClob() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Array getArray() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Struct getStruct() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Date getDate(final Calendar calendar) throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Time getTime(final Calendar calendar) throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Timestamp getTimestamp(final Calendar calendar) throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public URL getURL() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public NClob getNClob() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public SQLXML getSQLXML() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public String getNString() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public Reader getNCharacterStream() throws SQLException {
+ throw getOperationNotSupported(this.getClass());
+ }
+
+ @Override
+ public T getObject(final Class type) throws SQLException {
+ final Object value;
+ if (type == Byte.class) {
+ value = getByte();
+ } else if (type == Short.class) {
+ value = getShort();
+ } else if (type == Integer.class) {
+ value = getInt();
+ } else if (type == Long.class) {
+ value = getLong();
+ } else if (type == Float.class) {
+ value = getFloat();
+ } else if (type == Double.class) {
+ value = getDouble();
+ } else if (type == Boolean.class) {
+ value = getBoolean();
+ } else if (type == BigDecimal.class) {
+ value = getBigDecimal();
+ } else if (type == String.class) {
+ value = getString();
+ } else if (type == byte[].class) {
+ value = getBytes();
+ } else {
+ value = getObject();
+ }
+ return !type.isPrimitive() && wasNull ? null : type.cast(value);
+ }
+
+ private static SQLException getOperationNotSupported(final Class type) {
+ return new SQLException(String.format("Operation not supported for type: %s.", type.getName()));
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/ArrowFlightJdbcAccessorFactory.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/ArrowFlightJdbcAccessorFactory.java
new file mode 100644
index 0000000000000..813b40a8070f7
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/ArrowFlightJdbcAccessorFactory.java
@@ -0,0 +1,214 @@
+/*
+ * 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.arrow.driver.jdbc.accessor;
+
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.impl.ArrowFlightJdbcNullVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.binary.ArrowFlightJdbcBinaryVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcDateVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcDurationVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcIntervalVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeStampVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.complex.ArrowFlightJdbcDenseUnionVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.complex.ArrowFlightJdbcFixedSizeListVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.complex.ArrowFlightJdbcLargeListVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.complex.ArrowFlightJdbcListVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.complex.ArrowFlightJdbcMapVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.complex.ArrowFlightJdbcStructVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.complex.ArrowFlightJdbcUnionVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.numeric.ArrowFlightJdbcBaseIntVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.numeric.ArrowFlightJdbcBitVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.numeric.ArrowFlightJdbcDecimalVectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.numeric.ArrowFlightJdbcFloat4VectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.numeric.ArrowFlightJdbcFloat8VectorAccessor;
+import org.apache.arrow.driver.jdbc.accessor.impl.text.ArrowFlightJdbcVarCharVectorAccessor;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DateMilliVector;
+import org.apache.arrow.vector.Decimal256Vector;
+import org.apache.arrow.vector.DecimalVector;
+import org.apache.arrow.vector.DurationVector;
+import org.apache.arrow.vector.FixedSizeBinaryVector;
+import org.apache.arrow.vector.Float4Vector;
+import org.apache.arrow.vector.Float8Vector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.IntervalDayVector;
+import org.apache.arrow.vector.IntervalYearVector;
+import org.apache.arrow.vector.LargeVarBinaryVector;
+import org.apache.arrow.vector.LargeVarCharVector;
+import org.apache.arrow.vector.NullVector;
+import org.apache.arrow.vector.SmallIntVector;
+import org.apache.arrow.vector.TimeMicroVector;
+import org.apache.arrow.vector.TimeMilliVector;
+import org.apache.arrow.vector.TimeNanoVector;
+import org.apache.arrow.vector.TimeSecVector;
+import org.apache.arrow.vector.TimeStampVector;
+import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.UInt1Vector;
+import org.apache.arrow.vector.UInt2Vector;
+import org.apache.arrow.vector.UInt4Vector;
+import org.apache.arrow.vector.UInt8Vector;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.complex.DenseUnionVector;
+import org.apache.arrow.vector.complex.FixedSizeListVector;
+import org.apache.arrow.vector.complex.LargeListVector;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.complex.StructVector;
+import org.apache.arrow.vector.complex.UnionVector;
+
+/**
+ * Factory to instantiate the accessors.
+ */
+public class ArrowFlightJdbcAccessorFactory {
+
+ /**
+ * Create an accessor according to its type.
+ *
+ * @param vector an instance of an arrow vector.
+ * @param getCurrentRow a supplier to check which row is being accessed.
+ * @return an instance of one of the accessors.
+ */
+ public static ArrowFlightJdbcAccessor createAccessor(ValueVector vector,
+ IntSupplier getCurrentRow,
+ WasNullConsumer setCursorWasNull) {
+ if (vector instanceof UInt1Vector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((UInt1Vector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof UInt2Vector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((UInt2Vector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof UInt4Vector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((UInt4Vector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof UInt8Vector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((UInt8Vector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof TinyIntVector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((TinyIntVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof SmallIntVector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((SmallIntVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof IntVector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((IntVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof BigIntVector) {
+ return new ArrowFlightJdbcBaseIntVectorAccessor((BigIntVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof Float4Vector) {
+ return new ArrowFlightJdbcFloat4VectorAccessor((Float4Vector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof Float8Vector) {
+ return new ArrowFlightJdbcFloat8VectorAccessor((Float8Vector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof BitVector) {
+ return new ArrowFlightJdbcBitVectorAccessor((BitVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof DecimalVector) {
+ return new ArrowFlightJdbcDecimalVectorAccessor((DecimalVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof Decimal256Vector) {
+ return new ArrowFlightJdbcDecimalVectorAccessor((Decimal256Vector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof VarBinaryVector) {
+ return new ArrowFlightJdbcBinaryVectorAccessor((VarBinaryVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof LargeVarBinaryVector) {
+ return new ArrowFlightJdbcBinaryVectorAccessor((LargeVarBinaryVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof FixedSizeBinaryVector) {
+ return new ArrowFlightJdbcBinaryVectorAccessor((FixedSizeBinaryVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof TimeStampVector) {
+ return new ArrowFlightJdbcTimeStampVectorAccessor((TimeStampVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof TimeNanoVector) {
+ return new ArrowFlightJdbcTimeVectorAccessor((TimeNanoVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof TimeMicroVector) {
+ return new ArrowFlightJdbcTimeVectorAccessor((TimeMicroVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof TimeMilliVector) {
+ return new ArrowFlightJdbcTimeVectorAccessor((TimeMilliVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof TimeSecVector) {
+ return new ArrowFlightJdbcTimeVectorAccessor((TimeSecVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof DateDayVector) {
+ return new ArrowFlightJdbcDateVectorAccessor(((DateDayVector) vector), getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof DateMilliVector) {
+ return new ArrowFlightJdbcDateVectorAccessor(((DateMilliVector) vector), getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof VarCharVector) {
+ return new ArrowFlightJdbcVarCharVectorAccessor((VarCharVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof LargeVarCharVector) {
+ return new ArrowFlightJdbcVarCharVectorAccessor((LargeVarCharVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof DurationVector) {
+ return new ArrowFlightJdbcDurationVectorAccessor((DurationVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof IntervalDayVector) {
+ return new ArrowFlightJdbcIntervalVectorAccessor(((IntervalDayVector) vector), getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof IntervalYearVector) {
+ return new ArrowFlightJdbcIntervalVectorAccessor(((IntervalYearVector) vector), getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof StructVector) {
+ return new ArrowFlightJdbcStructVectorAccessor((StructVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof MapVector) {
+ return new ArrowFlightJdbcMapVectorAccessor((MapVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof ListVector) {
+ return new ArrowFlightJdbcListVectorAccessor((ListVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof LargeListVector) {
+ return new ArrowFlightJdbcLargeListVectorAccessor((LargeListVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof FixedSizeListVector) {
+ return new ArrowFlightJdbcFixedSizeListVectorAccessor((FixedSizeListVector) vector,
+ getCurrentRow, setCursorWasNull);
+ } else if (vector instanceof UnionVector) {
+ return new ArrowFlightJdbcUnionVectorAccessor((UnionVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof DenseUnionVector) {
+ return new ArrowFlightJdbcDenseUnionVectorAccessor((DenseUnionVector) vector, getCurrentRow,
+ setCursorWasNull);
+ } else if (vector instanceof NullVector || vector == null) {
+ return new ArrowFlightJdbcNullVectorAccessor(setCursorWasNull);
+ }
+
+ throw new UnsupportedOperationException("Unsupported vector type: " + vector.getClass().getName());
+ }
+
+ /**
+ * Functional interface used to propagate that the value accessed was null or not.
+ */
+ @FunctionalInterface
+ public interface WasNullConsumer {
+ void setWasNull(boolean wasNull);
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/ArrowFlightJdbcNullVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/ArrowFlightJdbcNullVectorAccessor.java
new file mode 100644
index 0000000000000..f40a5797293f9
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/ArrowFlightJdbcNullVectorAccessor.java
@@ -0,0 +1,48 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.NullVector;
+
+/**
+ * Accessor for the Arrow type {@link NullVector}.
+ */
+public class ArrowFlightJdbcNullVectorAccessor extends ArrowFlightJdbcAccessor {
+ public ArrowFlightJdbcNullVectorAccessor(
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(null, setCursorWasNull);
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return Object.class;
+ }
+
+ @Override
+ public boolean wasNull() {
+ return true;
+ }
+
+ @Override
+ public Object getObject() {
+ this.wasNullConsumer.setWasNull(true);
+ return null;
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/binary/ArrowFlightJdbcBinaryVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/binary/ArrowFlightJdbcBinaryVectorAccessor.java
new file mode 100644
index 0000000000000..c50d734972134
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/binary/ArrowFlightJdbcBinaryVectorAccessor.java
@@ -0,0 +1,137 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.binary;
+
+import java.io.ByteArrayInputStream;
+import java.io.CharArrayReader;
+import java.io.InputStream;
+import java.io.Reader;
+import java.nio.charset.StandardCharsets;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.FixedSizeBinaryVector;
+import org.apache.arrow.vector.LargeVarBinaryVector;
+import org.apache.arrow.vector.VarBinaryVector;
+
+/**
+ * Accessor for the Arrow types: {@link FixedSizeBinaryVector}, {@link VarBinaryVector}
+ * and {@link LargeVarBinaryVector}.
+ */
+public class ArrowFlightJdbcBinaryVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ private interface ByteArrayGetter {
+ byte[] get(int index);
+ }
+
+ private final ByteArrayGetter getter;
+
+ public ArrowFlightJdbcBinaryVectorAccessor(FixedSizeBinaryVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ this(vector::get, currentRowSupplier, setCursorWasNull);
+ }
+
+ public ArrowFlightJdbcBinaryVectorAccessor(VarBinaryVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ this(vector::get, currentRowSupplier, setCursorWasNull);
+ }
+
+ public ArrowFlightJdbcBinaryVectorAccessor(LargeVarBinaryVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ this(vector::get, currentRowSupplier, setCursorWasNull);
+ }
+
+ private ArrowFlightJdbcBinaryVectorAccessor(ByteArrayGetter getter,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.getter = getter;
+ }
+
+ @Override
+ public byte[] getBytes() {
+ byte[] bytes = getter.get(getCurrentRow());
+ this.wasNull = bytes == null;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+
+ return bytes;
+ }
+
+ @Override
+ public Object getObject() {
+ return this.getBytes();
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return byte[].class;
+ }
+
+ @Override
+ public String getString() {
+ byte[] bytes = this.getBytes();
+ if (bytes == null) {
+ return null;
+ }
+
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public InputStream getAsciiStream() {
+ byte[] bytes = getBytes();
+ if (bytes == null) {
+ return null;
+ }
+
+ return new ByteArrayInputStream(bytes);
+ }
+
+ @Override
+ public InputStream getUnicodeStream() {
+ byte[] bytes = getBytes();
+ if (bytes == null) {
+ return null;
+ }
+
+ return new ByteArrayInputStream(bytes);
+ }
+
+ @Override
+ public InputStream getBinaryStream() {
+ byte[] bytes = getBytes();
+ if (bytes == null) {
+ return null;
+ }
+
+ return new ByteArrayInputStream(bytes);
+ }
+
+ @Override
+ public Reader getCharacterStream() {
+ String string = getString();
+ if (string == null) {
+ return null;
+ }
+
+ return new CharArrayReader(string.toCharArray());
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDateVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDateVectorAccessor.java
new file mode 100644
index 0000000000000..f6c14a47f521c
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDateVectorAccessor.java
@@ -0,0 +1,137 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.calendar;
+
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcDateVectorGetter.Getter;
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcDateVectorGetter.Holder;
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcDateVectorGetter.createGetter;
+import static org.apache.arrow.driver.jdbc.utils.DateTimeUtils.getTimestampValue;
+import static org.apache.calcite.avatica.util.DateTimeUtils.MILLIS_PER_DAY;
+import static org.apache.calcite.avatica.util.DateTimeUtils.unixDateToString;
+
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Calendar;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.driver.jdbc.utils.DateTimeUtils;
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DateMilliVector;
+import org.apache.arrow.vector.ValueVector;
+
+/**
+ * Accessor for the Arrow types: {@link DateDayVector} and {@link DateMilliVector}.
+ */
+public class ArrowFlightJdbcDateVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ private final Getter getter;
+ private final TimeUnit timeUnit;
+ private final Holder holder;
+
+ /**
+ * Instantiate an accessor for a {@link DateDayVector}.
+ *
+ * @param vector an instance of a DateDayVector.
+ * @param currentRowSupplier the supplier to track the lines.
+ * @param setCursorWasNull the consumer to set if value was null.
+ */
+ public ArrowFlightJdbcDateVectorAccessor(DateDayVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.holder = new Holder();
+ this.getter = createGetter(vector);
+ this.timeUnit = getTimeUnitForVector(vector);
+ }
+
+ /**
+ * Instantiate an accessor for a {@link DateMilliVector}.
+ *
+ * @param vector an instance of a DateMilliVector.
+ * @param currentRowSupplier the supplier to track the lines.
+ */
+ public ArrowFlightJdbcDateVectorAccessor(DateMilliVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.holder = new Holder();
+ this.getter = createGetter(vector);
+ this.timeUnit = getTimeUnitForVector(vector);
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return Date.class;
+ }
+
+ @Override
+ public Object getObject() {
+ return this.getDate(null);
+ }
+
+ @Override
+ public Date getDate(Calendar calendar) {
+ fillHolder();
+ if (this.wasNull) {
+ return null;
+ }
+
+ long value = holder.value;
+ long milliseconds = this.timeUnit.toMillis(value);
+
+ long millisWithCalendar = DateTimeUtils.applyCalendarOffset(milliseconds, calendar);
+
+ return new Date(getTimestampValue(millisWithCalendar).getTime());
+ }
+
+ private void fillHolder() {
+ getter.get(getCurrentRow(), holder);
+ this.wasNull = holder.isSet == 0;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+ }
+
+ @Override
+ public Timestamp getTimestamp(Calendar calendar) {
+ Date date = getDate(calendar);
+ if (date == null) {
+ return null;
+ }
+ return new Timestamp(date.getTime());
+ }
+
+ @Override
+ public String getString() {
+ fillHolder();
+ if (wasNull) {
+ return null;
+ }
+ long milliseconds = timeUnit.toMillis(holder.value);
+ return unixDateToString((int) (milliseconds / MILLIS_PER_DAY));
+ }
+
+ protected static TimeUnit getTimeUnitForVector(ValueVector vector) {
+ if (vector instanceof DateDayVector) {
+ return TimeUnit.DAYS;
+ } else if (vector instanceof DateMilliVector) {
+ return TimeUnit.MILLISECONDS;
+ }
+
+ throw new IllegalArgumentException("Invalid Arrow vector");
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDateVectorGetter.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDateVectorGetter.java
new file mode 100644
index 0000000000000..ea545851a3aeb
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDateVectorGetter.java
@@ -0,0 +1,67 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.calendar;
+
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DateMilliVector;
+import org.apache.arrow.vector.holders.NullableDateDayHolder;
+import org.apache.arrow.vector.holders.NullableDateMilliHolder;
+
+/**
+ * Auxiliary class used to unify data access on TimeStampVectors.
+ */
+final class ArrowFlightJdbcDateVectorGetter {
+
+ private ArrowFlightJdbcDateVectorGetter() {
+ // Prevent instantiation.
+ }
+
+ /**
+ * Auxiliary class meant to unify Date*Vector#get implementations with different classes of ValueHolders.
+ */
+ static class Holder {
+ int isSet; // Tells if value is set; 0 = not set, 1 = set
+ long value; // Holds actual value in its respective timeunit
+ }
+
+ /**
+ * Functional interface used to unify Date*Vector#get implementations.
+ */
+ @FunctionalInterface
+ interface Getter {
+ void get(int index, Holder holder);
+ }
+
+ static Getter createGetter(DateDayVector vector) {
+ NullableDateDayHolder auxHolder = new NullableDateDayHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ static Getter createGetter(DateMilliVector vector) {
+ NullableDateMilliHolder auxHolder = new NullableDateMilliHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDurationVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDurationVectorAccessor.java
new file mode 100644
index 0000000000000..22a0e6f892378
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcDurationVectorAccessor.java
@@ -0,0 +1,54 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.calendar;
+
+import java.time.Duration;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.DurationVector;
+
+/**
+ * Accessor for the Arrow type {@link DurationVector}.
+ */
+public class ArrowFlightJdbcDurationVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ private final DurationVector vector;
+
+ public ArrowFlightJdbcDurationVectorAccessor(DurationVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return Duration.class;
+ }
+
+ @Override
+ public Object getObject() {
+ Duration duration = vector.getObject(getCurrentRow());
+ this.wasNull = duration == null;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+
+ return duration;
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcIntervalVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcIntervalVectorAccessor.java
new file mode 100644
index 0000000000000..283dc9160a9e9
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcIntervalVectorAccessor.java
@@ -0,0 +1,126 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.calendar;
+
+import static org.apache.arrow.driver.jdbc.utils.IntervalStringUtils.formatIntervalDay;
+import static org.apache.arrow.driver.jdbc.utils.IntervalStringUtils.formatIntervalYear;
+import static org.apache.arrow.vector.util.DateUtility.yearsToMonths;
+
+import java.sql.SQLException;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.BaseFixedWidthVector;
+import org.apache.arrow.vector.IntervalDayVector;
+import org.apache.arrow.vector.IntervalYearVector;
+import org.apache.arrow.vector.holders.NullableIntervalDayHolder;
+import org.apache.arrow.vector.holders.NullableIntervalYearHolder;
+import org.joda.time.Period;
+
+/**
+ * Accessor for the Arrow type {@link IntervalDayVector}.
+ */
+public class ArrowFlightJdbcIntervalVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ private final BaseFixedWidthVector vector;
+ private final StringGetter stringGetter;
+ private final Class objectClass;
+
+ /**
+ * Instantiate an accessor for a {@link IntervalDayVector}.
+ *
+ * @param vector an instance of a IntervalDayVector.
+ * @param currentRowSupplier the supplier to track the rows.
+ * @param setCursorWasNull the consumer to set if value was null.
+ */
+ public ArrowFlightJdbcIntervalVectorAccessor(IntervalDayVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ stringGetter = (index) -> {
+ final NullableIntervalDayHolder holder = new NullableIntervalDayHolder();
+ vector.get(index, holder);
+ if (holder.isSet == 0) {
+ return null;
+ } else {
+ final int days = holder.days;
+ final int millis = holder.milliseconds;
+ return formatIntervalDay(new Period().plusDays(days).plusMillis(millis));
+ }
+ };
+ objectClass = java.time.Duration.class;
+ }
+
+ /**
+ * Instantiate an accessor for a {@link IntervalYearVector}.
+ *
+ * @param vector an instance of a IntervalYearVector.
+ * @param currentRowSupplier the supplier to track the rows.
+ * @param setCursorWasNull the consumer to set if value was null.
+ */
+ public ArrowFlightJdbcIntervalVectorAccessor(IntervalYearVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ stringGetter = (index) -> {
+ final NullableIntervalYearHolder holder = new NullableIntervalYearHolder();
+ vector.get(index, holder);
+ if (holder.isSet == 0) {
+ return null;
+ } else {
+ final int interval = holder.value;
+ final int years = (interval / yearsToMonths);
+ final int months = (interval % yearsToMonths);
+ return formatIntervalYear(new Period().plusYears(years).plusMonths(months));
+ }
+ };
+ objectClass = java.time.Period.class;
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return objectClass;
+ }
+
+ @Override
+ public String getString() throws SQLException {
+ String result = stringGetter.get(getCurrentRow());
+ wasNull = result == null;
+ wasNullConsumer.setWasNull(wasNull);
+ return result;
+ }
+
+ @Override
+ public Object getObject() {
+ Object object = vector.getObject(getCurrentRow());
+ wasNull = object == null;
+ wasNullConsumer.setWasNull(wasNull);
+ return object;
+ }
+
+ /**
+ * Functional interface used to unify Interval*Vector#getAsStringBuilder implementations.
+ */
+ @FunctionalInterface
+ interface StringGetter {
+ String get(int index);
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeStampVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeStampVectorAccessor.java
new file mode 100644
index 0000000000000..a23883baf1e9b
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeStampVectorAccessor.java
@@ -0,0 +1,185 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.calendar;
+
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeStampVectorGetter.Getter;
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeStampVectorGetter.Holder;
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeStampVectorGetter.createGetter;
+
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.Calendar;
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.TimeStampVector;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.util.DateUtility;
+
+/**
+ * Accessor for the Arrow types extending from {@link TimeStampVector}.
+ */
+public class ArrowFlightJdbcTimeStampVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ private final TimeZone timeZone;
+ private final Getter getter;
+ private final TimeUnit timeUnit;
+ private final LongToLocalDateTime longToLocalDateTime;
+ private final Holder holder;
+
+ /**
+ * Functional interface used to convert a number (in any time resolution) to LocalDateTime.
+ */
+ interface LongToLocalDateTime {
+ LocalDateTime fromLong(long value);
+ }
+
+ /**
+ * Instantiate a ArrowFlightJdbcTimeStampVectorAccessor for given vector.
+ */
+ public ArrowFlightJdbcTimeStampVectorAccessor(TimeStampVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.holder = new Holder();
+ this.getter = createGetter(vector);
+
+ this.timeZone = getTimeZoneForVector(vector);
+ this.timeUnit = getTimeUnitForVector(vector);
+ this.longToLocalDateTime = getLongToLocalDateTimeForVector(vector, this.timeZone);
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return Timestamp.class;
+ }
+
+ @Override
+ public Object getObject() {
+ return this.getTimestamp(null);
+ }
+
+ private LocalDateTime getLocalDateTime(Calendar calendar) {
+ getter.get(getCurrentRow(), holder);
+ this.wasNull = holder.isSet == 0;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+ if (this.wasNull) {
+ return null;
+ }
+
+ long value = holder.value;
+
+ LocalDateTime localDateTime = this.longToLocalDateTime.fromLong(value);
+
+ if (calendar != null) {
+ TimeZone timeZone = calendar.getTimeZone();
+ long millis = this.timeUnit.toMillis(value);
+ localDateTime = localDateTime
+ .minus(timeZone.getOffset(millis) - this.timeZone.getOffset(millis), ChronoUnit.MILLIS);
+ }
+ return localDateTime;
+ }
+
+ @Override
+ public Date getDate(Calendar calendar) {
+ LocalDateTime localDateTime = getLocalDateTime(calendar);
+ if (localDateTime == null) {
+ return null;
+ }
+
+ return new Date(Timestamp.valueOf(localDateTime).getTime());
+ }
+
+ @Override
+ public Time getTime(Calendar calendar) {
+ LocalDateTime localDateTime = getLocalDateTime(calendar);
+ if (localDateTime == null) {
+ return null;
+ }
+
+ return new Time(Timestamp.valueOf(localDateTime).getTime());
+ }
+
+ @Override
+ public Timestamp getTimestamp(Calendar calendar) {
+ LocalDateTime localDateTime = getLocalDateTime(calendar);
+ if (localDateTime == null) {
+ return null;
+ }
+
+ return Timestamp.valueOf(localDateTime);
+ }
+
+ protected static TimeUnit getTimeUnitForVector(TimeStampVector vector) {
+ ArrowType.Timestamp arrowType =
+ (ArrowType.Timestamp) vector.getField().getFieldType().getType();
+
+ switch (arrowType.getUnit()) {
+ case NANOSECOND:
+ return TimeUnit.NANOSECONDS;
+ case MICROSECOND:
+ return TimeUnit.MICROSECONDS;
+ case MILLISECOND:
+ return TimeUnit.MILLISECONDS;
+ case SECOND:
+ return TimeUnit.SECONDS;
+ default:
+ throw new UnsupportedOperationException("Invalid Arrow time unit");
+ }
+ }
+
+ protected static LongToLocalDateTime getLongToLocalDateTimeForVector(TimeStampVector vector,
+ TimeZone timeZone) {
+ String timeZoneID = timeZone.getID();
+
+ ArrowType.Timestamp arrowType =
+ (ArrowType.Timestamp) vector.getField().getFieldType().getType();
+
+ switch (arrowType.getUnit()) {
+ case NANOSECOND:
+ return nanoseconds -> DateUtility.getLocalDateTimeFromEpochNano(nanoseconds, timeZoneID);
+ case MICROSECOND:
+ return microseconds -> DateUtility.getLocalDateTimeFromEpochMicro(microseconds, timeZoneID);
+ case MILLISECOND:
+ return milliseconds -> DateUtility.getLocalDateTimeFromEpochMilli(milliseconds, timeZoneID);
+ case SECOND:
+ return seconds -> DateUtility.getLocalDateTimeFromEpochMilli(
+ TimeUnit.SECONDS.toMillis(seconds), timeZoneID);
+ default:
+ throw new UnsupportedOperationException("Invalid Arrow time unit");
+ }
+ }
+
+ protected static TimeZone getTimeZoneForVector(TimeStampVector vector) {
+ ArrowType.Timestamp arrowType =
+ (ArrowType.Timestamp) vector.getField().getFieldType().getType();
+
+ String timezoneName = arrowType.getTimezone();
+ if (timezoneName == null) {
+ return TimeZone.getTimeZone("UTC");
+ }
+
+ return TimeZone.getTimeZone(timezoneName);
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeStampVectorGetter.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeStampVectorGetter.java
new file mode 100644
index 0000000000000..03fb35face722
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeStampVectorGetter.java
@@ -0,0 +1,156 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.calendar;
+
+import org.apache.arrow.vector.TimeStampMicroTZVector;
+import org.apache.arrow.vector.TimeStampMicroVector;
+import org.apache.arrow.vector.TimeStampMilliTZVector;
+import org.apache.arrow.vector.TimeStampMilliVector;
+import org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.arrow.vector.TimeStampNanoVector;
+import org.apache.arrow.vector.TimeStampSecTZVector;
+import org.apache.arrow.vector.TimeStampSecVector;
+import org.apache.arrow.vector.TimeStampVector;
+import org.apache.arrow.vector.holders.NullableTimeStampMicroHolder;
+import org.apache.arrow.vector.holders.NullableTimeStampMicroTZHolder;
+import org.apache.arrow.vector.holders.NullableTimeStampMilliHolder;
+import org.apache.arrow.vector.holders.NullableTimeStampMilliTZHolder;
+import org.apache.arrow.vector.holders.NullableTimeStampNanoHolder;
+import org.apache.arrow.vector.holders.NullableTimeStampNanoTZHolder;
+import org.apache.arrow.vector.holders.NullableTimeStampSecHolder;
+import org.apache.arrow.vector.holders.NullableTimeStampSecTZHolder;
+
+/**
+ * Auxiliary class used to unify data access on TimeStampVectors.
+ */
+final class ArrowFlightJdbcTimeStampVectorGetter {
+
+ private ArrowFlightJdbcTimeStampVectorGetter() {
+ // Prevent instantiation.
+ }
+
+ /**
+ * Auxiliary class meant to unify TimeStamp*Vector#get implementations with different classes of ValueHolders.
+ */
+ static class Holder {
+ int isSet; // Tells if value is set; 0 = not set, 1 = set
+ long value; // Holds actual value in its respective timeunit
+ }
+
+ /**
+ * Functional interface used to unify TimeStamp*Vector#get implementations.
+ */
+ @FunctionalInterface
+ interface Getter {
+ void get(int index, Holder holder);
+ }
+
+ static Getter createGetter(TimeStampVector vector) {
+ if (vector instanceof TimeStampNanoVector) {
+ return createGetter((TimeStampNanoVector) vector);
+ } else if (vector instanceof TimeStampNanoTZVector) {
+ return createGetter((TimeStampNanoTZVector) vector);
+ } else if (vector instanceof TimeStampMicroVector) {
+ return createGetter((TimeStampMicroVector) vector);
+ } else if (vector instanceof TimeStampMicroTZVector) {
+ return createGetter((TimeStampMicroTZVector) vector);
+ } else if (vector instanceof TimeStampMilliVector) {
+ return createGetter((TimeStampMilliVector) vector);
+ } else if (vector instanceof TimeStampMilliTZVector) {
+ return createGetter((TimeStampMilliTZVector) vector);
+ } else if (vector instanceof TimeStampSecVector) {
+ return createGetter((TimeStampSecVector) vector);
+ } else if (vector instanceof TimeStampSecTZVector) {
+ return createGetter((TimeStampSecTZVector) vector);
+ }
+
+ throw new UnsupportedOperationException("Unsupported Timestamp vector type");
+ }
+
+ private static Getter createGetter(TimeStampNanoVector vector) {
+ NullableTimeStampNanoHolder auxHolder = new NullableTimeStampNanoHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ private static Getter createGetter(TimeStampNanoTZVector vector) {
+ NullableTimeStampNanoTZHolder auxHolder = new NullableTimeStampNanoTZHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ private static Getter createGetter(TimeStampMicroVector vector) {
+ NullableTimeStampMicroHolder auxHolder = new NullableTimeStampMicroHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ private static Getter createGetter(TimeStampMicroTZVector vector) {
+ NullableTimeStampMicroTZHolder auxHolder = new NullableTimeStampMicroTZHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ private static Getter createGetter(TimeStampMilliVector vector) {
+ NullableTimeStampMilliHolder auxHolder = new NullableTimeStampMilliHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ private static Getter createGetter(TimeStampMilliTZVector vector) {
+ NullableTimeStampMilliTZHolder auxHolder = new NullableTimeStampMilliTZHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ private static Getter createGetter(TimeStampSecVector vector) {
+ NullableTimeStampSecHolder auxHolder = new NullableTimeStampSecHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ private static Getter createGetter(TimeStampSecTZVector vector) {
+ NullableTimeStampSecTZHolder auxHolder = new NullableTimeStampSecTZHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeVectorAccessor.java
new file mode 100644
index 0000000000000..6c2173d5e5656
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeVectorAccessor.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.driver.jdbc.accessor.impl.calendar;
+
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeVectorGetter.Getter;
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeVectorGetter.Holder;
+import static org.apache.arrow.driver.jdbc.accessor.impl.calendar.ArrowFlightJdbcTimeVectorGetter.createGetter;
+
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Calendar;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.ArrowFlightJdbcTime;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.driver.jdbc.utils.DateTimeUtils;
+import org.apache.arrow.vector.TimeMicroVector;
+import org.apache.arrow.vector.TimeMilliVector;
+import org.apache.arrow.vector.TimeNanoVector;
+import org.apache.arrow.vector.TimeSecVector;
+import org.apache.arrow.vector.ValueVector;
+
+/**
+ * Accessor for the Arrow types: {@link TimeNanoVector}, {@link TimeMicroVector}, {@link TimeMilliVector}
+ * and {@link TimeSecVector}.
+ */
+public class ArrowFlightJdbcTimeVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ private final Getter getter;
+ private final TimeUnit timeUnit;
+ private final Holder holder;
+
+ /**
+ * Instantiate an accessor for a {@link TimeNanoVector}.
+ *
+ * @param vector an instance of a TimeNanoVector.
+ * @param currentRowSupplier the supplier to track the lines.
+ * @param setCursorWasNull the consumer to set if value was null.
+ */
+ public ArrowFlightJdbcTimeVectorAccessor(TimeNanoVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.holder = new Holder();
+ this.getter = createGetter(vector);
+ this.timeUnit = getTimeUnitForVector(vector);
+ }
+
+ /**
+ * Instantiate an accessor for a {@link TimeMicroVector}.
+ *
+ * @param vector an instance of a TimeMicroVector.
+ * @param currentRowSupplier the supplier to track the lines.
+ * @param setCursorWasNull the consumer to set if value was null.
+ */
+ public ArrowFlightJdbcTimeVectorAccessor(TimeMicroVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.holder = new Holder();
+ this.getter = createGetter(vector);
+ this.timeUnit = getTimeUnitForVector(vector);
+ }
+
+ /**
+ * Instantiate an accessor for a {@link TimeMilliVector}.
+ *
+ * @param vector an instance of a TimeMilliVector.
+ * @param currentRowSupplier the supplier to track the lines.
+ */
+ public ArrowFlightJdbcTimeVectorAccessor(TimeMilliVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.holder = new Holder();
+ this.getter = createGetter(vector);
+ this.timeUnit = getTimeUnitForVector(vector);
+ }
+
+ /**
+ * Instantiate an accessor for a {@link TimeSecVector}.
+ *
+ * @param vector an instance of a TimeSecVector.
+ * @param currentRowSupplier the supplier to track the lines.
+ */
+ public ArrowFlightJdbcTimeVectorAccessor(TimeSecVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.holder = new Holder();
+ this.getter = createGetter(vector);
+ this.timeUnit = getTimeUnitForVector(vector);
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return Time.class;
+ }
+
+ @Override
+ public Object getObject() {
+ return this.getTime(null);
+ }
+
+ @Override
+ public Time getTime(Calendar calendar) {
+ fillHolder();
+ if (this.wasNull) {
+ return null;
+ }
+
+ long value = holder.value;
+ long milliseconds = this.timeUnit.toMillis(value);
+
+ return new ArrowFlightJdbcTime(DateTimeUtils.applyCalendarOffset(milliseconds, calendar));
+ }
+
+ private void fillHolder() {
+ getter.get(getCurrentRow(), holder);
+ this.wasNull = holder.isSet == 0;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+ }
+
+ @Override
+ public Timestamp getTimestamp(Calendar calendar) {
+ Time time = getTime(calendar);
+ if (time == null) {
+ return null;
+ }
+ return new Timestamp(time.getTime());
+ }
+
+ protected static TimeUnit getTimeUnitForVector(ValueVector vector) {
+ if (vector instanceof TimeNanoVector) {
+ return TimeUnit.NANOSECONDS;
+ } else if (vector instanceof TimeMicroVector) {
+ return TimeUnit.MICROSECONDS;
+ } else if (vector instanceof TimeMilliVector) {
+ return TimeUnit.MILLISECONDS;
+ } else if (vector instanceof TimeSecVector) {
+ return TimeUnit.SECONDS;
+ }
+
+ throw new IllegalArgumentException("Invalid Arrow vector");
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeVectorGetter.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeVectorGetter.java
new file mode 100644
index 0000000000000..fb254c694014d
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/calendar/ArrowFlightJdbcTimeVectorGetter.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.driver.jdbc.accessor.impl.calendar;
+
+import org.apache.arrow.vector.TimeMicroVector;
+import org.apache.arrow.vector.TimeMilliVector;
+import org.apache.arrow.vector.TimeNanoVector;
+import org.apache.arrow.vector.TimeSecVector;
+import org.apache.arrow.vector.holders.NullableTimeMicroHolder;
+import org.apache.arrow.vector.holders.NullableTimeMilliHolder;
+import org.apache.arrow.vector.holders.NullableTimeNanoHolder;
+import org.apache.arrow.vector.holders.NullableTimeSecHolder;
+
+/**
+ * Auxiliary class used to unify data access on Time*Vectors.
+ */
+final class ArrowFlightJdbcTimeVectorGetter {
+
+ private ArrowFlightJdbcTimeVectorGetter() {
+ // Prevent instantiation.
+ }
+
+ /**
+ * Auxiliary class meant to unify TimeStamp*Vector#get implementations with different classes of ValueHolders.
+ */
+ static class Holder {
+ int isSet; // Tells if value is set; 0 = not set, 1 = set
+ long value; // Holds actual value in its respective timeunit
+ }
+
+ /**
+ * Functional interface used to unify TimeStamp*Vector#get implementations.
+ */
+ @FunctionalInterface
+ interface Getter {
+ void get(int index, Holder holder);
+ }
+
+ static Getter createGetter(TimeNanoVector vector) {
+ NullableTimeNanoHolder auxHolder = new NullableTimeNanoHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ static Getter createGetter(TimeMicroVector vector) {
+ NullableTimeMicroHolder auxHolder = new NullableTimeMicroHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ static Getter createGetter(TimeMilliVector vector) {
+ NullableTimeMilliHolder auxHolder = new NullableTimeMilliHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+
+ static Getter createGetter(TimeSecVector vector) {
+ NullableTimeSecHolder auxHolder = new NullableTimeSecHolder();
+ return (index, holder) -> {
+ vector.get(index, auxHolder);
+ holder.isSet = auxHolder.isSet;
+ holder.value = auxHolder.value;
+ };
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/AbstractArrowFlightJdbcListVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/AbstractArrowFlightJdbcListVectorAccessor.java
new file mode 100644
index 0000000000000..d3338608f8352
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/AbstractArrowFlightJdbcListVectorAccessor.java
@@ -0,0 +1,73 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.sql.Array;
+import java.util.List;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.ArrowFlightJdbcArray;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.FixedSizeListVector;
+import org.apache.arrow.vector.complex.LargeListVector;
+import org.apache.arrow.vector.complex.ListVector;
+
+/**
+ * Base Accessor for the Arrow types {@link ListVector}, {@link LargeListVector} and {@link FixedSizeListVector}.
+ */
+public abstract class AbstractArrowFlightJdbcListVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ protected AbstractArrowFlightJdbcListVectorAccessor(IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return List.class;
+ }
+
+ protected abstract long getStartOffset(int index);
+
+ protected abstract long getEndOffset(int index);
+
+ protected abstract FieldVector getDataVector();
+
+ protected abstract boolean isNull(int index);
+
+ @Override
+ public final Array getArray() {
+ int index = getCurrentRow();
+ FieldVector dataVector = getDataVector();
+
+ this.wasNull = isNull(index);
+ this.wasNullConsumer.setWasNull(this.wasNull);
+ if (this.wasNull) {
+ return null;
+ }
+
+ long startOffset = getStartOffset(index);
+ long endOffset = getEndOffset(index);
+
+ long valuesCount = endOffset - startOffset;
+ return new ArrowFlightJdbcArray(dataVector, startOffset, valuesCount);
+ }
+}
+
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/AbstractArrowFlightJdbcUnionVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/AbstractArrowFlightJdbcUnionVectorAccessor.java
new file mode 100644
index 0000000000000..0465765f18358
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/AbstractArrowFlightJdbcUnionVectorAccessor.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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.net.URL;
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.Clob;
+import java.sql.Date;
+import java.sql.NClob;
+import java.sql.Ref;
+import java.sql.SQLException;
+import java.sql.SQLXML;
+import java.sql.Struct;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Calendar;
+import java.util.Map;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.driver.jdbc.accessor.impl.ArrowFlightJdbcNullVectorAccessor;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.complex.DenseUnionVector;
+import org.apache.arrow.vector.complex.UnionVector;
+
+/**
+ * Base accessor for {@link UnionVector} and {@link DenseUnionVector}.
+ */
+public abstract class AbstractArrowFlightJdbcUnionVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ /**
+ * Array of accessors for each type contained in UnionVector.
+ * Index corresponds to UnionVector and DenseUnionVector typeIds which are both limited to 128.
+ */
+ private final ArrowFlightJdbcAccessor[] accessors = new ArrowFlightJdbcAccessor[128];
+
+ private final ArrowFlightJdbcNullVectorAccessor nullAccessor =
+ new ArrowFlightJdbcNullVectorAccessor((boolean wasNull) -> {
+ });
+
+ protected AbstractArrowFlightJdbcUnionVectorAccessor(IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ }
+
+ protected abstract ArrowFlightJdbcAccessor createAccessorForVector(ValueVector vector);
+
+ protected abstract byte getCurrentTypeId();
+
+ protected abstract ValueVector getVectorByTypeId(byte typeId);
+
+ /**
+ * Returns an accessor for UnionVector child vector on current row.
+ *
+ * @return ArrowFlightJdbcAccessor for child vector on current row.
+ */
+ protected ArrowFlightJdbcAccessor getAccessor() {
+ // Get the typeId and child vector for the current row being accessed.
+ byte typeId = this.getCurrentTypeId();
+ ValueVector vector = this.getVectorByTypeId(typeId);
+
+ if (typeId < 0) {
+ // typeId may be negative if the current row has no type defined.
+ return this.nullAccessor;
+ }
+
+ // Ensure there is an accessor for given typeId
+ if (this.accessors[typeId] == null) {
+ this.accessors[typeId] = this.createAccessorForVector(vector);
+ }
+
+ return this.accessors[typeId];
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return getAccessor().getObjectClass();
+ }
+
+ @Override
+ public boolean wasNull() {
+ return getAccessor().wasNull();
+ }
+
+ @Override
+ public String getString() throws SQLException {
+ return getAccessor().getString();
+ }
+
+ @Override
+ public boolean getBoolean() throws SQLException {
+ return getAccessor().getBoolean();
+ }
+
+ @Override
+ public byte getByte() throws SQLException {
+ return getAccessor().getByte();
+ }
+
+ @Override
+ public short getShort() throws SQLException {
+ return getAccessor().getShort();
+ }
+
+ @Override
+ public int getInt() throws SQLException {
+ return getAccessor().getInt();
+ }
+
+ @Override
+ public long getLong() throws SQLException {
+ return getAccessor().getLong();
+ }
+
+ @Override
+ public float getFloat() throws SQLException {
+ return getAccessor().getFloat();
+ }
+
+ @Override
+ public double getDouble() throws SQLException {
+ return getAccessor().getDouble();
+ }
+
+ @Override
+ public BigDecimal getBigDecimal() throws SQLException {
+ return getAccessor().getBigDecimal();
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(int i) throws SQLException {
+ return getAccessor().getBigDecimal(i);
+ }
+
+ @Override
+ public byte[] getBytes() throws SQLException {
+ return getAccessor().getBytes();
+ }
+
+ @Override
+ public InputStream getAsciiStream() throws SQLException {
+ return getAccessor().getAsciiStream();
+ }
+
+ @Override
+ public InputStream getUnicodeStream() throws SQLException {
+ return getAccessor().getUnicodeStream();
+ }
+
+ @Override
+ public InputStream getBinaryStream() throws SQLException {
+ return getAccessor().getBinaryStream();
+ }
+
+ @Override
+ public Object getObject() throws SQLException {
+ return getAccessor().getObject();
+ }
+
+ @Override
+ public Reader getCharacterStream() throws SQLException {
+ return getAccessor().getCharacterStream();
+ }
+
+ @Override
+ public Object getObject(Map> map) throws SQLException {
+ return getAccessor().getObject(map);
+ }
+
+ @Override
+ public Ref getRef() throws SQLException {
+ return getAccessor().getRef();
+ }
+
+ @Override
+ public Blob getBlob() throws SQLException {
+ return getAccessor().getBlob();
+ }
+
+ @Override
+ public Clob getClob() throws SQLException {
+ return getAccessor().getClob();
+ }
+
+ @Override
+ public Array getArray() throws SQLException {
+ return getAccessor().getArray();
+ }
+
+ @Override
+ public Struct getStruct() throws SQLException {
+ return getAccessor().getStruct();
+ }
+
+ @Override
+ public Date getDate(Calendar calendar) throws SQLException {
+ return getAccessor().getDate(calendar);
+ }
+
+ @Override
+ public Time getTime(Calendar calendar) throws SQLException {
+ return getAccessor().getTime(calendar);
+ }
+
+ @Override
+ public Timestamp getTimestamp(Calendar calendar) throws SQLException {
+ return getAccessor().getTimestamp(calendar);
+ }
+
+ @Override
+ public URL getURL() throws SQLException {
+ return getAccessor().getURL();
+ }
+
+ @Override
+ public NClob getNClob() throws SQLException {
+ return getAccessor().getNClob();
+ }
+
+ @Override
+ public SQLXML getSQLXML() throws SQLException {
+ return getAccessor().getSQLXML();
+ }
+
+ @Override
+ public String getNString() throws SQLException {
+ return getAccessor().getNString();
+ }
+
+ @Override
+ public Reader getNCharacterStream() throws SQLException {
+ return getAccessor().getNCharacterStream();
+ }
+
+ @Override
+ public T getObject(Class type) throws SQLException {
+ return getAccessor().getObject(type);
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcDenseUnionVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcDenseUnionVectorAccessor.java
new file mode 100644
index 0000000000000..ba5b83ade63ab
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcDenseUnionVectorAccessor.java
@@ -0,0 +1,66 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.complex.DenseUnionVector;
+
+/**
+ * Accessor for the Arrow type {@link DenseUnionVector}.
+ */
+public class ArrowFlightJdbcDenseUnionVectorAccessor
+ extends AbstractArrowFlightJdbcUnionVectorAccessor {
+
+ private final DenseUnionVector vector;
+
+ /**
+ * Instantiate an accessor for a {@link DenseUnionVector}.
+ *
+ * @param vector an instance of a DenseUnionVector.
+ * @param currentRowSupplier the supplier to track the rows.
+ * @param setCursorWasNull the consumer to set if value was null.
+ */
+ public ArrowFlightJdbcDenseUnionVectorAccessor(DenseUnionVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ }
+
+ @Override
+ protected ArrowFlightJdbcAccessor createAccessorForVector(ValueVector vector) {
+ return ArrowFlightJdbcAccessorFactory.createAccessor(vector,
+ () -> this.vector.getOffset(this.getCurrentRow()), (boolean wasNull) -> {
+ });
+ }
+
+ @Override
+ protected byte getCurrentTypeId() {
+ int index = getCurrentRow();
+ return this.vector.getTypeId(index);
+ }
+
+ @Override
+ protected ValueVector getVectorByTypeId(byte typeId) {
+ return this.vector.getVectorByType(typeId);
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcFixedSizeListVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcFixedSizeListVectorAccessor.java
new file mode 100644
index 0000000000000..7bdd3abfd0cad
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcFixedSizeListVectorAccessor.java
@@ -0,0 +1,70 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.util.List;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.FixedSizeListVector;
+
+/**
+ * Accessor for the Arrow type {@link FixedSizeListVector}.
+ */
+public class ArrowFlightJdbcFixedSizeListVectorAccessor
+ extends AbstractArrowFlightJdbcListVectorAccessor {
+
+ private final FixedSizeListVector vector;
+
+ public ArrowFlightJdbcFixedSizeListVectorAccessor(FixedSizeListVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ }
+
+ @Override
+ protected long getStartOffset(int index) {
+ return (long) vector.getListSize() * index;
+ }
+
+ @Override
+ protected long getEndOffset(int index) {
+ return (long) vector.getListSize() * (index + 1);
+ }
+
+ @Override
+ protected FieldVector getDataVector() {
+ return vector.getDataVector();
+ }
+
+ @Override
+ protected boolean isNull(int index) {
+ return vector.isNull(index);
+ }
+
+ @Override
+ public Object getObject() {
+ List object = vector.getObject(getCurrentRow());
+ this.wasNull = object == null;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+
+ return object;
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcLargeListVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcLargeListVectorAccessor.java
new file mode 100644
index 0000000000000..f7608bb06e5fe
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcLargeListVectorAccessor.java
@@ -0,0 +1,70 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.util.List;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.LargeListVector;
+
+/**
+ * Accessor for the Arrow type {@link LargeListVector}.
+ */
+public class ArrowFlightJdbcLargeListVectorAccessor
+ extends AbstractArrowFlightJdbcListVectorAccessor {
+
+ private final LargeListVector vector;
+
+ public ArrowFlightJdbcLargeListVectorAccessor(LargeListVector vector,
+ IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ }
+
+ @Override
+ protected long getStartOffset(int index) {
+ return vector.getOffsetBuffer().getLong((long) index * LargeListVector.OFFSET_WIDTH);
+ }
+
+ @Override
+ protected long getEndOffset(int index) {
+ return vector.getOffsetBuffer().getLong((long) (index + 1) * LargeListVector.OFFSET_WIDTH);
+ }
+
+ @Override
+ protected FieldVector getDataVector() {
+ return vector.getDataVector();
+ }
+
+ @Override
+ protected boolean isNull(int index) {
+ return vector.isNull(index);
+ }
+
+ @Override
+ public Object getObject() {
+ List object = vector.getObject(getCurrentRow());
+ this.wasNull = object == null;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+
+ return object;
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcListVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcListVectorAccessor.java
new file mode 100644
index 0000000000000..a329a344073a6
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcListVectorAccessor.java
@@ -0,0 +1,70 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.util.List;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.BaseRepeatedValueVector;
+import org.apache.arrow.vector.complex.ListVector;
+
+/**
+ * Accessor for the Arrow type {@link ListVector}.
+ */
+public class ArrowFlightJdbcListVectorAccessor extends AbstractArrowFlightJdbcListVectorAccessor {
+
+ private final ListVector vector;
+
+ public ArrowFlightJdbcListVectorAccessor(ListVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ }
+
+ @Override
+ protected long getStartOffset(int index) {
+ return vector.getOffsetBuffer().getInt((long) index * BaseRepeatedValueVector.OFFSET_WIDTH);
+ }
+
+ @Override
+ protected long getEndOffset(int index) {
+ return vector.getOffsetBuffer()
+ .getInt((long) (index + 1) * BaseRepeatedValueVector.OFFSET_WIDTH);
+ }
+
+ @Override
+ protected FieldVector getDataVector() {
+ return vector.getDataVector();
+ }
+
+ @Override
+ protected boolean isNull(int index) {
+ return vector.isNull(index);
+ }
+
+ @Override
+ public Object getObject() {
+ List object = vector.getObject(getCurrentRow());
+ this.wasNull = object == null;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+
+ return object;
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcMapVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcMapVectorAccessor.java
new file mode 100644
index 0000000000000..bf1225b33de64
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcMapVectorAccessor.java
@@ -0,0 +1,92 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.util.Map;
+import java.util.function.IntSupplier;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.BaseRepeatedValueVector;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.complex.impl.UnionMapReader;
+import org.apache.arrow.vector.util.JsonStringHashMap;
+
+/**
+ * Accessor for the Arrow type {@link MapVector}.
+ */
+public class ArrowFlightJdbcMapVectorAccessor extends AbstractArrowFlightJdbcListVectorAccessor {
+
+ private final MapVector vector;
+
+ public ArrowFlightJdbcMapVectorAccessor(MapVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return Map.class;
+ }
+
+ @Override
+ public Object getObject() {
+ int index = getCurrentRow();
+
+ this.wasNull = vector.isNull(index);
+ this.wasNullConsumer.setWasNull(this.wasNull);
+ if (this.wasNull) {
+ return null;
+ }
+
+ Map result = new JsonStringHashMap<>();
+ UnionMapReader reader = vector.getReader();
+
+ reader.setPosition(index);
+ while (reader.next()) {
+ Object key = reader.key().readObject();
+ Object value = reader.value().readObject();
+
+ result.put(key, value);
+ }
+
+ return result;
+ }
+
+ @Override
+ protected long getStartOffset(int index) {
+ return vector.getOffsetBuffer().getInt((long) index * BaseRepeatedValueVector.OFFSET_WIDTH);
+ }
+
+ @Override
+ protected long getEndOffset(int index) {
+ return vector.getOffsetBuffer()
+ .getInt((long) (index + 1) * BaseRepeatedValueVector.OFFSET_WIDTH);
+ }
+
+ @Override
+ protected boolean isNull(int index) {
+ return vector.isNull(index);
+ }
+
+ @Override
+ protected FieldVector getDataVector() {
+ return vector.getDataVector();
+ }
+}
diff --git a/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcStructVectorAccessor.java b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcStructVectorAccessor.java
new file mode 100644
index 0000000000000..8a7ac117113c5
--- /dev/null
+++ b/java/flight/flight-sql-jdbc-driver/src/main/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcStructVectorAccessor.java
@@ -0,0 +1,75 @@
+/*
+ * 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.arrow.driver.jdbc.accessor.impl.complex;
+
+import java.sql.Struct;
+import java.util.List;
+import java.util.Map;
+import java.util.function.IntSupplier;
+import java.util.stream.Collectors;
+
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessor;
+import org.apache.arrow.driver.jdbc.accessor.ArrowFlightJdbcAccessorFactory;
+import org.apache.arrow.vector.complex.StructVector;
+import org.apache.calcite.avatica.util.StructImpl;
+
+/**
+ * Accessor for the Arrow type {@link StructVector}.
+ */
+public class ArrowFlightJdbcStructVectorAccessor extends ArrowFlightJdbcAccessor {
+
+ private final StructVector vector;
+
+ public ArrowFlightJdbcStructVectorAccessor(StructVector vector, IntSupplier currentRowSupplier,
+ ArrowFlightJdbcAccessorFactory.WasNullConsumer setCursorWasNull) {
+ super(currentRowSupplier, setCursorWasNull);
+ this.vector = vector;
+ }
+
+ @Override
+ public Class getObjectClass() {
+ return Map.class;
+ }
+
+ @Override
+ public Object getObject() {
+ Map object = vector.getObject(getCurrentRow());
+ this.wasNull = object == null;
+ this.wasNullConsumer.setWasNull(this.wasNull);
+
+ return object;
+ }
+
+ @Override
+ public Struct getStruct() {
+ int currentRow = getCurrentRow();
+
+ this.wasNull = vector.isNull(currentRow);
+ this.wasNullConsumer.setWasNull(this.wasNull);
+ if (this.wasNull) {
+ return null;
+ }
+
+ List