diff --git a/dev/release/rat_exclude_files.txt b/dev/release/rat_exclude_files.txt index ea54278ede7..5b29f39214e 100644 --- a/dev/release/rat_exclude_files.txt +++ b/dev/release/rat_exclude_files.txt @@ -232,3 +232,4 @@ r/vignettes/*.Rmd ruby/red-arrow/.yardopts rust/arrow/test/data/*.csv rust/rust-toolchain +java/flight/flight-jdbc/src/main/resources/META-INF/services/java.sql.Driver diff --git a/java/flight/flight-jdbc/README.md b/java/flight/flight-jdbc/README.md new file mode 100644 index 00000000000..2673fbbda27 --- /dev/null +++ b/java/flight/flight-jdbc/README.md @@ -0,0 +1,22 @@ + + +# Apache Arrow Flight JDBC Driver + +JDBC Driver for executing queries against Flight servers. diff --git a/java/flight/flight-jdbc/pom.xml b/java/flight/flight-jdbc/pom.xml new file mode 100644 index 00000000000..3d8bdc9522d --- /dev/null +++ b/java/flight/flight-jdbc/pom.xml @@ -0,0 +1,254 @@ + + + + 4.0.0 + + org.apache.arrow + arrow-java-root + 0.16.0-SNAPSHOT + ../../pom.xml + + + flight-jdbc + Arrow Flight JDBC + (Experimental)Flight JDBC Driver + jar + + + 1.24.0 + 3.7.1 + 1 + + + + + org.apache.arrow + flight-core + ${project.version} + + + org.apache.arrow + arrow-memory + ${project.version} + compile + + + org.apache.arrow + arrow-vector + ${project.version} + compile + + + io.grpc + grpc-api + ${dep.grpc.version} + + + io.grpc + grpc-stub + ${dep.grpc.version} + + + io.grpc + grpc-protobuf + ${dep.grpc.version} + + + com.google.protobuf + protobuf-java + ${dep.protobuf.version} + + + com.google.guava + guava + + + org.slf4j + slf4j-api + + + + + + + kr.motd.maven + os-maven-plugin + 1.5.0.Final + + + + + maven-surefire-plugin + + false + + ${project.basedir}/../../../testing/data + + + + + org.apache.maven.plugins + maven-shade-plugin + 3.1.1 + + + shade-main + package + + shade + + + true + shaded + + + io.grpc:* + com.google.protobuf:* + + + + + com.google.protobuf + arrow.flight.com.google.protobuf + + + + + + + + + shade-ext + package + + shade + + + true + shaded-ext + + + io.grpc:* + com.google.protobuf:* + com.google.guava:* + + + + + com.google.protobuf + arrow.flight.com.google.protobuf + + + com.google.common + arrow.flight.com.google.common + + + + + + + + + + + org.xolstice.maven.plugins + protobuf-maven-plugin + 0.5.0 + + com.google.protobuf:protoc:${dep.protobuf.version}:exe:${os.detected.classifier} + false + grpc-java + io.grpc:protoc-gen-grpc-java:${dep.grpc.version}:exe:${os.detected.classifier} + + + + src + + ${basedir}/../../../format/ + ${project.build.directory}/generated-sources/protobuf + + + compile + compile-custom + + + + test + + ${basedir}/src/test/protobuf + ${project.build.directory}/generated-test-sources//protobuf + + + compile + compile-custom + + + + + + org.apache.maven.plugins + maven-dependency-plugin + + + analyze + verify + + analyze-only + + + + io.netty:netty-tcnative-boringssl-static:* + + + + + + + org.codehaus.mojo + build-helper-maven-plugin + 1.9.1 + + + add-generated-sources-to-classpath + generate-sources + + add-source + + + + ${project.build.directory}/generated-sources/protobuf + + + + + + + maven-assembly-plugin + 3.0.0 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + + diff --git a/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/Driver.java b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/Driver.java new file mode 100644 index 00000000000..73d499d0d60 --- /dev/null +++ b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/Driver.java @@ -0,0 +1,81 @@ +/* + * 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.jdbc; + +import java.sql.Connection; +import java.sql.DriverPropertyInfo; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.util.Properties; +import java.util.logging.Logger; + +import org.slf4j.LoggerFactory; + +/** + * Driver. + */ +public class Driver implements java.sql.Driver { + + private static final org.slf4j.Logger logger = LoggerFactory.getLogger(Driver.class); + + /** JDBC connection string prefix. */ + private static final String PREFIX = "jdbc:arrow://"; + + @Override + public Connection connect(String url, Properties properties) throws SQLException { + logger.info("connect() url={}", url); + //TODO this needs much more work to parse full URLs but this is enough to get end to end tests running + String c = url.substring(PREFIX.length()); + int i = c.indexOf(':'); + if (i == -1) { + return new FlightConnection(c, 50051); + } else { + return new FlightConnection(c.substring(0,i), Integer.parseInt(c.substring(i + 1))); + } + } + + @Override + public boolean acceptsURL(String url) throws SQLException { + return url != null && url.startsWith(PREFIX); + } + + @Override + public DriverPropertyInfo[] getPropertyInfo(String s, Properties properties) throws SQLException { + return new DriverPropertyInfo[0]; + } + + @Override + public int getMajorVersion() { + return 0; + } + + @Override + public int getMinorVersion() { + return 16; + } + + @Override + public boolean jdbcCompliant() { + return false; + } + + @Override + public Logger getParentLogger() throws SQLFeatureNotSupportedException { + throw new SQLFeatureNotSupportedException(); + } +} diff --git a/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightConnection.java b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightConnection.java new file mode 100644 index 00000000000..27735fbae21 --- /dev/null +++ b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightConnection.java @@ -0,0 +1,320 @@ +/* + * 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.jdbc; + +import java.sql.Array; +import java.sql.Blob; +import java.sql.CallableStatement; +import java.sql.Clob; +import java.sql.DatabaseMetaData; +import java.sql.NClob; +import java.sql.SQLClientInfoException; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.SQLWarning; +import java.sql.SQLXML; +import java.sql.Savepoint; +import java.sql.Struct; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.Executor; + +/** + * Connection. + */ +public class FlightConnection implements java.sql.Connection { + + protected final String host; + protected final int port; + + public FlightConnection(String host, int port) { + this.host = host; + this.port = port; + } + + @Override + public FlightStatement createStatement() throws SQLException { + return new FlightStatement(this); + } + + @Override + public FlightPreparedStatement prepareStatement(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public CallableStatement prepareCall(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public String nativeSQL(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean getAutoCommit() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setAutoCommit(boolean b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void commit() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void rollback() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void close() throws SQLException { + + } + + @Override + public boolean isClosed() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public DatabaseMetaData getMetaData() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isReadOnly() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setReadOnly(boolean b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public String getCatalog() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setCatalog(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getTransactionIsolation() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setTransactionIsolation(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public SQLWarning getWarnings() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void clearWarnings() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public FlightStatement createStatement(int i, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public FlightPreparedStatement prepareStatement(String s, int i, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public CallableStatement prepareCall(String s, int i, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Map> getTypeMap() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setTypeMap(Map> map) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getHoldability() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setHoldability(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Savepoint setSavepoint() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Savepoint setSavepoint(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void rollback(Savepoint savepoint) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void releaseSavepoint(Savepoint savepoint) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public FlightStatement createStatement(int i, int i1, int i2) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public FlightPreparedStatement prepareStatement(String s, int i, int i1, int i2) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public CallableStatement prepareCall(String s, int i, int i1, int i2) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public FlightPreparedStatement prepareStatement(String s, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public FlightPreparedStatement prepareStatement(String s, int[] ints) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public FlightPreparedStatement prepareStatement(String s, String[] strings) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Clob createClob() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Blob createBlob() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public NClob createNClob() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public SQLXML createSQLXML() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isValid(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setClientInfo(String s, String s1) throws SQLClientInfoException { + throw new SQLClientInfoException(); + + } + + @Override + public String getClientInfo(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Properties getClientInfo() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setClientInfo(Properties properties) throws SQLClientInfoException { + throw new SQLClientInfoException(); + } + + @Override + public Array createArrayOf(String s, Object[] objects) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Struct createStruct(String s, Object[] objects) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public String getSchema() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setSchema(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void abort(Executor executor) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNetworkTimeout(Executor executor, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getNetworkTimeout() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public T unwrap(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isWrapperFor(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } +} diff --git a/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightPreparedStatement.java b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightPreparedStatement.java new file mode 100644 index 00000000000..5ebe2f19d7c --- /dev/null +++ b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightPreparedStatement.java @@ -0,0 +1,323 @@ +/* + * 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.jdbc; + +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.ParameterMetaData; +import java.sql.Ref; +import java.sql.ResultSetMetaData; +import java.sql.RowId; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.SQLXML; +import java.sql.Time; +import java.sql.Timestamp; +import java.util.Calendar; + +/** + * PreparedStatement. + */ +public class FlightPreparedStatement extends FlightStatement implements java.sql.PreparedStatement { + + public FlightPreparedStatement(FlightConnection flightConnection) { + super(flightConnection); + } + + @Override + public FlightResultSet executeQuery() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int executeUpdate() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNull(int i, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBoolean(int i, boolean b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setByte(int i, byte b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setShort(int i, short i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setInt(int i, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setLong(int i, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setFloat(int i, float v) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setDouble(int i, double v) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBigDecimal(int i, BigDecimal bigDecimal) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setString(int i, String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBytes(int i, byte[] bytes) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setDate(int i, Date date) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setTime(int i, Time time) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setTimestamp(int i, Timestamp timestamp) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setAsciiStream(int i, InputStream inputStream, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setUnicodeStream(int i, InputStream inputStream, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBinaryStream(int i, InputStream inputStream, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void clearParameters() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setObject(int i, Object o, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setObject(int i, Object o) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean execute() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void addBatch() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setCharacterStream(int i, Reader reader, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setRef(int i, Ref ref) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBlob(int i, Blob blob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setClob(int i, Clob clob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setArray(int i, Array array) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public ResultSetMetaData getMetaData() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setDate(int i, Date date, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setTime(int i, Time time, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setTimestamp(int i, Timestamp timestamp, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNull(int i, int i1, String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setURL(int i, URL url) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public ParameterMetaData getParameterMetaData() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setRowId(int i, RowId rowId) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNString(int i, String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNCharacterStream(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNClob(int i, NClob nClob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setClob(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBlob(int i, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNClob(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setSQLXML(int i, SQLXML sqlxml) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setObject(int i, Object o, int i1, int i2) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setAsciiStream(int i, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBinaryStream(int i, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setCharacterStream(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setAsciiStream(int i, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBinaryStream(int i, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setCharacterStream(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNCharacterStream(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setClob(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setBlob(int i, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setNClob(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } +} diff --git a/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightResultSet.java b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightResultSet.java new file mode 100644 index 00000000000..f852d934d99 --- /dev/null +++ b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightResultSet.java @@ -0,0 +1,1062 @@ +/* + * 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.jdbc; + +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.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.RowId; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.SQLWarning; +import java.sql.SQLXML; +import java.sql.Statement; +import java.sql.Time; +import java.sql.Timestamp; +import java.util.Calendar; +import java.util.Map; + +import org.apache.arrow.flight.FlightStream; +import org.apache.arrow.vector.VectorSchemaRoot; + +/** + * ResultSet. + */ +public class FlightResultSet implements java.sql.ResultSet { + + /** Stream of RecordBatch instances. */ + private final FlightStream stream; + + /** The current record batch. */ + private VectorSchemaRoot root; + + /** Current row index into the current batch. */ + private int batchIndex; + + /** Current row index into the stream. */ + private int rowIndex = -1; + + /** Cache whether the last accessor method encountered a null value. */ + private boolean wasNull; + + /** + * Create a ResultSet to wrap a FlightStream. + */ + public FlightResultSet(final FlightStream stream) { + this.stream = stream; + + // fetch the first batch right away so we have meta-data + getNextBatch(); + } + + @Override + public boolean next() throws SQLException { + if (batchIndex + 1 == root.getRowCount()) { + if (getNextBatch()) { + batchIndex++; + rowIndex++; + return true; + } else { + return false; + } + } else { + batchIndex++; + rowIndex++; + return true; + } + } + + private boolean getNextBatch() { + batchIndex = -1; + if (stream.next()) { + root = stream.getRoot(); + return true; + } else { + root = null; + return false; + } + } + + @Override + public Object getObject(int columnIndex) throws SQLException { + final Object value = this.root.getFieldVectors().get(columnIndex - 1).getObject(batchIndex); + this.wasNull = value == null; + return value; + } + + @Override + public String getString(int i) throws SQLException { + return ResultSetHelper.getString(getObject(i)); + } + + @Override + public boolean getBoolean(int i) throws SQLException { + return ResultSetHelper.getBoolean(getObject(i)); + } + + @Override + public boolean wasNull() throws SQLException { + return wasNull; + } + + @Override + public void close() throws SQLException { + try { + stream.close(); + } catch (Exception e) { + throw new SQLException(e); + } + } + + @Override + public byte getByte(int i) throws SQLException { + return ResultSetHelper.getByte(getObject(i)); + } + + @Override + public short getShort(int i) throws SQLException { + return ResultSetHelper.getShort(getObject(i)); + } + + @Override + public int getInt(int i) throws SQLException { + return ResultSetHelper.getInt(getObject(i)); + } + + @Override + public long getLong(int i) throws SQLException { + return ResultSetHelper.getLong(getObject(i)); + } + + @Override + public float getFloat(int i) throws SQLException { + return ResultSetHelper.getFloat(getObject(i)); + } + + @Override + public double getDouble(int i) throws SQLException { + return ResultSetHelper.getDouble(getObject(i)); + } + + @Deprecated + @Override + public BigDecimal getBigDecimal(int i, int scale) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public byte[] getBytes(int i) throws SQLException { + return ResultSetHelper.getBytes(getObject(i)); + } + + @Override + public Date getDate(int i) throws SQLException { + return ResultSetHelper.getDate(getObject(i)); + } + + @Override + public Time getTime(int i) throws SQLException { + return ResultSetHelper.getTime(getObject(i)); + } + + @Override + public Timestamp getTimestamp(int i) throws SQLException { + return ResultSetHelper.getTimestamp(getObject(i)); + } + + @Override + public InputStream getAsciiStream(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public InputStream getUnicodeStream(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public InputStream getBinaryStream(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public String getString(String columnName) throws SQLException { + return ResultSetHelper.getString(getObject(columnName)); + } + + @Override + public boolean getBoolean(String columnName) throws SQLException { + return ResultSetHelper.getBoolean(getObject(columnName)); + } + + @Override + public byte getByte(String columnName) throws SQLException { + return ResultSetHelper.getByte(getObject(columnName)); + } + + @Override + public short getShort(String columnName) throws SQLException { + return ResultSetHelper.getShort(getObject(columnName)); + } + + @Override + public int getInt(String columnName) throws SQLException { + return ResultSetHelper.getInt(getObject(columnName)); + } + + @Override + public long getLong(String columnName) throws SQLException { + return ResultSetHelper.getLong(getObject(columnName)); + } + + @Override + public float getFloat(String columnName) throws SQLException { + return ResultSetHelper.getFloat(getObject(columnName)); + } + + @Override + public double getDouble(String columnName) throws SQLException { + return ResultSetHelper.getDouble(getObject(columnName)); + } + + @Deprecated + @Override + public BigDecimal getBigDecimal(String columnName, int scale) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public byte[] getBytes(String columnName) throws SQLException { + return ResultSetHelper.getBytes(getObject(columnName)); + } + + @Override + public Date getDate(String columnName) throws SQLException { + return ResultSetHelper.getDate(getObject(columnName)); + } + + @Override + public Time getTime(String columnName) throws SQLException { + return ResultSetHelper.getTime(getObject(columnName)); + } + + @Override + public Timestamp getTimestamp(String columnName) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public InputStream getAsciiStream(String columnName) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public InputStream getUnicodeStream(String columnName) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public InputStream getBinaryStream(String columnName) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public SQLWarning getWarnings() throws SQLException { + return null; + } + + @Override + public void clearWarnings() throws SQLException { + } + + @Override + public String getCursorName() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public ResultSetMetaData getMetaData() throws SQLException { + return new FlightResultSetMetaData(this.root.getFieldVectors()); + } + + @Override + public Object getObject(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int findColumn(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Reader getCharacterStream(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Reader getCharacterStream(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public BigDecimal getBigDecimal(int i) throws SQLException { + return ResultSetHelper.getBigDecimal(getObject(i)); + } + + @Override + public BigDecimal getBigDecimal(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isBeforeFirst() throws SQLException { + return rowIndex < 0; + } + + @Override + public boolean isAfterLast() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isFirst() throws SQLException { + return rowIndex == 0; + } + + @Override + public boolean isLast() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void beforeFirst() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void afterLast() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean first() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean last() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getRow() throws SQLException { + return rowIndex; + } + + @Override + public boolean absolute(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean relative(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean previous() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getFetchDirection() throws SQLException { + return ResultSet.FETCH_FORWARD; + } + + @Override + public void setFetchDirection(int fetchDirection) throws SQLException { + if (fetchDirection != ResultSet.FETCH_FORWARD) { + throw new SQLFeatureNotSupportedException(); + } + } + + @Override + public int getFetchSize() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setFetchSize(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getType() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getConcurrency() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean rowUpdated() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean rowInserted() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean rowDeleted() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNull(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBoolean(int i, boolean b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateByte(int i, byte b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateShort(int i, short i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateInt(int i, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateLong(int i, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateFloat(int i, float v) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateDouble(int i, double v) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBigDecimal(int i, BigDecimal bigDecimal) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateString(int i, String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBytes(int i, byte[] bytes) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateDate(int i, Date date) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateTime(int i, Time time) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateTimestamp(int i, Timestamp timestamp) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateAsciiStream(int i, InputStream inputStream, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBinaryStream(int i, InputStream inputStream, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateCharacterStream(int i, Reader reader, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateObject(int i, Object o, int i1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateObject(int i, Object o) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNull(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBoolean(String s, boolean b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateByte(String s, byte b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateShort(String s, short i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateInt(String s, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateLong(String s, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateFloat(String s, float v) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateDouble(String s, double v) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBigDecimal(String s, BigDecimal bigDecimal) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateString(String s, String s1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBytes(String s, byte[] bytes) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateDate(String s, Date date) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateTime(String s, Time time) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateTimestamp(String s, Timestamp timestamp) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateAsciiStream(String s, InputStream inputStream, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBinaryStream(String s, InputStream inputStream, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateCharacterStream(String s, Reader reader, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateObject(String s, Object o, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateObject(String s, Object o) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void insertRow() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateRow() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void deleteRow() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void refreshRow() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void cancelRowUpdates() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void moveToInsertRow() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void moveToCurrentRow() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Statement getStatement() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Object getObject(int i, Map> map) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Ref getRef(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Blob getBlob(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Clob getClob(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Array getArray(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Object getObject(String s, Map> map) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Ref getRef(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Blob getBlob(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Clob getClob(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Array getArray(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Date getDate(int i, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Date getDate(String s, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Time getTime(int i, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Time getTime(String s, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Timestamp getTimestamp(int i, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Timestamp getTimestamp(String s, Calendar calendar) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public URL getURL(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public URL getURL(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateRef(int i, Ref ref) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateRef(String s, Ref ref) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBlob(int i, Blob blob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBlob(String s, Blob blob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateClob(int i, Clob clob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateClob(String s, Clob clob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateArray(int i, Array array) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateArray(String s, Array array) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public RowId getRowId(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public RowId getRowId(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateRowId(int i, RowId rowId) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateRowId(String s, RowId rowId) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getHoldability() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isClosed() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNString(int i, String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNString(String s, String s1) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNClob(int i, NClob nClob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNClob(String s, NClob nClob) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public NClob getNClob(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public NClob getNClob(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public SQLXML getSQLXML(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public SQLXML getSQLXML(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateSQLXML(int i, SQLXML sqlxml) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateSQLXML(String s, SQLXML sqlxml) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public String getNString(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public String getNString(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Reader getNCharacterStream(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Reader getNCharacterStream(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNCharacterStream(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNCharacterStream(String s, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateAsciiStream(int i, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBinaryStream(int i, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateCharacterStream(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateAsciiStream(String s, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBinaryStream(String s, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateCharacterStream(String s, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBlob(int i, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBlob(String s, InputStream inputStream, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateClob(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateClob(String s, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNClob(int i, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNClob(String s, Reader reader, long l) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNCharacterStream(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNCharacterStream(String s, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateAsciiStream(int i, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBinaryStream(int i, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateCharacterStream(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateAsciiStream(String s, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBinaryStream(String s, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateCharacterStream(String s, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBlob(int i, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateBlob(String s, InputStream inputStream) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateClob(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateClob(String s, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNClob(int i, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void updateNClob(String s, Reader reader) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public T getObject(int i, Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public T getObject(String s, Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public T unwrap(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isWrapperFor(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } +} diff --git a/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightResultSetMetaData.java b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightResultSetMetaData.java new file mode 100644 index 00000000000..f3c1d7cfa06 --- /dev/null +++ b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightResultSetMetaData.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.jdbc; + +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.Types; +import java.util.List; + +import org.apache.arrow.vector.FieldVector; + +/** ResultSetMetaData. */ +public class FlightResultSetMetaData implements ResultSetMetaData { + + private final List fields; + + public FlightResultSetMetaData(final List fields) { + this.fields = fields; + } + + @Override + public int getColumnCount() throws SQLException { + return fields.size(); + } + + @Override + public boolean isAutoIncrement(int i) throws SQLException { + return false; + } + + @Override + public boolean isCaseSensitive(int i) throws SQLException { + return false; + } + + @Override + public boolean isSearchable(int i) throws SQLException { + return false; + } + + @Override + public boolean isCurrency(int i) throws SQLException { + return false; + } + + @Override + public int isNullable(int i) throws SQLException { + return ResultSetMetaData.columnNullable; + } + + @Override + public boolean isSigned(int i) throws SQLException { + return true; + } + + @Override + public int getColumnDisplaySize(int i) throws SQLException { + return 0; + } + + @Override + public String getColumnLabel(int i) throws SQLException { + return null; + } + + @Override + public String getColumnName(int i) throws SQLException { + return fields.get(i - 1).getName(); + } + + @Override + public String getSchemaName(int i) throws SQLException { + return null; + } + + @Override + public int getPrecision(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getScale(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public String getTableName(int i) throws SQLException { + return null; + } + + @Override + public String getCatalogName(int i) throws SQLException { + return null; + } + + @Override + public int getColumnType(int i) throws SQLException { + switch (fields.get(i - 1).getMinorType()) { + case INT: + return Types.INTEGER; + default: + return Types.OTHER; + } + } + + @Override + public String getColumnTypeName(int i) throws SQLException { + return String.valueOf(fields.get(i - 1).getMinorType()); + } + + @Override + public boolean isReadOnly(int i) throws SQLException { + return false; + } + + @Override + public boolean isWritable(int i) throws SQLException { + return false; + } + + @Override + public boolean isDefinitelyWritable(int i) throws SQLException { + return false; + } + + @Override + public String getColumnClassName(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public T unwrap(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isWrapperFor(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } +} diff --git a/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightStatement.java b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightStatement.java new file mode 100644 index 00000000000..3f62d04786a --- /dev/null +++ b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/FlightStatement.java @@ -0,0 +1,275 @@ +/* + * 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.jdbc; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.SQLWarning; +import java.util.concurrent.TimeUnit; + +import org.apache.arrow.flight.CallOption; +import org.apache.arrow.flight.CallOptions; +import org.apache.arrow.flight.FlightClient; +import org.apache.arrow.flight.FlightStream; +import org.apache.arrow.flight.Location; +import org.apache.arrow.flight.Ticket; +import org.apache.arrow.memory.RootAllocator; + +/** + * Statement. + */ +public class FlightStatement implements java.sql.Statement { + + protected final FlightConnection flightConnection; + + public FlightStatement(FlightConnection flightConnection) { + this.flightConnection = flightConnection; + } + + @Override + public ResultSet executeQuery(String query) throws SQLException { + + FlightClient client = FlightClient.builder() + .allocator(new RootAllocator(Long.MAX_VALUE)) + .location(Location.forGrpcInsecure(flightConnection.host, flightConnection.port)) + .build(); + + CallOption callOptions = CallOptions.timeout(5, TimeUnit.SECONDS); + + Ticket ticket = new Ticket(query.getBytes()); + + FlightStream stream = client.getStream(ticket, callOptions); + + return new FlightResultSet(stream); + } + + @Override + public int executeUpdate(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void close() throws SQLException { + } + + @Override + public int getMaxFieldSize() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setMaxFieldSize(int i) throws SQLException { + + } + + @Override + public int getMaxRows() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setMaxRows(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setEscapeProcessing(boolean b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getQueryTimeout() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setQueryTimeout(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void cancel() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public SQLWarning getWarnings() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void clearWarnings() throws SQLException { + } + + @Override + public void setCursorName(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean execute(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public ResultSet getResultSet() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getUpdateCount() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean getMoreResults() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getFetchDirection() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setFetchDirection(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getFetchSize() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setFetchSize(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getResultSetConcurrency() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getResultSetType() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void addBatch(String s) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void clearBatch() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int[] executeBatch() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public Connection getConnection() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean getMoreResults(int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public ResultSet getGeneratedKeys() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int executeUpdate(String s, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int executeUpdate(String s, int[] ints) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int executeUpdate(String s, String[] strings) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean execute(String s, int i) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean execute(String s, int[] ints) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean execute(String s, String[] strings) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public int getResultSetHoldability() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isClosed() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isPoolable() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void setPoolable(boolean b) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public void closeOnCompletion() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isCloseOnCompletion() throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public T unwrap(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + @Override + public boolean isWrapperFor(Class aClass) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } +} diff --git a/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/ResultSetHelper.java b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/ResultSetHelper.java new file mode 100644 index 00000000000..220decb95a5 --- /dev/null +++ b/java/flight/flight-jdbc/src/main/java/org/apache/arrow/jdbc/ResultSetHelper.java @@ -0,0 +1,161 @@ +/* + * 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.jdbc; + +import java.math.BigDecimal; +import java.sql.Date; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.sql.Time; +import java.sql.Timestamp; + +/** + * Helper methods for converting between data types. + */ +public class ResultSetHelper { + + /** Convert value to String. */ + public static String getString(final Object value) throws SQLException { + return value == null ? null : String.valueOf(value); + } + + /** Convert value to boolean. */ + public static boolean getBoolean(final Object value) throws SQLException { + if (value == null) { + return false; + } else if (value instanceof Boolean) { + return (Boolean) value; + } else if (value instanceof String) { + return ((String) value).equalsIgnoreCase("true"); + } else { + throw unsupportedConversion("boolean", value); + } + } + + /** Convert value to byte. */ + public static byte getByte(final Object value) throws SQLException { + if (value == null) { + return 0; + } else if (value instanceof Number) { + return ((Number) value).byteValue(); + } else if (value instanceof String) { + return Byte.parseByte((String) value); + } else { + throw unsupportedConversion("byte", value); + } + } + + /** Convert value to short. */ + public static short getShort(final Object value) throws SQLException { + if (value == null) { + return 0; + } else if (value instanceof Number) { + return ((Number) value).shortValue(); + } else if (value instanceof String) { + return Short.parseShort((String) value); + } else { + throw unsupportedConversion("short", value); + } + } + + /** Convert value to int. */ + public static int getInt(final Object value) throws SQLException { + if (value == null) { + return 0; + } else if (value instanceof Number) { + return ((Number) value).intValue(); + } else if (value instanceof String) { + return Integer.parseInt((String) value); + } else { + throw unsupportedConversion("int", value); + } + } + + /** Convert value to String. */ + public static long getLong(final Object value) throws SQLException { + if (value == null) { + return 0; + } else if (value instanceof Number) { + return ((Number) value).longValue(); + } else if (value instanceof String) { + return Long.parseLong((String) value); + } else { + throw unsupportedConversion("long", value); + } + } + + /** Convert value to float. */ + public static float getFloat(final Object value) throws SQLException { + if (value == null) { + return 0; + } else if (value instanceof Number) { + return ((Number) value).floatValue(); + } else if (value instanceof String) { + return Float.parseFloat((String) value); + } else { + throw unsupportedConversion("float", value); + } + } + + /** Convert value to double. */ + public static double getDouble(final Object value) throws SQLException { + if (value == null) { + return 0; + } else if (value instanceof Number) { + return ((Number) value).doubleValue(); + } else if (value instanceof String) { + return Double.parseDouble((String) value); + } else { + throw unsupportedConversion("double", value); + } + } + + /** Convert value to BigDecimal. */ + public static BigDecimal getBigDecimal(final Object value) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + /** Convert value to byte[]. */ + public static byte[] getBytes(final Object value) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + /** Convert value to Date. */ + public static Date getDate(final Object value) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + /** Convert value to Time. */ + public static Time getTime(final Object value) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + /** Convert value to Timestamp. */ + public static Timestamp getTimestamp(final Object value) throws SQLException { + throw new SQLFeatureNotSupportedException(); + } + + /** Convenience method for building an exception for unsupported conversions. */ + private static SQLException unsupportedConversion(String t, Object value) { + if (value == null) { + return new SQLException(String.format("Cannot convert null value to type %s", t)); + } else { + return new SQLException(String.format("Cannot convert %s value '%s' to type %s", value.getClass(), value, t)); + } + } +} diff --git a/java/flight/flight-jdbc/src/main/resources/META-INF/services/java.sql.Driver b/java/flight/flight-jdbc/src/main/resources/META-INF/services/java.sql.Driver new file mode 100644 index 00000000000..df3a575ba1d --- /dev/null +++ b/java/flight/flight-jdbc/src/main/resources/META-INF/services/java.sql.Driver @@ -0,0 +1 @@ +org.apache.arrow.jdbc.Driver \ No newline at end of file diff --git a/java/flight/flight-jdbc/src/test/java/org/apache/arrow/jdbc/DriverTest.java b/java/flight/flight-jdbc/src/test/java/org/apache/arrow/jdbc/DriverTest.java new file mode 100644 index 00000000000..904c3920f58 --- /dev/null +++ b/java/flight/flight-jdbc/src/test/java/org/apache/arrow/jdbc/DriverTest.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.arrow.jdbc; + +import static org.junit.Assert.*; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; +import java.sql.Types; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +import org.junit.Ignore; +import org.junit.Test; + +import com.google.common.collect.ImmutableList; + +/** + * JDBC Driver unit tests. + */ +public class DriverTest { + + final Driver driver = new org.apache.arrow.jdbc.Driver(); + + @Test + public void acceptsValidUrl() throws SQLException { + assertTrue(driver.acceptsURL("jdbc:arrow://localhost:50051")); + } + + @Test + public void rejectsInvalidUrl() throws SQLException { + assertFalse(driver.acceptsURL("jdbc:mysql://localhost:50051")); + } + + @Test + public void rejectsNullUrl() throws SQLException { + assertFalse(driver.acceptsURL(null)); + } + + /** + * Note that this is a manual integration test that requires the Rust flight-server example to be running. + */ + @Test + @Ignore + public void executeQuery() throws SQLException { + try (Connection conn = DriverManager.getConnection("jdbc:arrow://localhost:50051", new Properties())) { + try (Statement stmt = conn.createStatement()) { + try (ResultSet rs = stmt.executeQuery("SELECT id FROM alltypes_plain")) { + + ResultSetMetaData md = rs.getMetaData(); + assertEquals(1, md.getColumnCount()); + assertEquals("c0", md.getColumnName(1)); + assertEquals(Types.INTEGER, md.getColumnType(1)); + + List ids = new ArrayList<>(); + while (rs.next()) { + ids.add(rs.getInt(1)); + } + assertEquals(ImmutableList.of(4, 5, 6, 7, 2, 3, 0, 1), ids); + } + } + } + } +} diff --git a/java/flight/flight-jdbc/src/test/java/org/apache/arrow/jdbc/ResultSetHelperTest.java b/java/flight/flight-jdbc/src/test/java/org/apache/arrow/jdbc/ResultSetHelperTest.java new file mode 100644 index 00000000000..22c71eaadd5 --- /dev/null +++ b/java/flight/flight-jdbc/src/test/java/org/apache/arrow/jdbc/ResultSetHelperTest.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.arrow.jdbc; + +import static org.junit.Assert.*; + +import java.sql.SQLException; + +import org.junit.Test; + +/** + * JDBC Driver unit tests. + */ +public class ResultSetHelperTest { + + //TODO add exhaustive tests based on suggested conversions in JDBC specification + + @Test + public void getString() throws SQLException { + assertNull(ResultSetHelper.getString(null)); + assertEquals("a", ResultSetHelper.getString("a")); + assertEquals("123", ResultSetHelper.getString(123)); + } + + @Test + public void getInt() throws SQLException { + assertEquals(0, ResultSetHelper.getInt(null)); + assertEquals(123, ResultSetHelper.getInt("123")); + assertEquals(123, ResultSetHelper.getInt(123)); + } + +} diff --git a/java/pom.xml b/java/pom.xml index 8397ef29e7d..4aadde92c23 100644 --- a/java/pom.xml +++ b/java/pom.xml @@ -137,6 +137,7 @@ **/client/build/** **/*.tbl **/*.iml + **/java.sql.Driver @@ -682,6 +683,7 @@ plasma flight/flight-core flight/flight-grpc + flight/flight-jdbc performance algorithm adapter/avro