You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2013/11/18 09:56:46 UTC
[1/3] TAJO-176: Implement Tajo JDBC Driver. (Keuntae Park via jihoon)
Updated Branches:
refs/heads/master 6e2db3baf -> 342fd47ff
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetBase.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetBase.java
new file mode 100644
index 0000000..f4d685f
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetBase.java
@@ -0,0 +1,1129 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.net.URL;
+import java.sql.*;
+import java.util.Calendar;
+import java.util.Map;
+
+public abstract class TajoResultSetBase implements ResultSet {
+ protected int curRow;
+ protected long totalRow;
+ protected boolean wasNull;
+ protected Schema schema;
+ protected Tuple cur;
+
+ protected void init() {
+ cur = null;
+ curRow = 0;
+ totalRow = 0;
+ wasNull = false;
+ }
+
+ private void handleNull(Datum d) {
+ wasNull = (d instanceof NullDatum);
+ }
+
+ @Override
+ public void beforeFirst() throws SQLException {
+ init();
+ }
+
+ @Override
+ public boolean getBoolean(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asBool();
+ }
+
+ @Override
+ public boolean getBoolean(String colName) throws SQLException {
+ Datum datum = cur.get(findColumn(colName));
+ handleNull(datum);
+ return datum.asBool();
+ }
+
+ @Override
+ public byte getByte(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asByte();
+ }
+
+ @Override
+ public byte getByte(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asByte();
+ }
+
+ @Override
+ public byte[] getBytes(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asByteArray();
+ }
+
+ @Override
+ public byte[] getBytes(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asByteArray();
+ }
+
+ @Override
+ public double getDouble(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asFloat8();
+ }
+
+ @Override
+ public double getDouble(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asFloat8();
+ }
+
+ @Override
+ public float getFloat(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asFloat4();
+ }
+
+ @Override
+ public float getFloat(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asFloat4();
+ }
+
+ @Override
+ public int getInt(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asInt4();
+ }
+
+ @Override
+ public int getInt(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asInt4();
+ }
+
+ @Override
+ public long getLong(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asInt8();
+ }
+
+ @Override
+ public long getLong(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asInt8();
+ }
+
+ @Override
+ public Object getObject(int fieldId) throws SQLException {
+ Datum d = cur.get(fieldId - 1);
+ handleNull(d);
+
+ TajoDataTypes.Type dataType = schema.getColumn(fieldId - 1).getDataType().getType();
+
+ switch(dataType) {
+ case BOOLEAN: return d.asBool();
+ case INT1:
+ case INT2: return d.asInt2();
+ case INT4: return d.asInt4();
+ case INT8: return d.asInt8();
+ case TEXT:
+ case CHAR:
+ case DATE:
+ case VARCHAR: return d.asChars();
+ case FLOAT4: return d.asFloat4();
+ case FLOAT8: return d.asFloat8();
+ case DECIMAL:
+ case NUMERIC: return d.asFloat8();
+ default: return d.asChars();
+ }
+ }
+
+ @Override
+ public Object getObject(String name) throws SQLException {
+ return getObject(findColumn(name));
+ }
+
+ @Override
+ public short getShort(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asInt2();
+ }
+
+ @Override
+ public short getShort(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asInt2();
+ }
+
+ @Override
+ public String getString(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ handleNull(datum);
+ return datum.asChars();
+ }
+
+ @Override
+ public String getString(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ handleNull(datum);
+ return datum.asChars();
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> clazz) throws SQLException {
+ throw new SQLFeatureNotSupportedException("isWrapperFor not supported");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> clazz) throws SQLException {
+ throw new SQLFeatureNotSupportedException("unwrap not supported");
+ }
+
+ @Override
+ public boolean absolute(int row) throws SQLException {
+ throw new SQLFeatureNotSupportedException("absolute not supported");
+ }
+
+ @Override
+ public void afterLast() throws SQLException {
+ while (this.next())
+ ;
+ }
+
+ @Override
+ public void cancelRowUpdates() throws SQLException {
+ throw new SQLFeatureNotSupportedException("cancelRowUpdates not supported");
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ throw new SQLFeatureNotSupportedException("clearWarnings not supported");
+ }
+
+ @Override
+ public void deleteRow() throws SQLException {
+ throw new SQLFeatureNotSupportedException("deleteRow not supported");
+ }
+
+ @Override
+ public int findColumn(String colName) throws SQLException {
+ return schema.getColumnIdByName(colName);
+ }
+
+ @Override
+ public boolean first() throws SQLException {
+ this.beforeFirst();
+ return this.next();
+ }
+
+ @Override
+ public Array getArray(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getArray not supported");
+ }
+
+ @Override
+ public Array getArray(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getArray not supported");
+ }
+
+ @Override
+ public InputStream getAsciiStream(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getAsciiStream not supported");
+ }
+
+ @Override
+ public InputStream getAsciiStream(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getAsciiStream not supported");
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBigDecimal not supported");
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBigDecimal not supported");
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(int index, int x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBigDecimal not supported");
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String name, int x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBigDecimal not supported");
+ }
+
+ @Override
+ public InputStream getBinaryStream(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBinaryStream not supported");
+ }
+
+ @Override
+ public InputStream getBinaryStream(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBinaryStream not supported");
+ }
+
+ @Override
+ public Blob getBlob(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBlob not supported");
+ }
+
+ @Override
+ public Blob getBlob(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getBlob not supported");
+ }
+
+ @Override
+ public Reader getCharacterStream(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getCharacterStream not supported");
+ }
+
+ @Override
+ public Reader getCharacterStream(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getCharacterStream not supported");
+ }
+
+ @Override
+ public Clob getClob(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getClob not supported");
+ }
+
+ @Override
+ public Clob getClob(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getClob not supported");
+ }
+
+ @Override
+ public int getConcurrency() throws SQLException {
+ return ResultSet.CONCUR_READ_ONLY;
+ }
+
+ @Override
+ public String getCursorName() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getCursorName not supported");
+ }
+
+ @Override
+ public Date getDate(int index) throws SQLException {
+ Object obj = getObject(index);
+ if (obj == null) {
+ return null;
+ }
+
+ try {
+ return Date.valueOf((String) obj);
+ } catch (Exception e) {
+ throw new SQLException("Cannot convert column " + index
+ + " to date: " + e.toString());
+ }
+ }
+
+ @Override
+ public Date getDate(String name) throws SQLException {
+ return getDate(findColumn(name));
+ }
+
+ @Override
+ public Date getDate(int index, Calendar x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getDate not supported");
+ }
+
+ @Override
+ public Date getDate(String name, Calendar x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getDate not supported");
+ }
+
+ @Override
+ public int getFetchDirection() throws SQLException {
+ return ResultSet.FETCH_FORWARD;
+ }
+
+ @Override
+ public int getFetchSize() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getFetchSize not supported");
+ }
+
+ @Override
+ public int getHoldability() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getHoldability not supported");
+ }
+
+ @Override
+ public ResultSetMetaData getMetaData() throws SQLException {
+ return new TajoResultSetMetaData(schema);
+ }
+
+ @Override
+ public Reader getNCharacterStream(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getNCharacterStream not supported");
+ }
+
+ @Override
+ public Reader getNCharacterStream(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getNCharacterStream not supported");
+ }
+
+ @Override
+ public NClob getNClob(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getNClob not supported");
+ }
+
+ @Override
+ public NClob getNClob(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getNClob not supported");
+ }
+
+ @Override
+ public String getNString(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getNString not supported");
+ }
+
+ @Override
+ public String getNString(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getNString not supported");
+ }
+
+ @Override
+ public Object getObject(int index, Map<String, Class<?>> x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getObject not supported");
+ }
+
+ @Override
+ public Object getObject(String name, Map<String, Class<?>> x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getObject not supported");
+ }
+
+ public <T> T getObject(String name, Class<T> x)
+ throws SQLException {
+ //JDK 1.7
+ throw new SQLFeatureNotSupportedException("getObject not supported");
+ }
+
+ public <T> T getObject(int index, Class<T> x)
+ throws SQLException {
+ //JDK 1.7
+ throw new SQLFeatureNotSupportedException("getObject not supported");
+ }
+
+ @Override
+ public Ref getRef(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getRef not supported");
+ }
+
+ @Override
+ public Ref getRef(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getRef not supported");
+ }
+
+ @Override
+ public int getRow() throws SQLException {
+ return curRow;
+ }
+
+ @Override
+ public RowId getRowId(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getRowId not supported");
+ }
+
+ @Override
+ public RowId getRowId(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getRowId not supported");
+ }
+
+ @Override
+ public SQLXML getSQLXML(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getSQLXML not supported");
+ }
+
+ @Override
+ public SQLXML getSQLXML(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getSQLXML not supported");
+ }
+
+ @Override
+ public Statement getStatement() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getStatement not supported");
+ }
+
+ @Override
+ public Time getTime(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTime not supported");
+ }
+
+ @Override
+ public Time getTime(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTime not supported");
+ }
+
+ @Override
+ public Time getTime(int index, Calendar x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTime not supported");
+ }
+
+ @Override
+ public Time getTime(String name, Calendar x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTime not supported");
+ }
+
+ @Override
+ public Timestamp getTimestamp(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTimestamp not supported");
+ }
+
+ @Override
+ public Timestamp getTimestamp(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTimestamp not supported");
+ }
+
+ @Override
+ public Timestamp getTimestamp(int index, Calendar x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTimestamp not supported");
+ }
+
+ @Override
+ public Timestamp getTimestamp(String name, Calendar x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTimestamp not supported");
+ }
+
+ @Override
+ public int getType() throws SQLException {
+ return ResultSet.TYPE_FORWARD_ONLY;
+ }
+
+ @Override
+ public URL getURL(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getURL not supported");
+ }
+
+ @Override
+ public URL getURL(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getURL not supported");
+ }
+
+ @Override
+ public InputStream getUnicodeStream(int index) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getUnicodeStream not supported");
+ }
+
+ @Override
+ public InputStream getUnicodeStream(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getUnicodeStream not supported");
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getWarnings not supported");
+ }
+
+ @Override
+ public void insertRow() throws SQLException {
+ throw new SQLFeatureNotSupportedException("insertRow not supported");
+ }
+
+ @Override
+ public boolean isAfterLast() throws SQLException {
+ return this.curRow > this.totalRow;
+ }
+
+ @Override
+ public boolean isBeforeFirst() throws SQLException {
+ return this.curRow == 0;
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return this.curRow == -1;
+ }
+
+ @Override
+ public boolean isFirst() throws SQLException {
+ return this.curRow == 1;
+ }
+
+ @Override
+ public boolean isLast() throws SQLException {
+ return this.curRow == this.totalRow;
+ }
+
+ @Override
+ public boolean last() throws SQLException {
+ Tuple last = null;
+ while (this.next()) {
+ last = cur;
+ }
+ cur = last;
+ return true;
+ }
+
+ @Override
+ public void moveToCurrentRow() throws SQLException {
+ throw new SQLFeatureNotSupportedException("moveToCurrentRow not supported");
+ }
+
+ @Override
+ public void moveToInsertRow() throws SQLException {
+ throw new SQLFeatureNotSupportedException("moveToInsertRow not supported");
+ }
+
+ @Override
+ public boolean next() throws SQLException {
+ try {
+ if (totalRow <= 0) {
+ return false;
+ }
+
+ cur = nextTuple();
+ curRow++;
+ if (cur != null) {
+ return true;
+ }
+ } catch (IOException e) {
+ throw new SQLException(e.getMessage());
+ }
+ return false;
+ }
+
+ protected abstract Tuple nextTuple() throws IOException;
+
+ @Override
+ public boolean previous() throws SQLException {
+ throw new SQLFeatureNotSupportedException("previous not supported");
+ }
+
+ @Override
+ public void refreshRow() throws SQLException {
+ throw new SQLFeatureNotSupportedException("refreshRow not supported");
+ }
+
+ @Override
+ public boolean relative(int rows) throws SQLException {
+ throw new SQLFeatureNotSupportedException("relative not supported");
+ }
+
+ @Override
+ public boolean rowDeleted() throws SQLException {
+ throw new SQLFeatureNotSupportedException("rowDeleted not supported");
+ }
+
+ @Override
+ public boolean rowInserted() throws SQLException {
+ throw new SQLFeatureNotSupportedException("rowInserted not supported");
+ }
+
+ @Override
+ public boolean rowUpdated() throws SQLException {
+ throw new SQLFeatureNotSupportedException("rowUpdated not supported");
+ }
+
+ @Override
+ public void setFetchDirection(int direction) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setFetchDirection not supported");
+ }
+
+ @Override
+ public void setFetchSize(int size) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setFetchSize not supported");
+ }
+
+ @Override
+ public void updateArray(int index, Array x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateArray not supported");
+ }
+
+ @Override
+ public void updateArray(String name, Array x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateArray not supported");
+ }
+
+ @Override
+ public void updateAsciiStream(int index, InputStream x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateAsciiStream not supported");
+ }
+
+ @Override
+ public void updateAsciiStream(String name, InputStream x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateAsciiStream not supported");
+ }
+
+ @Override
+ public void updateAsciiStream(int index, InputStream x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateAsciiStream not supported");
+ }
+
+ @Override
+ public void updateAsciiStream(String name, InputStream x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateAsciiStream not supported");
+ }
+
+ @Override
+ public void updateAsciiStream(int index, InputStream x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateAsciiStream not supported");
+ }
+
+ @Override
+ public void updateAsciiStream(String name, InputStream x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateAsciiStream not supported");
+ }
+
+ @Override
+ public void updateBigDecimal(int index, BigDecimal x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBigDecimal not supported");
+ }
+
+ @Override
+ public void updateBigDecimal(String name, BigDecimal x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBigDecimal not supported");
+ }
+
+ @Override
+ public void updateBinaryStream(int index, InputStream x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBinaryStream not supported");
+ }
+
+ @Override
+ public void updateBinaryStream(String name, InputStream x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBinaryStream not supported");
+ }
+
+ @Override
+ public void updateBinaryStream(int index, InputStream x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBinaryStream not supported");
+ }
+
+ @Override
+ public void updateBinaryStream(String name, InputStream x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBinaryStream not supported");
+ }
+
+ @Override
+ public void updateBinaryStream(int index, InputStream x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBinaryStream not supported");
+ }
+
+ @Override
+ public void updateBinaryStream(String name, InputStream x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBinaryStream not supported");
+ }
+
+ @Override
+ public void updateBlob(int index, Blob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBlob not supported");
+ }
+
+ @Override
+ public void updateBlob(String name, Blob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBlob not supported");
+ }
+
+ @Override
+ public void updateBlob(int index, InputStream x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBlob not supported");
+ }
+
+ @Override
+ public void updateBlob(String name, InputStream x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBlob not supported");
+ }
+
+ @Override
+ public void updateBlob(int index, InputStream x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBlob not supported");
+ }
+
+ @Override
+ public void updateBlob(String name, InputStream x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBlob not supported");
+ }
+
+ @Override
+ public void updateBoolean(int index, boolean x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBoolean not supported");
+ }
+
+ @Override
+ public void updateBoolean(String name, boolean x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateBoolean not supported");
+ }
+
+ @Override
+ public void updateByte(int index, byte x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateByte not supported");
+ }
+
+ @Override
+ public void updateByte(String name, byte x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateByte not supported");
+ }
+
+ @Override
+ public void updateBytes(int index, byte[] x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateByte not supported");
+ }
+
+ @Override
+ public void updateBytes(String name, byte[] x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateByte not supported");
+ }
+
+ @Override
+ public void updateCharacterStream(int index, Reader x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateCharacterStream not supported");
+ }
+
+ @Override
+ public void updateCharacterStream(String name, Reader x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateCharacterStream not supported");
+ }
+
+ @Override
+ public void updateCharacterStream(int index, Reader x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateCharacterStream not supported");
+ }
+
+ @Override
+ public void updateCharacterStream(String name, Reader x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateCharacterStream not supported");
+ }
+
+ @Override
+ public void updateCharacterStream(int index, Reader x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateCharacterStream not supported");
+ }
+
+ @Override
+ public void updateCharacterStream(String name, Reader x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateCharacterStream not supported");
+ }
+
+ @Override
+ public void updateClob(int index, Clob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateClob not supported");
+ }
+
+ @Override
+ public void updateClob(String name, Clob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateClob not supported");
+ }
+
+ @Override
+ public void updateClob(int index, Reader x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateClob not supported");
+ }
+
+ @Override
+ public void updateClob(String name, Reader x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateClob not supported");
+ }
+
+ @Override
+ public void updateClob(int index, Reader x, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateClob not supported");
+ }
+
+ @Override
+ public void updateClob(String name, Reader x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateClob not supported");
+ }
+
+ @Override
+ public void updateDate(int index, Date x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateDate not supported");
+ }
+
+ @Override
+ public void updateDate(String name, Date x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateDate not supported");
+ }
+
+ @Override
+ public void updateDouble(int index, double x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateDouble not supported");
+ }
+
+ @Override
+ public void updateDouble(String name, double x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateDouble not supported");
+ }
+
+ @Override
+ public void updateFloat(int index, float x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateFloat not supported");
+ }
+
+ @Override
+ public void updateFloat(String name, float x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateFloat not supported");
+ }
+
+ @Override
+ public void updateInt(int index, int x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateInt not supported");
+ }
+
+ @Override
+ public void updateInt(String name, int x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateInt not supported");
+ }
+
+ @Override
+ public void updateLong(int index, long x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateLong not supported");
+ }
+
+ @Override
+ public void updateLong(String name, long x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateLong not supported");
+ }
+
+ @Override
+ public void updateNCharacterStream(int index, Reader x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNCharacterStream not supported");
+ }
+
+ @Override
+ public void updateNCharacterStream(String name, Reader x)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNCharacterStream not supported");
+ }
+
+ @Override
+ public void updateNCharacterStream(int index, Reader x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNCharacterStream not supported");
+ }
+
+ @Override
+ public void updateNCharacterStream(String name, Reader x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNCharacterStream not supported");
+ }
+
+ @Override
+ public void updateNClob(int index, NClob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNClob not supported");
+ }
+
+ @Override
+ public void updateNClob(String name, NClob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNClob not supported");
+ }
+
+ @Override
+ public void updateNClob(int index, Reader x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNClob not supported");
+ }
+
+ @Override
+ public void updateNClob(String name, Reader x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNClob not supported");
+ }
+
+ @Override
+ public void updateNClob(int index, Reader x, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNClob not supported");
+ }
+
+ @Override
+ public void updateNClob(String name, Reader x, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNClob not supported");
+ }
+
+ @Override
+ public void updateNString(int arg0, String x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNString not supported");
+ }
+
+ @Override
+ public void updateNString(String name, String x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNString not supported");
+ }
+
+ @Override
+ public void updateNull(int arg0) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNull not supported");
+ }
+
+ @Override
+ public void updateNull(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateNull not supported");
+ }
+
+ @Override
+ public void updateObject(int index, Object x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateObject not supported");
+ }
+
+ @Override
+ public void updateObject(String name, Object x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateObject not supported");
+ }
+
+ @Override
+ public void updateObject(int index, Object x, int length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateObject not supported");
+ }
+
+ @Override
+ public void updateObject(String name, Object x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateObject not supported");
+ }
+
+ @Override
+ public void updateRef(int index, Ref x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateRef not supported");
+ }
+
+ @Override
+ public void updateRef(String name, Ref x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateRef not supported");
+ }
+
+ @Override
+ public void updateRow() throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateRow not supported");
+ }
+
+ @Override
+ public void updateRowId(int index, RowId arg1) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateRowId not supported");
+ }
+
+ @Override
+ public void updateRowId(String name, RowId x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateRowId not supported");
+ }
+
+ @Override
+ public void updateSQLXML(int index, SQLXML arg1) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateSQLXML not supported");
+ }
+
+ @Override
+ public void updateSQLXML(String name, SQLXML x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateSQLXML not supported");
+
+ }
+
+ @Override
+ public void updateShort(int index, short x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateShort not supported");
+
+ }
+
+ @Override
+ public void updateShort(String name, short x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateShort not supported");
+
+ }
+
+ @Override
+ public void updateString(int index, String x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateString not supported");
+
+ }
+
+ @Override
+ public void updateString(String name, String arg1) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateString not supported");
+
+ }
+
+ @Override
+ public void updateTime(int index, Time x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateTime not supported");
+
+ }
+
+ @Override
+ public void updateTime(String name, Time x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateTime not supported");
+
+ }
+
+ @Override
+ public void updateTimestamp(int index, Timestamp x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateTimestamp not supported");
+
+ }
+
+ @Override
+ public void updateTimestamp(String name, Timestamp x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("updateTimestamp not supported");
+
+ }
+
+ @Override
+ public boolean wasNull() throws SQLException {
+ return wasNull;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetMetaData.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetMetaData.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetMetaData.java
new file mode 100644
index 0000000..5a04ad0
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSetMetaData.java
@@ -0,0 +1,160 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.common.TajoDataTypes.DataType;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+
+public class TajoResultSetMetaData implements ResultSetMetaData {
+ Schema schema;
+
+
+ public TajoResultSetMetaData(Schema schema) {
+ this.schema = schema;
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> clazz) throws SQLException {
+ throw new SQLFeatureNotSupportedException("isWrapperFor not supported");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> clazz) throws SQLException {
+ throw new SQLFeatureNotSupportedException("unwrap not supported");
+ }
+
+ @Override
+ public String getCatalogName(int column) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getCatalogName not supported");
+ }
+
+ @Override
+ public String getColumnClassName(int column) throws SQLException {
+ return schema.getColumn(column - 1).getClass().getName();
+ }
+
+ @Override
+ public int getColumnCount() throws SQLException {
+ if(schema == null) {
+ return 0;
+ }
+ return schema.getColumnNum();
+ }
+
+ @Override
+ public int getColumnDisplaySize(int column) throws SQLException {
+ return TajoDriver.columnDisplaySize(getColumnType(column));
+ }
+
+ @Override
+ public String getColumnLabel(int column) throws SQLException {
+ return schema.getColumn(column - 1).getQualifiedName();
+ }
+
+ @Override
+ public String getColumnName(int column) throws SQLException {
+ return schema.getColumn(column - 1).getColumnName();
+ }
+
+ @Override
+ public int getColumnType(int column) throws SQLException {
+ DataType type = schema.getColumn(column - 1).getDataType();
+
+ return TajoDriver.tajoTypeToSqlType(type);
+ }
+
+ @Override
+ public String getColumnTypeName(int column) throws SQLException {
+ DataType type = schema.getColumn(column - 1).getDataType();
+
+ return TajoDriver.toSqlType(type);
+ }
+
+ @Override
+ public int getPrecision(int column) throws SQLException {
+ return TajoDriver.columnDisplaySize(getColumnType(column));
+ }
+
+ @Override
+ public int getScale(int column) throws SQLException {
+ return TajoDriver.columnScale(getColumnType(column));
+ }
+
+ @Override
+ public String getSchemaName(int column) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getSchemaName not supported");
+ }
+
+ @Override
+ public String getTableName(int column) throws SQLException {
+ return schema.getColumn(column - 1).getQualifier();
+ }
+
+ @Override
+ public boolean isAutoIncrement(int column) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean isCaseSensitive(int column) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean isCurrency(int column) throws SQLException {
+ throw new SQLFeatureNotSupportedException("isCurrency not supported");
+ }
+
+ @Override
+ public boolean isDefinitelyWritable(int column) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public int isNullable(int column) throws SQLException {
+ return ResultSetMetaData.columnNullable;
+ }
+
+ @Override
+ public boolean isReadOnly(int column) throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean isSearchable(int column) throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean isSigned(int column) throws SQLException {
+ throw new SQLFeatureNotSupportedException("isSigned not supported");
+ }
+
+ @Override
+ public boolean isWritable(int column) throws SQLException {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoStatement.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoStatement.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoStatement.java
new file mode 100644
index 0000000..6002fcd
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoStatement.java
@@ -0,0 +1,289 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.client.TajoClient;
+
+import java.sql.*;
+
+public class TajoStatement implements Statement {
+ private TajoClient tajoClient;
+ private int fetchSize = 200;
+
+ /**
+ * We need to keep a reference to the result set to support the following:
+ * <code>
+ * statement.execute(String sql);
+ * statement.getResultSet();
+ * </code>.
+ */
+ private ResultSet resultSet = null;
+
+ /**
+ * Add SQLWarnings to the warningChain if needed.
+ */
+ private SQLWarning warningChain = null;
+
+ /**
+ * Keep state so we can fail certain calls made after close().
+ */
+ private boolean isClosed = false;
+
+ public TajoStatement(TajoClient tajoClient) {
+ this.tajoClient = tajoClient;
+ }
+
+ @Override
+ public void addBatch(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("addBatch not supported");
+ }
+
+ @Override
+ public void cancel() throws SQLException {
+ throw new SQLFeatureNotSupportedException("cancel not supported");
+ }
+
+ @Override
+ public void clearBatch() throws SQLException {
+ throw new SQLFeatureNotSupportedException("clearBatch not supported");
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ warningChain = null;
+ }
+
+ @Override
+ public void close() throws SQLException {
+ resultSet = null;
+ isClosed = true;
+ }
+
+ public void closeOnCompletion() throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("closeOnCompletion not supported");
+ }
+
+ @Override
+ public boolean execute(String sql) throws SQLException {
+ resultSet = executeQuery(sql);
+
+ return resultSet != null;
+ }
+
+ @Override
+ public boolean execute(String sql, int autoGeneratedKeys) throws SQLException {
+ throw new SQLFeatureNotSupportedException("execute not supported");
+ }
+
+ @Override
+ public boolean execute(String sql, int[] columnIndexes) throws SQLException {
+ throw new SQLFeatureNotSupportedException("execute not supported");
+ }
+
+ @Override
+ public boolean execute(String sql, String[] columnNames) throws SQLException {
+ throw new SQLFeatureNotSupportedException("execute not supported");
+ }
+
+ @Override
+ public int[] executeBatch() throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeBatch not supported");
+ }
+
+ @Override
+ public ResultSet executeQuery(String sql) throws SQLException {
+ if (isClosed) {
+ throw new SQLFeatureNotSupportedException("Can't execute after statement has been closed");
+ }
+
+ try {
+ return tajoClient.executeQueryAndGetResult(sql);
+ } catch (Exception e) {
+ throw new SQLFeatureNotSupportedException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public int executeUpdate(String sql) throws SQLException {
+ try {
+ tajoClient.executeQuery(sql);
+
+ return 1;
+ } catch (Exception ex) {
+ throw new SQLFeatureNotSupportedException(ex.toString());
+ }
+ }
+
+ @Override
+ public int executeUpdate(String sql, int autoGeneratedKeys) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeUpdate not supported");
+ }
+
+ @Override
+ public int executeUpdate(String sql, int[] columnIndexes) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeUpdate not supported");
+ }
+
+ @Override
+ public int executeUpdate(String sql, String[] columnNames) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeUpdate not supported");
+ }
+
+ @Override
+ public Connection getConnection() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getConnection not supported");
+ }
+
+ @Override
+ public int getFetchDirection() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getFetchDirection not supported");
+ }
+
+ @Override
+ public int getFetchSize() throws SQLException {
+ return fetchSize;
+ }
+
+ @Override
+ public ResultSet getGeneratedKeys() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getGeneratedKeys not supported");
+ }
+
+ @Override
+ public int getMaxFieldSize() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxFieldSize not supported");
+ }
+
+ @Override
+ public int getMaxRows() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxRows not supported");
+ }
+
+ @Override
+ public boolean getMoreResults() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMoreResults not supported");
+ }
+
+ @Override
+ public boolean getMoreResults(int current) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMoreResults not supported");
+ }
+
+ @Override
+ public int getQueryTimeout() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getQueryTimeout not supported");
+ }
+
+ @Override
+ public ResultSet getResultSet() throws SQLException {
+ return resultSet;
+ }
+
+ @Override
+ public int getResultSetConcurrency() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getResultSetConcurrency not supported");
+ }
+
+ @Override
+ public int getResultSetHoldability() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getResultSetHoldability not supported");
+ }
+
+ @Override
+ public int getResultSetType() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getResultSetType not supported");
+ }
+
+ @Override
+ public int getUpdateCount() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ return warningChain;
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return isClosed;
+ }
+
+ public boolean isCloseOnCompletion() throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("isCloseOnCompletion not supported");
+ }
+
+ @Override
+ public boolean isPoolable() throws SQLException {
+ throw new SQLFeatureNotSupportedException("isPoolable not supported");
+ }
+
+ @Override
+ public void setCursorName(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setCursorName not supported");
+ }
+
+ @Override
+ public void setEscapeProcessing(boolean enable) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setEscapeProcessing not supported");
+ }
+
+ @Override
+ public void setFetchDirection(int direction) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setFetchDirection not supported");
+ }
+
+ @Override
+ public void setFetchSize(int rows) throws SQLException {
+ fetchSize = rows;
+ }
+
+ @Override
+ public void setMaxFieldSize(int max) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setMaxFieldSize not supported");
+ }
+
+ @Override
+ public void setMaxRows(int max) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setMaxRows not supported");
+ }
+
+ @Override
+ public void setPoolable(boolean poolable) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setPoolable not supported");
+ }
+
+ @Override
+ public void setQueryTimeout(int seconds) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setQueryTimeout not supported");
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ throw new SQLFeatureNotSupportedException("isWrapperFor not supported");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ throw new SQLFeatureNotSupportedException("unwrap not supported");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
index 487d7af..3fd38ec 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
@@ -22,7 +22,6 @@ import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.tajo.IntegrationTest;
import org.apache.tajo.TpchTestBase;
-import org.apache.tajo.client.ResultSetUtil;
import org.apache.tajo.util.TUtil;
import org.junit.AfterClass;
import org.junit.BeforeClass;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java
deleted file mode 100644
index 48c424d..0000000
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestResultSetImpl.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/**
- * 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.tajo.engine.query;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.tajo.IntegrationTest;
-import org.apache.tajo.TajoTestingCluster;
-import org.apache.tajo.TpchTestBase;
-import org.apache.tajo.catalog.*;
-import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
-import org.apache.tajo.catalog.statistics.TableStats;
-import org.apache.tajo.client.ResultSetImpl;
-import org.apache.tajo.common.TajoDataTypes.Type;
-import org.apache.tajo.conf.TajoConf;
-import org.apache.tajo.datum.DatumFactory;
-import org.apache.tajo.storage.*;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-import java.io.IOException;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-
-import static org.junit.Assert.*;
-
-@Category(IntegrationTest.class)
-public class TestResultSetImpl {
- private static TajoTestingCluster util;
- private static TajoConf conf;
- private static TableDesc desc;
- private static AbstractStorageManager sm;
- private static TableMeta scoreMeta;
- private static Schema scoreSchema;
-
- @BeforeClass
- public static void setup() throws Exception {
- util = TpchTestBase.getInstance().getTestingCluster();
- conf = util.getConfiguration();
- sm = StorageManagerFactory.getStorageManager(conf);
-
- scoreSchema = new Schema();
- scoreSchema.addColumn("deptname", Type.TEXT);
- scoreSchema.addColumn("score", Type.INT4);
- scoreMeta = CatalogUtil.newTableMeta(StoreType.CSV);
- TableStats stats = new TableStats();
-
- Path p = sm.getTablePath("score");
- sm.getFileSystem().mkdirs(p);
- Appender appender = StorageManagerFactory.getStorageManager(conf).getAppender(scoreMeta, scoreSchema,
- new Path(p, "score"));
- appender.init();
- int deptSize = 100;
- int tupleNum = 10000;
- Tuple tuple;
- long written = 0;
- for (int i = 0; i < tupleNum; i++) {
- tuple = new VTuple(2);
- String key = "test" + (i % deptSize);
- tuple.put(0, DatumFactory.createText(key));
- tuple.put(1, DatumFactory.createInt4(i + 1));
- written += key.length() + Integer.SIZE;
- appender.addTuple(tuple);
- }
- appender.close();
- stats.setNumRows(tupleNum);
- stats.setNumBytes(written);
- stats.setAvgRows(tupleNum);
- stats.setNumBlocks(1000);
- stats.setNumPartitions(100);
- desc = new TableDesc("score", scoreSchema, scoreMeta, p);
- desc.setStats(stats);
- }
-
- @AfterClass
- public static void terminate() throws IOException {
-
- }
-
- @Test
- public void test() throws IOException, SQLException {
- ResultSetImpl rs = new ResultSetImpl(null, null, conf, desc);
- ResultSetMetaData meta = rs.getMetaData();
- assertNotNull(meta);
- Schema schema = scoreSchema;
- assertEquals(schema.getColumnNum(), meta.getColumnCount());
- for (int i = 0; i < meta.getColumnCount(); i++) {
- assertEquals(schema.getColumn(i).getColumnName(), meta.getColumnName(i + 1));
- assertEquals(schema.getColumn(i).getQualifier(), meta.getTableName(i + 1));
- assertEquals(schema.getColumn(i).getDataType().getClass().getCanonicalName(),
- meta.getColumnTypeName(i + 1));
- }
-
- int i = 0;
- assertTrue(rs.isBeforeFirst());
- for (; rs.next(); i++) {
- assertEquals("test"+i%100, rs.getString(1));
- assertEquals("test"+i%100, rs.getString("deptname"));
- assertEquals(i+1, rs.getInt(2));
- assertEquals(i+1, rs.getInt("score"));
- }
- assertEquals(10000, i);
- assertTrue(rs.isAfterLast());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
index 264afb7..102414b 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
@@ -18,7 +18,6 @@
package org.apache.tajo.engine.query;
-import org.apache.tajo.client.ResultSetUtil;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
new file mode 100644
index 0000000..0052827
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestResultSet.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.tajo.jdbc;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.IntegrationTest;
+import org.apache.tajo.TajoTestingCluster;
+import org.apache.tajo.TpchTestBase;
+import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
+import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.storage.*;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.io.IOException;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+
+import static org.junit.Assert.*;
+
+@Category(IntegrationTest.class)
+public class TestResultSet {
+ private static TajoTestingCluster util;
+ private static TajoConf conf;
+ private static TableDesc desc;
+ private static AbstractStorageManager sm;
+ private static TableMeta scoreMeta;
+ private static Schema scoreSchema;
+
+ @BeforeClass
+ public static void setup() throws Exception {
+ util = TpchTestBase.getInstance().getTestingCluster();
+ conf = util.getConfiguration();
+ sm = StorageManagerFactory.getStorageManager(conf);
+
+ scoreSchema = new Schema();
+ scoreSchema.addColumn("deptname", Type.TEXT);
+ scoreSchema.addColumn("score", Type.INT4);
+ scoreMeta = CatalogUtil.newTableMeta(StoreType.CSV);
+ TableStats stats = new TableStats();
+
+ Path p = sm.getTablePath("score");
+ sm.getFileSystem().mkdirs(p);
+ Appender appender = StorageManagerFactory.getStorageManager(conf).getAppender(scoreMeta, scoreSchema,
+ new Path(p, "score"));
+ appender.init();
+ int deptSize = 100;
+ int tupleNum = 10000;
+ Tuple tuple;
+ long written = 0;
+ for (int i = 0; i < tupleNum; i++) {
+ tuple = new VTuple(2);
+ String key = "test" + (i % deptSize);
+ tuple.put(0, DatumFactory.createText(key));
+ tuple.put(1, DatumFactory.createInt4(i + 1));
+ written += key.length() + Integer.SIZE;
+ appender.addTuple(tuple);
+ }
+ appender.close();
+ stats.setNumRows(tupleNum);
+ stats.setNumBytes(written);
+ stats.setAvgRows(tupleNum);
+ stats.setNumBlocks(1000);
+ stats.setNumPartitions(100);
+ desc = new TableDesc("score", scoreSchema, scoreMeta, p);
+ desc.setStats(stats);
+ }
+
+ @AfterClass
+ public static void terminate() throws IOException {
+
+ }
+
+ @Test
+ public void test() throws IOException, SQLException {
+ TajoResultSet rs = new TajoResultSet(null, null, conf, desc);
+ ResultSetMetaData meta = rs.getMetaData();
+ assertNotNull(meta);
+ Schema schema = scoreSchema;
+ assertEquals(schema.getColumnNum(), meta.getColumnCount());
+ for (int i = 0; i < meta.getColumnCount(); i++) {
+ assertEquals(schema.getColumn(i).getColumnName(), meta.getColumnName(i + 1));
+ assertEquals(schema.getColumn(i).getQualifier(), meta.getTableName(i + 1));
+// assertEquals(schema.getColumn(i).getDataType().getClass().getCanonicalName(),
+// meta.getColumnTypeName(i + 1));
+ System.out.println(">>>>>>>>>>" + meta.getColumnTypeName(i + 1));
+ }
+
+ int i = 0;
+ assertTrue(rs.isBeforeFirst());
+ for (; rs.next(); i++) {
+ assertEquals("test"+i%100, rs.getString(1));
+ assertEquals("test"+i%100, rs.getString("deptname"));
+ assertEquals(i+1, rs.getInt(2));
+ assertEquals(i+1, rs.getInt("score"));
+ }
+ assertEquals(10000, i);
+ assertTrue(rs.isAfterLast());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java
new file mode 100644
index 0000000..98a7ed2
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/jdbc/TestTajoJdbc.java
@@ -0,0 +1,361 @@
+/**
+ * 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.tajo.jdbc;
+
+import com.google.common.collect.Maps;
+import org.apache.tajo.IntegrationTest;
+import org.apache.tajo.TpchTestBase;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.util.NetUtils;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.net.InetSocketAddress;
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+@Category(IntegrationTest.class)
+public class TestTajoJdbc {
+ private static TpchTestBase tpch;
+ private static Connection conn;
+
+ private static String connUri;
+ @BeforeClass
+ public static void setUp() throws Exception {
+ tpch = TpchTestBase.getInstance();
+
+ TajoConf tajoConf = tpch.getTestingCluster().getMaster().getContext().getConf();
+ InetSocketAddress tajoMasterAddress =
+ NetUtils.createSocketAddr(tajoConf.getVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS));
+
+ Class.forName("org.apache.tajo.jdbc.TajoDriver").newInstance();
+
+ connUri = "jdbc:tajo://" + tajoMasterAddress.getHostName() + ":" + tajoMasterAddress.getPort();
+ conn = DriverManager.getConnection(connUri);
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ if(conn != null) {
+ conn.close();
+ }
+ }
+
+ @Test
+ public void testStatement() throws Exception {
+ Statement stmt = null;
+ ResultSet res = null;
+ try {
+ stmt = conn.createStatement();
+
+ res = stmt.executeQuery("select l_returnflag, l_linestatus, count(*) as count_order from lineitem " +
+ "group by l_returnflag, l_linestatus order by l_returnflag, l_linestatus");
+
+ try {
+ Map<String,Integer> result = Maps.newHashMap();
+ result.put("NO", 3);
+ result.put("RF", 2);
+
+ assertNotNull(res);
+ assertTrue(res.next());
+ assertTrue(result.get(res.getString(1) + res.getString(2)) == res.getInt(3));
+ assertTrue(res.next());
+ assertTrue(result.get(res.getString(1) + res.getString(2)) == res.getInt(3));
+ assertFalse(res.next());
+
+ ResultSetMetaData rsmd = res.getMetaData();
+ assertEquals(3, rsmd.getColumnCount());
+ assertEquals("l_returnflag", rsmd.getColumnName(1));
+ assertEquals("l_linestatus", rsmd.getColumnName(2));
+ assertEquals("count_order", rsmd.getColumnName(3));
+ } finally {
+ res.close();
+ }
+ } finally {
+ if(res != null) {
+ res.close();
+ }
+ if(stmt != null) {
+ stmt.close();
+ }
+ }
+ }
+
+ @Test
+ public void testPreparedStatement() throws Exception {
+ PreparedStatement stmt = null;
+ ResultSet res = null;
+ try {
+ /*
+ test data set
+ 1,17.0,N
+ 1,36.0,N
+ 2,38.0,N
+ 3,45.0,R
+ 3,49.0,R
+ */
+
+ String sql =
+ "select l_orderkey, l_quantity, l_returnflag from lineitem where l_quantity > ? and l_returnflag = ?";
+
+ stmt = conn.prepareStatement(sql);
+
+ stmt.setInt(1, 20);
+ stmt.setString(2, "N");
+
+ res = stmt.executeQuery();
+
+ ResultSetMetaData rsmd = res.getMetaData();
+ assertEquals(3, rsmd.getColumnCount());
+ assertEquals("l_orderkey", rsmd.getColumnName(1));
+ assertEquals("l_quantity", rsmd.getColumnName(2));
+ assertEquals("l_returnflag", rsmd.getColumnName(3));
+
+ try {
+ int numRows = 0;
+ String[] resultData = {"136.0N", "238.0N"};
+ while(res.next()) {
+ assertEquals(resultData[numRows],
+ ("" + res.getObject(1).toString() + res.getObject(2).toString() + res.getObject(3).toString()));
+ numRows++;
+ }
+ assertEquals(2, numRows);
+ } finally {
+ res.close();
+ }
+
+ stmt.setInt(1, 20);
+ stmt.setString(2, "R");
+
+ res = stmt.executeQuery();
+
+ rsmd = res.getMetaData();
+ assertEquals(3, rsmd.getColumnCount());
+ assertEquals("l_orderkey", rsmd.getColumnName(1));
+ assertEquals("l_quantity", rsmd.getColumnName(2));
+ assertEquals("l_returnflag", rsmd.getColumnName(3));
+
+ try {
+ int numRows = 0;
+ String[] resultData = {"345.0R", "349.0R"};
+ while(res.next()) {
+ assertEquals(resultData[numRows],
+ ("" + res.getObject(1).toString() + res.getObject(2).toString() + res.getObject(3).toString()));
+ numRows++;
+ }
+ assertEquals(2, numRows);
+ } finally {
+ res.close();
+ }
+ } finally {
+ if(res != null) {
+ res.close();
+ }
+ if(stmt != null) {
+ stmt.close();
+ }
+ }
+ }
+
+ @Test
+ public void testDatabaseMetaDataGetTable() throws Exception {
+ DatabaseMetaData dbmd = conn.getMetaData();
+
+ ResultSet rs = null;
+
+ try {
+ rs = dbmd.getTables(null, null, null, null);
+
+ ResultSetMetaData rsmd = rs.getMetaData();
+ int numCols = rsmd.getColumnCount();
+
+ assertEquals(5, numCols);
+ int numTables = 0;
+
+ List<String> tableNames = new ArrayList<String>(
+ tpch.getTestingCluster().getMaster().getCatalog().getAllTableNames());
+
+ Collections.sort(tableNames);
+
+ while(rs.next()) {
+ assertEquals(tableNames.get(numTables), rs.getString("TABLE_NAME"));
+ numTables++;
+ }
+
+ assertEquals(tableNames.size(), numTables);
+ } finally {
+ if(rs != null) {
+ rs.close();
+ }
+ }
+ }
+
+ @Test
+ public void testDatabaseMetaDataGetColumns() throws Exception {
+ DatabaseMetaData dbmd = conn.getMetaData();
+
+ ResultSet rs = null;
+
+ try {
+ String tableName = "lineitem";
+ rs = dbmd.getColumns(null, null, tableName, null);
+
+ ResultSetMetaData rsmd = rs.getMetaData();
+ int numCols = rsmd.getColumnCount();
+
+ assertEquals(22, numCols);
+ int numColumns = 0;
+
+ TableDesc tableDesc = tpch.getTestingCluster().getMaster().getCatalog().getTableDesc(tableName);
+ assertNotNull(tableDesc);
+
+ List<Column> columns = tableDesc.getSchema().getColumns();
+
+ while(rs.next()) {
+ assertEquals(tableName, rs.getString("TABLE_NAME"));
+ System.out.println(">>>>" + rs.getString("COLUMN_NAME"));
+ assertEquals(columns.get(numColumns).getColumnName(), rs.getString("COLUMN_NAME"));
+ //TODO assert type
+ numColumns++;
+ }
+
+ assertEquals(16, numColumns);
+ } finally {
+ if(rs != null) {
+ rs.close();
+ }
+ }
+ }
+
+ @Test
+ public void testMultipleConnections() throws Exception {
+ Connection[] conns = new Connection[2];
+ conns[0] = DriverManager.getConnection(connUri);
+ conns[1] = DriverManager.getConnection(connUri);
+
+ try {
+ for(int i = 0; i < conns.length; i++) {
+ Statement stmt = null;
+ ResultSet res = null;
+ try {
+ stmt = conns[i].createStatement();
+
+ res = stmt.executeQuery("select l_returnflag, l_linestatus, count(*) as count_order from lineitem " +
+ "group by l_returnflag, l_linestatus order by l_returnflag, l_linestatus");
+
+ try {
+ Map<String,Integer> result = Maps.newHashMap();
+ result.put("NO", 3);
+ result.put("RF", 2);
+
+ assertNotNull(res);
+ assertTrue(res.next());
+ assertTrue(result.get(res.getString(1) + res.getString(2)) == res.getInt(3));
+ assertTrue(res.next());
+ assertTrue(result.get(res.getString(1) + res.getString(2)) == res.getInt(3));
+ assertFalse(res.next());
+
+ ResultSetMetaData rsmd = res.getMetaData();
+ assertEquals(3, rsmd.getColumnCount());
+ assertEquals("l_returnflag", rsmd.getColumnName(1));
+ assertEquals("l_linestatus", rsmd.getColumnName(2));
+ assertEquals("count_order", rsmd.getColumnName(3));
+ } finally {
+ res.close();
+ }
+ } finally {
+ if(res != null) {
+ res.close();
+ }
+ if(stmt != null) {
+ stmt.close();
+ }
+ }
+ }
+ } finally {
+ conns[0].close();
+ conns[1].close();
+ }
+ }
+
+ @Test
+ public void testMultipleConnectionsSequentialClose() throws Exception {
+ Connection[] conns = new Connection[2];
+ conns[0] = DriverManager.getConnection(connUri);
+ conns[1] = DriverManager.getConnection(connUri);
+
+ try {
+ for(int i = 0; i < conns.length; i++) {
+ Statement stmt = null;
+ ResultSet res = null;
+ try {
+ stmt = conns[i].createStatement();
+
+ res = stmt.executeQuery("select l_returnflag, l_linestatus, count(*) as count_order from lineitem " +
+ "group by l_returnflag, l_linestatus order by l_returnflag, l_linestatus");
+
+ try {
+ Map<String,Integer> result = Maps.newHashMap();
+ result.put("NO", 3);
+ result.put("RF", 2);
+
+ assertNotNull(res);
+ assertTrue(res.next());
+ assertTrue(result.get(res.getString(1) + res.getString(2)) == res.getInt(3));
+ assertTrue(res.next());
+ assertTrue(result.get(res.getString(1) + res.getString(2)) == res.getInt(3));
+ assertFalse(res.next());
+
+ ResultSetMetaData rsmd = res.getMetaData();
+ assertEquals(3, rsmd.getColumnCount());
+ assertEquals("l_returnflag", rsmd.getColumnName(1));
+ assertEquals("l_linestatus", rsmd.getColumnName(2));
+ assertEquals("count_order", rsmd.getColumnName(3));
+ } finally {
+ res.close();
+ }
+ } finally {
+ if(res != null) {
+ res.close();
+ }
+ if(stmt != null) {
+ stmt.close();
+ }
+ conns[i].close();
+ }
+ }
+ } finally {
+ if(!conns[0].isClosed()) {
+ conns[0].close();
+ }
+ if(!conns[1].isClosed()) {
+ conns[1].close();
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-rpc/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/src/main/java/org/apache/tajo/rpc/NettyClientBase.java b/tajo-rpc/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
index 88c8244..a03396e 100644
--- a/tajo-rpc/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
+++ b/tajo-rpc/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
@@ -87,7 +87,9 @@ public abstract class NettyClientBase implements Closeable {
public void close() {
this.channel.close().awaitUninterruptibly();
this.bootstrap.releaseExternalResources();
- LOG.info("Proxy is disconnected from " +
- addr.getAddress().getHostAddress() + ":" + addr.getPort());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Proxy is disconnected from " +
+ addr.getAddress().getHostAddress() + ":" + addr.getPort());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-rpc/src/main/java/org/apache/tajo/rpc/RpcConnectionPool.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/src/main/java/org/apache/tajo/rpc/RpcConnectionPool.java b/tajo-rpc/src/main/java/org/apache/tajo/rpc/RpcConnectionPool.java
index 94f1720..8eacaf5 100644
--- a/tajo-rpc/src/main/java/org/apache/tajo/rpc/RpcConnectionPool.java
+++ b/tajo-rpc/src/main/java/org/apache/tajo/rpc/RpcConnectionPool.java
@@ -40,9 +40,9 @@ public class RpcConnectionPool {
}
public synchronized static RpcConnectionPool getPool(TajoConf conf) {
- if(instance == null) {
- instance = new RpcConnectionPool(conf);
- }
+ if(instance == null) {
+ instance = new RpcConnectionPool(conf);
+ }
return instance;
}
@@ -75,7 +75,9 @@ public class RpcConnectionPool {
}
try {
- LOG.info("CloseConnection [" + client.getKey() + "]");
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("CloseConnection [" + client.getKey() + "]");
+ }
synchronized(connections) {
connections.remove(client.getKey());
}
@@ -85,8 +87,10 @@ public class RpcConnectionPool {
}
}
- public void close() {
- LOG.info("Pool Closed");
+ public synchronized void close() {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Pool Closed");
+ }
synchronized(connections) {
for(NettyClientBase eachClient: connections.values()) {
try {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-rpc/src/main/java/org/apache/tajo/rpc/ServerCallable.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/src/main/java/org/apache/tajo/rpc/ServerCallable.java b/tajo-rpc/src/main/java/org/apache/tajo/rpc/ServerCallable.java
index 50ec204..c5303d8 100644
--- a/tajo-rpc/src/main/java/org/apache/tajo/rpc/ServerCallable.java
+++ b/tajo-rpc/src/main/java/org/apache/tajo/rpc/ServerCallable.java
@@ -35,14 +35,21 @@ public abstract class ServerCallable<T> {
protected long endTime;
protected Class protocol;
protected boolean asyncMode;
+ protected boolean closeConn;
public abstract T call(NettyClientBase client) throws Exception;
public ServerCallable(TajoConf conf, InetSocketAddress addr, Class protocol, boolean asyncMode) {
+ this(conf, addr, protocol, asyncMode, false);
+ }
+
+ public ServerCallable(TajoConf conf, InetSocketAddress addr, Class protocol,
+ boolean asyncMode, boolean closeConn) {
this.tajoConf = conf;
this.addr = addr;
this.protocol = protocol;
this.asyncMode = asyncMode;
+ this.closeConn = closeConn;
}
public void beforeCall() {
@@ -81,8 +88,10 @@ public abstract class ServerCallable<T> {
}
return call(client);
} catch (Throwable t) {
- RpcConnectionPool.getPool(tajoConf).closeConnection(client);
- client = null;
+ if(!closeConn) {
+ RpcConnectionPool.getPool(tajoConf).closeConnection(client);
+ client = null;
+ }
exceptions.add(t);
if(abort) {
throw new ServiceException(t.getMessage(), t);
@@ -92,7 +101,11 @@ public abstract class ServerCallable<T> {
}
} finally {
afterCall();
- RpcConnectionPool.getPool(tajoConf).releaseConnection(client);
+ if(closeConn) {
+ RpcConnectionPool.getPool(tajoConf).closeConnection(client);
+ } else {
+ RpcConnectionPool.getPool(tajoConf).releaseConnection(client);
+ }
}
try {
Thread.sleep(pause * (tries + 1));
@@ -118,8 +131,10 @@ public abstract class ServerCallable<T> {
client = RpcConnectionPool.getPool(tajoConf).getConnection(addr, protocol, asyncMode);
return call(client);
} catch (Throwable t) {
- RpcConnectionPool.getPool(tajoConf).closeConnection(client);
- client = null;
+ if(!closeConn) {
+ RpcConnectionPool.getPool(tajoConf).closeConnection(client);
+ client = null;
+ }
Throwable t2 = translateException(t);
if (t2 instanceof IOException) {
throw (IOException)t2;
@@ -128,7 +143,11 @@ public abstract class ServerCallable<T> {
}
} finally {
afterCall();
- RpcConnectionPool.getPool(tajoConf).releaseConnection(client);
+ if(closeConn) {
+ RpcConnectionPool.getPool(tajoConf).closeConnection(client);
+ } else {
+ RpcConnectionPool.getPool(tajoConf).releaseConnection(client);
+ }
}
}
[3/3] git commit: TAJO-176: Implement Tajo JDBC Driver. (Keuntae Park
via jihoon)
Posted by ji...@apache.org.
TAJO-176: Implement Tajo JDBC Driver. (Keuntae Park via jihoon)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/342fd47f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/342fd47f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/342fd47f
Branch: refs/heads/master
Commit: 342fd47ffd2a30f4941256ac4a5bc95707004599
Parents: 6e2db3b
Author: Jihoon Son <ji...@apache.org>
Authored: Mon Nov 18 17:56:06 2013 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Mon Nov 18 17:56:06 2013 +0900
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../org/apache/tajo/client/ResultSetImpl.java | 2173 ------------------
.../tajo/client/ResultSetMetaDataImpl.java | 260 ---
.../java/org/apache/tajo/client/TajoClient.java | 55 +-
.../org/apache/tajo/jdbc/MetaDataTuple.java | 194 ++
.../org/apache/tajo/jdbc/TajoConnection.java | 400 ++++
.../apache/tajo/jdbc/TajoDatabaseMetaData.java | 1196 ++++++++++
.../java/org/apache/tajo/jdbc/TajoDriver.java | 233 ++
.../apache/tajo/jdbc/TajoMetaDataResultSet.java | 77 +
.../apache/tajo/jdbc/TajoPreparedStatement.java | 660 ++++++
.../org/apache/tajo/jdbc/TajoResultSet.java | 150 ++
.../org/apache/tajo/jdbc/TajoResultSetBase.java | 1129 +++++++++
.../apache/tajo/jdbc/TajoResultSetMetaData.java | 160 ++
.../org/apache/tajo/jdbc/TajoStatement.java | 289 +++
.../tajo/engine/query/TestGroupByQuery.java | 1 -
.../tajo/engine/query/TestResultSetImpl.java | 125 -
.../apache/tajo/engine/query/TestSortQuery.java | 1 -
.../org/apache/tajo/jdbc/TestResultSet.java | 126 +
.../java/org/apache/tajo/jdbc/TestTajoJdbc.java | 361 +++
.../org/apache/tajo/rpc/NettyClientBase.java | 6 +-
.../org/apache/tajo/rpc/RpcConnectionPool.java | 16 +-
.../org/apache/tajo/rpc/ServerCallable.java | 31 +-
22 files changed, 5041 insertions(+), 2604 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 00aa48d..f91d2d1 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,8 @@ Release 0.2.0 - unreleased
NEW FEATURES
+ TAJO-176: Implement Tajo JDBC Driver. (Keuntae Park via jihoon)
+
TAJO-16: Enable Tajo catalog to access Hive metastore. (jaehwa)
TAJO-285: Add CREATE TABLE... BY PARTITION statement to parser. (hyunsik)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetImpl.java
deleted file mode 100644
index bfe0cdc..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetImpl.java
+++ /dev/null
@@ -1,2173 +0,0 @@
-/**
- * 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.tajo.client;
-
-import com.google.common.collect.Lists;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathFilter;
-import org.apache.tajo.QueryId;
-import org.apache.tajo.catalog.Schema;
-import org.apache.tajo.catalog.TableDesc;
-import org.apache.tajo.catalog.TableMeta;
-import org.apache.tajo.datum.Datum;
-import org.apache.tajo.datum.NullDatum;
-import org.apache.tajo.exception.UnsupportedException;
-import org.apache.tajo.storage.fragment.FileFragment;
-import org.apache.tajo.storage.MergeScanner;
-import org.apache.tajo.storage.Scanner;
-import org.apache.tajo.storage.Tuple;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.Reader;
-import java.math.BigDecimal;
-import java.net.URL;
-import java.sql.*;
-import java.sql.Date;
-import java.util.*;
-
-public class ResultSetImpl implements ResultSet {
- private final String cursorName = "tajo";
- private FileSystem fs;
- private Scanner scanner;
- private TableDesc desc;
- private Schema schema;
- private Tuple cur;
- private int curRow;
- private long totalRow;
- private boolean wasNull;
- private TajoClient tajoClient;
- QueryId queryId;
-
- public ResultSetImpl(TajoClient tajoClient, QueryId queryId) {
- this.tajoClient = tajoClient;
- this.queryId = queryId;
- init();
- }
-
- public ResultSetImpl(TajoClient tajoClient, QueryId queryId, Configuration conf, TableDesc desc) throws IOException {
- this.tajoClient = tajoClient;
- this.queryId = queryId;
- this.desc = desc;
- this.schema = desc.getSchema();
- if(desc != null) {
- fs = desc.getPath().getFileSystem(conf);
- this.totalRow = desc.getStats() != null ? desc.getStats().getNumRows() : 0;
- Collection<FileFragment> frags = getFragments(desc.getMeta(), desc.getPath());
- scanner = new MergeScanner(conf, schema, desc.getMeta(), frags);
- }
- init();
- }
-
- private void init() {
- cur = null;
- curRow = 0;
- }
-
- class FileNameComparator implements Comparator<FileStatus> {
-
- @Override
- public int compare(FileStatus f1, FileStatus f2) {
- return f2.getPath().getName().compareTo(f1.getPath().getName());
- }
- }
-
- private Collection<FileFragment> getFragments(TableMeta meta, Path tablePath)
- throws IOException {
- List<FileFragment> fraglist = Lists.newArrayList();
- FileStatus[] files = fs.listStatus(tablePath, new PathFilter() {
- @Override
- public boolean accept(Path path) {
- return path.getName().charAt(0) != '.';
- }
- });
- Arrays.sort(files, new FileNameComparator());
-
- String tbname = tablePath.getName();
- for (int i = 0; i < files.length; i++) {
- if (files[i].getLen() == 0) {
- continue;
- }
- fraglist.add(new FileFragment(tbname + "_" + i, files[i].getPath(), 0l, files[i].getLen()));
- }
- return fraglist;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.Wrapper#isWrapperFor(java.lang.Class)
- */
- @Override
- public boolean isWrapperFor(Class<?> arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.Wrapper#unwrap(java.lang.Class)
- */
- @Override
- public <T> T unwrap(Class<T> arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#absolute(int)
- */
- @Override
- public boolean absolute(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#afterLast()
- */
- @Override
- public void afterLast() throws SQLException {
- while (this.next())
- ;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#beforeFirst()
- */
- @Override
- public void beforeFirst() throws SQLException {
- try {
- if(scanner != null) {
- scanner.reset();
- }
- init();
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#cancelRowUpdates()
- */
- @Override
- public void cancelRowUpdates() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#clearWarnings()
- */
- @Override
- public void clearWarnings() throws SQLException {
- // TODO
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#close()
- */
- @Override
- public void close() throws SQLException {
- try {
- if(tajoClient != null) {
- this.tajoClient.closeQuery(queryId);
- }
- } catch (Exception e) {
- e.printStackTrace();
- }
- try {
- if(scanner != null) {
- this.scanner.close();
- }
- //TODO clean temp result file
- cur = null;
- curRow = -1;
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#deleteRow()
- */
- @Override
- public void deleteRow() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#findColumn(java.lang.String)
- */
- @Override
- public int findColumn(String colName) throws SQLException {
- return schema.getColumnIdByName(colName);
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#first()
- */
- @Override
- public boolean first() throws SQLException {
- this.beforeFirst();
- return this.next();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getArray(int)
- */
- @Override
- public Array getArray(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getArray(java.lang.String)
- */
- @Override
- public Array getArray(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getAsciiStream(int)
- */
- @Override
- public InputStream getAsciiStream(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getAsciiStream(java.lang.String)
- */
- @Override
- public InputStream getAsciiStream(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBigDecimal(int)
- */
- @Override
- public BigDecimal getBigDecimal(int fieldId) throws SQLException {
- // TODO
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBigDecimal(java.lang.String)
- */
- @Override
- public BigDecimal getBigDecimal(String fieldName) throws SQLException {
- // TODO
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBigDecimal(int, int)
- */
- @Override
- public BigDecimal getBigDecimal(int arg0, int arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBigDecimal(java.lang.String, int)
- */
- @Override
- public BigDecimal getBigDecimal(String arg0, int arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBinaryStream(int)
- */
- @Override
- public InputStream getBinaryStream(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBinaryStream(java.lang.String)
- */
- @Override
- public InputStream getBinaryStream(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBlob(int)
- */
- @Override
- public Blob getBlob(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBlob(java.lang.String)
- */
- @Override
- public Blob getBlob(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBoolean(int)
- */
- @Override
- public boolean getBoolean(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asBool();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBoolean(java.lang.String)
- */
- @Override
- public boolean getBoolean(String colName) throws SQLException {
- Datum datum = cur.get(findColumn(colName));
- handleNull(datum);
- return datum.asBool();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getByte(int)
- */
- @Override
- public byte getByte(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asByte();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getByte(java.lang.String)
- */
- @Override
- public byte getByte(String name) throws SQLException {
- Datum datum = cur.get(findColumn(name));
- handleNull(datum);
- return datum.asByte();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBytes(int)
- */
- @Override
- public byte[] getBytes(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asByteArray();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getBytes(java.lang.String)
- */
- @Override
- public byte[] getBytes(String name) throws SQLException {
- Datum datum = cur.get(findColumn(name));
- handleNull(datum);
- return datum.asByteArray();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getCharacterStream(int)
- */
- @Override
- public Reader getCharacterStream(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getCharacterStream(java.lang.String)
- */
- @Override
- public Reader getCharacterStream(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getClob(int)
- */
- @Override
- public Clob getClob(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getClob(java.lang.String)
- */
- @Override
- public Clob getClob(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getConcurrency()
- */
- @Override
- public int getConcurrency() throws SQLException {
- return ResultSet.CONCUR_READ_ONLY;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getCursorName()
- */
- @Override
- public String getCursorName() throws SQLException {
- return cursorName;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getDate(int)
- */
- @Override
- public Date getDate(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getDate(java.lang.String)
- */
- @Override
- public Date getDate(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getDate(int, java.util.Calendar)
- */
- @Override
- public Date getDate(int arg0, Calendar arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getDate(java.lang.String, java.util.Calendar)
- */
- @Override
- public Date getDate(String arg0, Calendar arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getDouble(int)
- */
- @Override
- public double getDouble(int fieldId) throws SQLException {
- Datum datum = cur.getDouble(fieldId - 1);
- handleNull(datum);
- return datum.asFloat8();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getDouble(java.lang.String)
- */
- @Override
- public double getDouble(String name) throws SQLException {
- Datum datum = cur.get(findColumn(name));
- handleNull(datum);
- return datum.asFloat8();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getFetchDirection()
- */
- @Override
- public int getFetchDirection() throws SQLException {
- return ResultSet.FETCH_FORWARD;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getFetchSize()
- */
- @Override
- public int getFetchSize() throws SQLException {
- return 0;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getFloat(int)
- */
- @Override
- public float getFloat(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asFloat4();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getFloat(java.lang.String)
- */
- @Override
- public float getFloat(String name) throws SQLException {
- Datum datum = cur.get(findColumn(name));
- handleNull(datum);
- return datum.asFloat4();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getHoldability()
- */
- @Override
- public int getHoldability() throws SQLException {
- return 0;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getInt(int)
- */
- @Override
- public int getInt(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asInt4();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getInt(java.lang.String)
- */
- @Override
- public int getInt(String name) throws SQLException {
- Datum datum = cur.get(findColumn(name));
- handleNull(datum);
- return datum.asInt4();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getLong(int)
- */
- @Override
- public long getLong(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asInt8();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getLong(java.lang.String)
- */
- @Override
- public long getLong(String name) throws SQLException {
- Datum datum = cur.get(findColumn(name));
- handleNull(datum);
- return datum.asInt8();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getMetaData()
- */
- @Override
- public ResultSetMetaData getMetaData() throws SQLException {
- return new ResultSetMetaDataImpl(desc);
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getNCharacterStream(int)
- */
- @Override
- public Reader getNCharacterStream(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getNCharacterStream(java.lang.String)
- */
- @Override
- public Reader getNCharacterStream(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getNClob(int)
- */
- @Override
- public NClob getNClob(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getNClob(java.lang.String)
- */
- @Override
- public NClob getNClob(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getNString(int)
- */
- @Override
- public String getNString(int fieldId) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getNString(java.lang.String)
- */
- @Override
- public String getNString(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getObject(int)
- */
- @Override
- public Object getObject(int fieldId) throws SQLException {
- Datum d = cur.get(fieldId - 1);
- handleNull(d);
-
- // TODO - to be changed to return Object type
- return d;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getObject(java.lang.String)
- */
- @Override
- public Object getObject(String arg0) throws SQLException {
- Datum d = cur.get(findColumn(arg0));
- handleNull(d);
- return d;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getObject(int, java.util.Map)
- */
- @Override
- public Object getObject(int arg0, Map<String, Class<?>> arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getObject(java.lang.String, java.util.Map)
- */
- @Override
- public Object getObject(String arg0, Map<String, Class<?>> arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getObject(java.lang.String, java.lang.Class)
- */
- public <T> T getObject(String arg0, Class<T> arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getObject(int, java.lang.Class)
- */
- public <T> T getObject(int arg0, Class<T> arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getRef(int)
- */
- @Override
- public Ref getRef(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getRef(java.lang.String)
- */
- @Override
- public Ref getRef(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getRow()
- */
- @Override
- public int getRow() throws SQLException {
- return curRow;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getRowId(int)
- */
- @Override
- public RowId getRowId(int fieldId) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getRowId(java.lang.String)
- */
- @Override
- public RowId getRowId(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getSQLXML(int)
- */
- @Override
- public SQLXML getSQLXML(int fieldId) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getSQLXML(java.lang.String)
- */
- @Override
- public SQLXML getSQLXML(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getShort(int)
- */
- @Override
- public short getShort(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asInt2();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getShort(java.lang.String)
- */
- @Override
- public short getShort(String name) throws SQLException {
- Datum datum = cur.get(findColumn(name));
- handleNull(datum);
- return datum.asInt2();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getStatement()
- */
- @Override
- public Statement getStatement() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getString(int)
- */
- @Override
- public String getString(int fieldId) throws SQLException {
- Datum datum = cur.get(fieldId - 1);
- handleNull(datum);
- return datum.asChars();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getString(java.lang.String)
- */
- @Override
- public String getString(String arg0) throws SQLException {
- Datum datum = cur.get(findColumn(arg0));
- handleNull(datum);
- return datum.asChars();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTime(int)
- */
- @Override
- public Time getTime(int fieldId) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTime(java.lang.String)
- */
- @Override
- public Time getTime(String name) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTime(int, java.util.Calendar)
- */
- @Override
- public Time getTime(int fieldId, Calendar arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTime(java.lang.String, java.util.Calendar)
- */
- @Override
- public Time getTime(String name, Calendar arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTimestamp(int)
- */
- @Override
- public Timestamp getTimestamp(int fieldId) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTimestamp(java.lang.String)
- */
- @Override
- public Timestamp getTimestamp(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTimestamp(int, java.util.Calendar)
- */
- @Override
- public Timestamp getTimestamp(int fieldId, Calendar arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getTimestamp(java.lang.String, java.util.Calendar)
- */
- @Override
- public Timestamp getTimestamp(String arg0, Calendar arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getType()
- */
- @Override
- public int getType() throws SQLException {
- return ResultSet.TYPE_FORWARD_ONLY;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getURL(int)
- */
- @Override
- public URL getURL(int fieldId) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getURL(java.lang.String)
- */
- @Override
- public URL getURL(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getUnicodeStream(int)
- */
- @Override
- public InputStream getUnicodeStream(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getUnicodeStream(java.lang.String)
- */
- @Override
- public InputStream getUnicodeStream(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#getWarnings()
- */
- @Override
- public SQLWarning getWarnings() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#insertRow()
- */
- @Override
- public void insertRow() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#isAfterLast()
- */
- @Override
- public boolean isAfterLast() throws SQLException {
- return this.curRow > this.totalRow;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#isBeforeFirst()
- */
- @Override
- public boolean isBeforeFirst() throws SQLException {
- return this.curRow == 0;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#isClosed()
- */
- @Override
- public boolean isClosed() throws SQLException {
- return this.curRow == -1;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#isFirst()
- */
- @Override
- public boolean isFirst() throws SQLException {
- return this.curRow == 1;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#isLast()
- */
- @Override
- public boolean isLast() throws SQLException {
- return this.curRow == this.totalRow;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#last()
- */
- @Override
- public boolean last() throws SQLException {
- Tuple last = null;
- while (this.next()) {
- last = cur;
- }
- cur = last;
- return true;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#moveToCurrentRow()
- */
- @Override
- public void moveToCurrentRow() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#moveToInsertRow()
- */
- @Override
- public void moveToInsertRow() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#next()
- */
- @Override
- public boolean next() throws SQLException {
- if(scanner == null) {
- return false;
- }
- try {
- if (totalRow <= 0)
- return false;
-
- cur = scanner.next();
- curRow++;
- if (cur != null) {
- return true;
- }
- } catch (IOException e) {
- e.printStackTrace();
- }
- return false;
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#previous()
- */
- @Override
- public boolean previous() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#refreshRow()
- */
- @Override
- public void refreshRow() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#relative(int)
- */
- @Override
- public boolean relative(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#rowDeleted()
- */
- @Override
- public boolean rowDeleted() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#rowInserted()
- */
- @Override
- public boolean rowInserted() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#rowUpdated()
- */
- @Override
- public boolean rowUpdated() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#setFetchDirection(int)
- */
- @Override
- public void setFetchDirection(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#setFetchSize(int)
- */
- @Override
- public void setFetchSize(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateArray(int, java.sql.Array)
- */
- @Override
- public void updateArray(int arg0, Array arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateArray(java.lang.String, java.sql.Array)
- */
- @Override
- public void updateArray(String arg0, Array arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateAsciiStream(int, java.io.InputStream)
- */
- @Override
- public void updateAsciiStream(int arg0, InputStream arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateAsciiStream(java.lang.String,
- * java.io.InputStream)
- */
- @Override
- public void updateAsciiStream(String arg0, InputStream arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateAsciiStream(int, java.io.InputStream, int)
- */
- @Override
- public void updateAsciiStream(int arg0, InputStream arg1, int arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateAsciiStream(java.lang.String,
- * java.io.InputStream, int)
- */
- @Override
- public void updateAsciiStream(String arg0, InputStream arg1, int arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateAsciiStream(int, java.io.InputStream, long)
- */
- @Override
- public void updateAsciiStream(int arg0, InputStream arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateAsciiStream(java.lang.String,
- * java.io.InputStream, long)
- */
- @Override
- public void updateAsciiStream(String arg0, InputStream arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBigDecimal(int, java.math.BigDecimal)
- */
- @Override
- public void updateBigDecimal(int arg0, BigDecimal arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBigDecimal(java.lang.String,
- * java.math.BigDecimal)
- */
- @Override
- public void updateBigDecimal(String arg0, BigDecimal arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBinaryStream(int, java.io.InputStream)
- */
- @Override
- public void updateBinaryStream(int arg0, InputStream arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBinaryStream(java.lang.String,
- * java.io.InputStream)
- */
- @Override
- public void updateBinaryStream(String arg0, InputStream arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBinaryStream(int, java.io.InputStream, int)
- */
- @Override
- public void updateBinaryStream(int arg0, InputStream arg1, int arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBinaryStream(java.lang.String,
- * java.io.InputStream, int)
- */
- @Override
- public void updateBinaryStream(String arg0, InputStream arg1, int arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBinaryStream(int, java.io.InputStream, long)
- */
- @Override
- public void updateBinaryStream(int arg0, InputStream arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBinaryStream(java.lang.String,
- * java.io.InputStream, long)
- */
- @Override
- public void updateBinaryStream(String arg0, InputStream arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBlob(int, java.sql.Blob)
- */
- @Override
- public void updateBlob(int arg0, Blob arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBlob(java.lang.String, java.sql.Blob)
- */
- @Override
- public void updateBlob(String arg0, Blob arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBlob(int, java.io.InputStream)
- */
- @Override
- public void updateBlob(int arg0, InputStream arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBlob(java.lang.String, java.io.InputStream)
- */
- @Override
- public void updateBlob(String arg0, InputStream arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBlob(int, java.io.InputStream, long)
- */
- @Override
- public void updateBlob(int arg0, InputStream arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBlob(java.lang.String, java.io.InputStream,
- * long)
- */
- @Override
- public void updateBlob(String arg0, InputStream arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBoolean(int, boolean)
- */
- @Override
- public void updateBoolean(int arg0, boolean arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBoolean(java.lang.String, boolean)
- */
- @Override
- public void updateBoolean(String arg0, boolean arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateByte(int, byte)
- */
- @Override
- public void updateByte(int arg0, byte arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateByte(java.lang.String, byte)
- */
- @Override
- public void updateByte(String arg0, byte arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBytes(int, byte[])
- */
- @Override
- public void updateBytes(int arg0, byte[] arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateBytes(java.lang.String, byte[])
- */
- @Override
- public void updateBytes(String arg0, byte[] arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateCharacterStream(int, java.io.Reader)
- */
- @Override
- public void updateCharacterStream(int arg0, Reader arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateCharacterStream(java.lang.String,
- * java.io.Reader)
- */
- @Override
- public void updateCharacterStream(String arg0, Reader arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateCharacterStream(int, java.io.Reader, int)
- */
- @Override
- public void updateCharacterStream(int arg0, Reader arg1, int arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateCharacterStream(java.lang.String,
- * java.io.Reader, int)
- */
- @Override
- public void updateCharacterStream(String arg0, Reader arg1, int arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateCharacterStream(int, java.io.Reader, long)
- */
- @Override
- public void updateCharacterStream(int arg0, Reader arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateCharacterStream(java.lang.String,
- * java.io.Reader, long)
- */
- @Override
- public void updateCharacterStream(String arg0, Reader arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateClob(int, java.sql.Clob)
- */
- @Override
- public void updateClob(int arg0, Clob arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateClob(java.lang.String, java.sql.Clob)
- */
- @Override
- public void updateClob(String arg0, Clob arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateClob(int, java.io.Reader)
- */
- @Override
- public void updateClob(int arg0, Reader arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateClob(java.lang.String, java.io.Reader)
- */
- @Override
- public void updateClob(String arg0, Reader arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateClob(int, java.io.Reader, long)
- */
- @Override
- public void updateClob(int arg0, Reader arg1, long arg2) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateClob(java.lang.String, java.io.Reader, long)
- */
- @Override
- public void updateClob(String arg0, Reader arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateDate(int, java.sql.Date)
- */
- @Override
- public void updateDate(int arg0, Date arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateDate(java.lang.String, java.sql.Date)
- */
- @Override
- public void updateDate(String arg0, Date arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateDouble(int, double)
- */
- @Override
- public void updateDouble(int arg0, double arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateDouble(java.lang.String, double)
- */
- @Override
- public void updateDouble(String arg0, double arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateFloat(int, float)
- */
- @Override
- public void updateFloat(int arg0, float arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateFloat(java.lang.String, float)
- */
- @Override
- public void updateFloat(String arg0, float arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateInt(int, int)
- */
- @Override
- public void updateInt(int arg0, int arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateInt(java.lang.String, int)
- */
- @Override
- public void updateInt(String arg0, int arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateLong(int, long)
- */
- @Override
- public void updateLong(int arg0, long arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateLong(java.lang.String, long)
- */
- @Override
- public void updateLong(String arg0, long arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNCharacterStream(int, java.io.Reader)
- */
- @Override
- public void updateNCharacterStream(int arg0, Reader arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNCharacterStream(java.lang.String,
- * java.io.Reader)
- */
- @Override
- public void updateNCharacterStream(String arg0, Reader arg1)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNCharacterStream(int, java.io.Reader, long)
- */
- @Override
- public void updateNCharacterStream(int arg0, Reader arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNCharacterStream(java.lang.String,
- * java.io.Reader, long)
- */
- @Override
- public void updateNCharacterStream(String arg0, Reader arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNClob(int, java.sql.NClob)
- */
- @Override
- public void updateNClob(int arg0, NClob arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNClob(java.lang.String, java.sql.NClob)
- */
- @Override
- public void updateNClob(String arg0, NClob arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNClob(int, java.io.Reader)
- */
- @Override
- public void updateNClob(int arg0, Reader arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNClob(java.lang.String, java.io.Reader)
- */
- @Override
- public void updateNClob(String arg0, Reader arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNClob(int, java.io.Reader, long)
- */
- @Override
- public void updateNClob(int arg0, Reader arg1, long arg2) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNClob(java.lang.String, java.io.Reader, long)
- */
- @Override
- public void updateNClob(String arg0, Reader arg1, long arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNString(int, java.lang.String)
- */
- @Override
- public void updateNString(int arg0, String arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNString(java.lang.String, java.lang.String)
- */
- @Override
- public void updateNString(String arg0, String arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNull(int)
- */
- @Override
- public void updateNull(int arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateNull(java.lang.String)
- */
- @Override
- public void updateNull(String arg0) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateObject(int, java.lang.Object)
- */
- @Override
- public void updateObject(int arg0, Object arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateObject(java.lang.String, java.lang.Object)
- */
- @Override
- public void updateObject(String arg0, Object arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateObject(int, java.lang.Object, int)
- */
- @Override
- public void updateObject(int arg0, Object arg1, int arg2) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateObject(java.lang.String, java.lang.Object,
- * int)
- */
- @Override
- public void updateObject(String arg0, Object arg1, int arg2)
- throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateRef(int, java.sql.Ref)
- */
- @Override
- public void updateRef(int arg0, Ref arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateRef(java.lang.String, java.sql.Ref)
- */
- @Override
- public void updateRef(String arg0, Ref arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateRow()
- */
- @Override
- public void updateRow() throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateRowId(int, java.sql.RowId)
- */
- @Override
- public void updateRowId(int arg0, RowId arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateRowId(java.lang.String, java.sql.RowId)
- */
- @Override
- public void updateRowId(String arg0, RowId arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateSQLXML(int, java.sql.SQLXML)
- */
- @Override
- public void updateSQLXML(int arg0, SQLXML arg1) throws SQLException {
- throw new UnsupportedException();
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateSQLXML(java.lang.String, java.sql.SQLXML)
- */
- @Override
- public void updateSQLXML(String arg0, SQLXML arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateShort(int, short)
- */
- @Override
- public void updateShort(int arg0, short arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateShort(java.lang.String, short)
- */
- @Override
- public void updateShort(String arg0, short arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateString(int, java.lang.String)
- */
- @Override
- public void updateString(int arg0, String arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateString(java.lang.String, java.lang.String)
- */
- @Override
- public void updateString(String arg0, String arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateTime(int, java.sql.Time)
- */
- @Override
- public void updateTime(int arg0, Time arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateTime(java.lang.String, java.sql.Time)
- */
- @Override
- public void updateTime(String arg0, Time arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateTimestamp(int, java.sql.Timestamp)
- */
- @Override
- public void updateTimestamp(int arg0, Timestamp arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#updateTimestamp(java.lang.String,
- * java.sql.Timestamp)
- */
- @Override
- public void updateTimestamp(String arg0, Timestamp arg1) throws SQLException {
- throw new UnsupportedException();
-
- }
-
- /*
- * (non-Javadoc)
- *
- * @see java.sql.ResultSet#wasNull()
- */
- @Override
- public boolean wasNull() throws SQLException {
- return wasNull;
- }
-
- private void handleNull(Datum d) {
- wasNull = (d instanceof NullDatum);
- }
-
- public boolean hasResult() {
- return scanner != null;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetMetaDataImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetMetaDataImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetMetaDataImpl.java
deleted file mode 100644
index 6e2d6e9..0000000
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/ResultSetMetaDataImpl.java
+++ /dev/null
@@ -1,260 +0,0 @@
-/**
- * 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.tajo.client;
-
-import org.apache.tajo.catalog.TableDesc;
-import org.apache.tajo.common.TajoDataTypes.DataType;
-import org.apache.tajo.exception.UnsupportedException;
-
-import java.nio.channels.UnsupportedAddressTypeException;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Types;
-
-public class ResultSetMetaDataImpl implements ResultSetMetaData {
- private TableDesc desc;
-
-
- public ResultSetMetaDataImpl(TableDesc desc) {
- this.desc = desc;
- }
-
- /* (non-Javadoc)
- * @see java.sql.Wrapper#isWrapperFor(java.lang.Class)
- */
- @Override
- public boolean isWrapperFor(Class<?> arg0) throws SQLException {
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.Wrapper#unwrap(java.lang.Class)
- */
- @Override
- public <T> T unwrap(Class<T> arg0) throws SQLException {
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getCatalogName(int)
- */
- @Override
- public String getCatalogName(int column) throws SQLException {
- // TODO Auto-generated method stub
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getColumnClassName(int)
- */
- @Override
- public String getColumnClassName(int column) throws SQLException {
- return desc.getSchema().getColumn(column - 1).getClass().getName();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getColumnCount()
- */
- @Override
- public int getColumnCount() throws SQLException {
- return desc.getSchema().getColumnNum();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getColumnDisplaySize(int)
- */
- @Override
- public int getColumnDisplaySize(int column) throws SQLException {
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getColumnLabel(int)
- */
- @Override
- public String getColumnLabel(int column) throws SQLException {
- return desc.getSchema().getColumn(column - 1).getQualifiedName();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getColumnName(int)
- */
- @Override
- public String getColumnName(int column) throws SQLException {
- return desc.getSchema().getColumn(column - 1).getColumnName();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getColumnType(int)
- */
- @Override
- public int getColumnType(int column) throws SQLException {
- // TODO
- DataType type = desc.getSchema().getColumn(column - 1).getDataType();
- switch (type.getType()) {
- case BOOLEAN:
- return Types.BOOLEAN;
- case INT1:
- return Types.TINYINT;
- case INT2:
- return Types.SMALLINT;
- case INT4:
- return Types.INTEGER;
- case INT8:
- return Types.BIGINT;
- case FLOAT4:
- return Types.FLOAT;
- case FLOAT8:
- return Types.DOUBLE;
- case DECIMAL:
- return Types.DECIMAL;
- case VARBINARY:
- return Types.VARBINARY;
- case CHAR:
- return Types.CHAR;
- case DATE:
- return Types.DATE;
- case VARCHAR:
- return Types.VARCHAR;
- case TEXT:
- return Types.VARCHAR;
- default:
- throw new UnsupportedException();
- }
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getColumnTypeName(int)
- */
- @Override
- public String getColumnTypeName(int column) throws SQLException {
- return desc.getSchema().getColumn(column - 1).
- getDataType().getClass().getCanonicalName();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getPrecision(int)
- */
- @Override
- public int getPrecision(int column) throws SQLException {
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getScale(int)
- */
- @Override
- public int getScale(int column) throws SQLException {
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getSchemaName(int)
- */
- @Override
- public String getSchemaName(int column) throws SQLException {
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#getTableName(int)
- */
- @Override
- public String getTableName(int column) throws SQLException {
- return desc.getSchema().getColumn(column - 1).getQualifier();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isAutoIncrement(int)
- */
- @Override
- public boolean isAutoIncrement(int column) throws SQLException {
- return false;
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isCaseSensitive(int)
- */
- @Override
- public boolean isCaseSensitive(int column) throws SQLException {
- return false;
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isCurrency(int)
- */
- @Override
- public boolean isCurrency(int column) throws SQLException {
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isDefinitelyWritable(int)
- */
- @Override
- public boolean isDefinitelyWritable(int column) throws SQLException {
- return false;
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isNullable(int)
- */
- @Override
- public int isNullable(int column) throws SQLException {
- // TODO Auto-generated method stub
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isReadOnly(int)
- */
- @Override
- public boolean isReadOnly(int column) throws SQLException {
- return true;
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isSearchable(int)
- */
- @Override
- public boolean isSearchable(int column) throws SQLException {
- // TODO
- return true;
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isSigned(int)
- */
- @Override
- public boolean isSigned(int column) throws SQLException {
- // TODO Auto-generated method stub
- throw new UnsupportedAddressTypeException();
- }
-
- /* (non-Javadoc)
- * @see java.sql.ResultSetMetaData#isWritable(int)
- */
- @Override
- public boolean isWritable(int column) throws SQLException {
- return false;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
index 7e8aee0..91a8196 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/client/TajoClient.java
@@ -37,6 +37,7 @@ import org.apache.tajo.ipc.QueryMasterClientProtocol;
import org.apache.tajo.ipc.QueryMasterClientProtocol.QueryMasterClientProtocolService;
import org.apache.tajo.ipc.TajoMasterClientProtocol;
import org.apache.tajo.ipc.TajoMasterClientProtocol.TajoMasterClientProtocolService;
+import org.apache.tajo.jdbc.TajoResultSet;
import org.apache.tajo.rpc.*;
import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.StringProto;
import org.apache.tajo.util.NetUtils;
@@ -83,9 +84,16 @@ public class TajoClient {
}
public void close() {
+ if(connPool != null) {
+ connPool.close();
+ }
queryMasterMap.clear();
}
+ public TajoConf getConf() {
+ return conf;
+ }
+
/**
* Call to QueryMaster closing query resources
* @param queryId
@@ -98,11 +106,9 @@ public class TajoClient {
QueryMasterClientProtocolService.BlockingInterface queryMasterService = qmClient.getStub();
queryMasterService.killQuery(null, queryId.getProto());
} catch (Exception e) {
- connPool.closeConnection(qmClient);
- qmClient = null;
LOG.warn("Fail to close a QueryMaster connection (qid=" + queryId + ", msg=" + e.getMessage() + ")", e);
} finally {
- connPool.releaseConnection(qmClient);
+ connPool.closeConnection(qmClient);
queryMasterMap.remove(queryId);
}
}
@@ -116,7 +122,7 @@ public class TajoClient {
*/
public GetQueryStatusResponse executeQuery(final String sql) throws ServiceException {
return new ServerCallable<GetQueryStatusResponse>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public GetQueryStatusResponse call(NettyClientBase client) throws ServiceException {
final QueryRequest.Builder builder = QueryRequest.newBuilder();
builder.setQuery(sql);
@@ -138,7 +144,7 @@ public class TajoClient {
public ResultSet executeQueryAndGetResult(final String sql)
throws ServiceException, IOException {
GetQueryStatusResponse response = new ServerCallable<GetQueryStatusResponse>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public GetQueryStatusResponse call(NettyClientBase client) throws ServiceException {
final QueryRequest.Builder builder = QueryRequest.newBuilder();
builder.setQuery(sql);
@@ -170,11 +176,9 @@ public class TajoClient {
QueryMasterClientProtocolService.BlockingInterface queryMasterService = qmClient.getStub();
res = queryMasterService.getQueryStatus(null, builder.build());
} catch (Exception e) {
- connPool.closeConnection(qmClient);
- qmClient = null;
throw new ServiceException(e.getMessage(), e);
} finally {
- connPool.releaseConnection(qmClient);
+ connPool.closeConnection(qmClient);
}
} else {
NettyClientBase tmClient = null;
@@ -196,19 +200,15 @@ public class TajoClient {
queryMasterMap.put(queryId, qmAddr);
} catch (Exception e) {
- connPool.closeConnection(qmClient);
- qmClient = null;
throw new ServiceException(e.getMessage(), e);
} finally {
- connPool.releaseConnection(qmClient);
+ connPool.closeConnection(qmClient);
}
}
} catch (Exception e) {
- connPool.closeConnection(tmClient);
- tmClient = null;
throw new ServiceException(e.getMessage(), e);
} finally {
- connPool.releaseConnection(tmClient);
+ connPool.closeConnection(tmClient);
}
}
return new QueryStatus(res);
@@ -228,9 +228,8 @@ public class TajoClient {
if (queryId.equals(QueryIdFactory.NULL_QUERY_ID)) {
return createNullResultSet(queryId);
}
-
TableDesc tableDesc = getResultDesc(queryId);
- return new ResultSetImpl(this, queryId, conf, tableDesc);
+ return new TajoResultSet(this, queryId, conf, tableDesc);
}
public ResultSet getQueryResultAndWait(QueryId queryId)
@@ -266,7 +265,7 @@ public class TajoClient {
}
public ResultSet createNullResultSet(QueryId queryId) throws IOException {
- return new ResultSetImpl(this, queryId);
+ return new TajoResultSet(this, queryId);
}
public TableDesc getResultDesc(QueryId queryId) throws ServiceException {
@@ -290,17 +289,15 @@ public class TajoClient {
return CatalogUtil.newTableDesc(response.getTableDesc());
} catch (Exception e) {
- connPool.closeConnection(client);
- client = null;
throw new ServiceException(e.getMessage(), e);
} finally {
- connPool.releaseConnection(client);
+ connPool.closeConnection(client);
}
}
public boolean updateQuery(final String sql) throws ServiceException {
return new ServerCallable<Boolean>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public Boolean call(NettyClientBase client) throws ServiceException {
QueryRequest.Builder builder = QueryRequest.newBuilder();
builder.setQuery(sql);
@@ -323,7 +320,7 @@ public class TajoClient {
*/
public boolean existTable(final String name) throws ServiceException {
return new ServerCallable<Boolean>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public Boolean call(NettyClientBase client) throws ServiceException {
StringProto.Builder builder = StringProto.newBuilder();
builder.setValue(name);
@@ -342,7 +339,7 @@ public class TajoClient {
*/
public boolean detachTable(final String tableName) throws ServiceException {
return new ServerCallable<Boolean>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public Boolean call(NettyClientBase client) throws ServiceException {
TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
@@ -356,7 +353,7 @@ public class TajoClient {
public TableDesc createExternalTable(final String name, final Schema schema, final Path path, final TableMeta meta)
throws SQLException, ServiceException {
return new ServerCallable<TableDesc>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public TableDesc call(NettyClientBase client) throws ServiceException, SQLException {
TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
@@ -383,7 +380,7 @@ public class TajoClient {
*/
public boolean dropTable(final String tableName) throws ServiceException {
return new ServerCallable<Boolean>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public Boolean call(NettyClientBase client) throws ServiceException {
TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
@@ -401,7 +398,7 @@ public class TajoClient {
*/
public List<String> getTableList() throws ServiceException {
return new ServerCallable<List<String>>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public List<String> call(NettyClientBase client) throws ServiceException {
TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
@@ -414,7 +411,7 @@ public class TajoClient {
public TableDesc getTableDesc(final String tableName) throws SQLException, ServiceException {
return new ServerCallable<TableDesc>(conf, tajoMasterAddr,
- TajoMasterClientProtocol.class, false) {
+ TajoMasterClientProtocol.class, false, true) {
public TableDesc call(NettyClientBase client) throws ServiceException, SQLException {
TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
@@ -456,12 +453,10 @@ public class TajoClient {
status = getQueryStatus(queryId);
}
} catch(Exception e) {
- connPool.closeConnection(tmClient);
- tmClient = null;
LOG.debug("Error when checking for application status", e);
return false;
} finally {
- connPool.releaseConnection(tmClient);
+ connPool.closeConnection(tmClient);
}
return true;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java
new file mode 100644
index 0000000..789f761
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/MetaDataTuple.java
@@ -0,0 +1,194 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.datum.*;
+import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.storage.Tuple;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MetaDataTuple implements Tuple {
+ List<Datum> values = new ArrayList<Datum>();
+
+ public MetaDataTuple(int size) {
+ values = new ArrayList<Datum>(size);
+ for(int i = 0; i < size; i++) {
+ values.add(NullDatum.get());
+ }
+ }
+
+ @Override
+ public int size() {
+ return values.size();
+ }
+
+ @Override
+ public boolean contains(int fieldid) {
+ return false;
+ }
+
+ @Override
+ public boolean isNull(int fieldid) {
+ return values.get(fieldid) == null || values.get(fieldid) instanceof NullDatum;
+ }
+
+ @Override
+ public void clear() {
+ values.clear();
+ }
+
+ @Override
+ public void put(int fieldId, Datum value) {
+ values.set(fieldId, value);
+ }
+
+ @Override
+ public void put(int fieldId, Datum[] values) {
+ throw new UnsupportedException("put");
+ }
+
+ @Override
+ public void put(int fieldId, Tuple tuple) {
+ throw new UnsupportedException("put");
+ }
+
+ @Override
+ public void put(Datum[] values) {
+ throw new UnsupportedException("put");
+ }
+
+ @Override
+ public Datum get(int fieldId) {
+ return getText(fieldId);
+ }
+
+ @Override
+ public void setOffset(long offset) {
+ throw new UnsupportedException("setOffset");
+ }
+
+ @Override
+ public long getOffset() {
+ throw new UnsupportedException("getOffset");
+ }
+
+ @Override
+ public BooleanDatum getBoolean(int fieldId) {
+ throw new UnsupportedException("getBoolean");
+ }
+
+ @Override
+ public BitDatum getByte(int fieldId) {
+ throw new UnsupportedException("getByte");
+ }
+
+ @Override
+ public CharDatum getChar(int fieldId) {
+ throw new UnsupportedException("getBoolean");
+ }
+
+ @Override
+ public BlobDatum getBytes(int fieldId) {
+ throw new UnsupportedException("BlobDatum");
+ }
+
+ @Override
+ public Int2Datum getShort(int fieldId) {
+ if(isNull(fieldId)) {
+ return null;
+ }
+ return new Int2Datum((short)Integer.parseInt(values.get(fieldId).toString()));
+ }
+
+ @Override
+ public Int4Datum getInt(int fieldId) {
+ if(isNull(fieldId)) {
+ return null;
+ }
+ return new Int4Datum(Integer.parseInt(values.get(fieldId).toString()));
+ }
+
+ @Override
+ public Int8Datum getLong(int fieldId) {
+ if(isNull(fieldId)) {
+ return null;
+ }
+ return new Int8Datum(Long.parseLong(values.get(fieldId).toString()));
+ }
+
+ @Override
+ public Float4Datum getFloat(int fieldId) {
+ if(isNull(fieldId)) {
+ return null;
+ }
+ return new Float4Datum(Float.parseFloat(values.get(fieldId).toString()));
+ }
+
+ @Override
+ public Float8Datum getDouble(int fieldId) {
+ if(isNull(fieldId)) {
+ return null;
+ }
+ return new Float8Datum(Float.parseFloat(values.get(fieldId).toString()));
+ }
+
+ @Override
+ public Inet4Datum getIPv4(int fieldId) {
+ throw new UnsupportedException("getIPv4");
+ }
+
+ @Override
+ public byte[] getIPv4Bytes(int fieldId) {
+ throw new UnsupportedException("getIPv4Bytes");
+ }
+
+ @Override
+ public InetAddress getIPv6(int fieldId) {
+ throw new UnsupportedException("getIPv6");
+ }
+
+ @Override
+ public byte[] getIPv6Bytes(int fieldId) {
+ throw new UnsupportedException("getIPv6Bytes");
+ }
+
+ @Override
+ public TextDatum getString(int fieldId) {
+ if(isNull(fieldId)) {
+ return null;
+ }
+ return new TextDatum(values.get(fieldId).toString());
+ }
+
+ @Override
+ public TextDatum getText(int fieldId) {
+ if(isNull(fieldId)) {
+ return null;
+ }
+ return new TextDatum(values.get(fieldId).toString());
+ }
+
+ @Override
+ public Tuple clone() throws CloneNotSupportedException {
+ throw new UnsupportedException("clone");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoConnection.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoConnection.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoConnection.java
new file mode 100644
index 0000000..47c88d1
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoConnection.java
@@ -0,0 +1,400 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.conf.TajoConf;
+
+import java.sql.*;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TajoConnection implements Connection {
+ private TajoClient tajoClient;
+
+ private String databaseName;
+
+ private AtomicBoolean closed = new AtomicBoolean(true);
+
+ private String uri;
+
+ public TajoConnection(String uri, Properties properties) throws SQLException {
+ if (!uri.startsWith(TajoDriver.TAJO_JDBC_URL_PREFIX)) {
+ throw new SQLException("Invalid URL: " + uri, "TAJO-001");
+ }
+
+ this.uri = uri;
+
+ // remove prefix
+ uri = uri.substring(TajoDriver.TAJO_JDBC_URL_PREFIX.length());
+
+
+ if (uri.isEmpty()) {
+ throw new SQLException("Invalid URL: " + uri, "TAJO-001");
+ }
+
+ // parse uri
+ // form: hostname:port/databasename
+ String[] parts = uri.split("/");
+ if(parts.length == 0 || parts[0].trim().isEmpty()) {
+ throw new SQLException("Invalid URL(No tajo master's host:port): " + uri, "TAJO-001");
+ }
+ String[] hostAndPort = parts[0].trim().split(":");
+ String host = hostAndPort[0];
+ int port = 0;
+ try {
+ port = Integer.parseInt(hostAndPort[1]);
+ } catch (Exception e) {
+ throw new SQLException("Invalid URL(Wrong tajo master's host:port): " + uri, "TAJO-001");
+ }
+
+ if(parts.length > 1) {
+ String[] tokens = parts[1].split("\\?");
+ databaseName = tokens[0].trim();
+ if(tokens.length > 1) {
+ String[] extraParamTokens = tokens[1].split("&");
+ for(String eachExtraParam: extraParamTokens) {
+ String[] paramTokens = eachExtraParam.split("=");
+ String extraParamKey = paramTokens[0];
+ String extraParamValue = paramTokens[1];
+ }
+ }
+ }
+
+ TajoConf tajoConf = new TajoConf();
+
+ tajoConf.setVar(TajoConf.ConfVars.TAJO_MASTER_CLIENT_RPC_ADDRESS, host + ":" + port);
+
+ if(properties != null) {
+ for(Map.Entry<Object, Object> entry: properties.entrySet()) {
+ tajoConf.set(entry.getKey().toString(), entry.getValue().toString());
+ }
+ }
+
+ try {
+ tajoClient = new TajoClient(tajoConf);
+ } catch (Exception e) {
+ throw new SQLException("Can't create tajo client:" + e.getMessage(), "TAJO-002");
+ }
+ closed.set(false);
+ }
+
+ public String getUri() {
+ return uri;
+ }
+
+ public TajoClient getTajoClient() {
+ return tajoClient;
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ }
+
+ @Override
+ public void close() throws SQLException {
+ if(!closed.get()) {
+ if(tajoClient != null) {
+ tajoClient.close();
+ }
+
+ closed.set(true);
+ }
+ }
+
+ @Override
+ public void commit() throws SQLException {
+ throw new SQLFeatureNotSupportedException("commit");
+ }
+
+ @Override
+ public Array createArrayOf(String arg0, Object[] arg1) throws SQLException {
+ throw new SQLFeatureNotSupportedException("createArrayOf");
+ }
+
+ @Override
+ public Blob createBlob() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createBlob");
+ }
+
+ @Override
+ public Clob createClob() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createClob");
+ }
+
+ @Override
+ public NClob createNClob() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createNClob");
+ }
+
+ @Override
+ public SQLXML createSQLXML() throws SQLException {
+ throw new SQLFeatureNotSupportedException("createSQLXML");
+ }
+
+ @Override
+ public Statement createStatement() throws SQLException {
+ if (isClosed()) {
+ throw new SQLException("Can't create Statement, connection is closed");
+ }
+ return new TajoStatement(tajoClient);
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("createStatement");
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("createStatement");
+ }
+
+ @Override
+ public Struct createStruct(String typeName, Object[] attributes)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("createStruct");
+ }
+
+ @Override
+ public boolean getAutoCommit() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public String getCatalog() throws SQLException {
+ return "";
+ }
+
+ @Override
+ public Properties getClientInfo() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getClientInfo");
+ }
+
+ @Override
+ public String getClientInfo(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getClientInfo");
+ }
+
+ @Override
+ public int getHoldability() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getHoldability");
+ }
+
+ @Override
+ public DatabaseMetaData getMetaData() throws SQLException {
+ return new TajoDatabaseMetaData(this);
+ }
+
+ @Override
+ public int getTransactionIsolation() throws SQLException {
+ return Connection.TRANSACTION_NONE;
+ }
+
+ @Override
+ public Map<String, Class<?>> getTypeMap() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getTypeMap");
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getWarnings");
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return closed.get();
+ }
+
+ @Override
+ public boolean isReadOnly() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean isValid(int timeout) throws SQLException {
+ throw new SQLFeatureNotSupportedException("isValid");
+ }
+
+ @Override
+ public String nativeSQL(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("nativeSQL");
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareCall");
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType,
+ int resultSetConcurrency) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareCall");
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType,
+ int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareCall");
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql) throws SQLException {
+ return new TajoPreparedStatement(tajoClient, sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys)
+ throws SQLException {
+ return new TajoPreparedStatement(tajoClient, sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int[] columnIndexes)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareStatement");
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, String[] columnNames)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareStatement");
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType,
+ int resultSetConcurrency) throws SQLException {
+ return new TajoPreparedStatement(tajoClient, sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType,
+ int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("prepareStatement");
+ }
+
+ @Override
+ public void releaseSavepoint(Savepoint savepoint) throws SQLException {
+ throw new SQLFeatureNotSupportedException("releaseSavepoint");
+ }
+
+ @Override
+ public void rollback() throws SQLException {
+ throw new SQLFeatureNotSupportedException("rollback");
+ }
+
+ @Override
+ public void rollback(Savepoint savepoint) throws SQLException {
+ throw new SQLFeatureNotSupportedException("rollback");
+ }
+
+ @Override
+ public void setAutoCommit(boolean autoCommit) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setAutoCommit");
+ }
+
+ @Override
+ public void setCatalog(String catalog) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setCatalog");
+ }
+
+ @Override
+ public void setClientInfo(Properties properties)
+ throws SQLClientInfoException {
+ throw new UnsupportedOperationException("setClientInfo");
+ }
+
+ @Override
+ public void setClientInfo(String name, String value)
+ throws SQLClientInfoException {
+ throw new UnsupportedOperationException("setClientInfo");
+ }
+
+ @Override
+ public void setHoldability(int holdability) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setHoldability");
+ }
+
+ @Override
+ public void setReadOnly(boolean readOnly) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setReadOnly");
+ }
+
+ @Override
+ public Savepoint setSavepoint() throws SQLException {
+ throw new SQLFeatureNotSupportedException("setSavepoint");
+ }
+
+ @Override
+ public Savepoint setSavepoint(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setSavepoint");
+ }
+
+ @Override
+ public void setTransactionIsolation(int level) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTransactionIsolation");
+ }
+
+ @Override
+ public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTypeMap");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> tClass) throws SQLException {
+ if (isWrapperFor(tClass)) {
+ return (T) this;
+ }
+ throw new SQLException("No wrapper for " + tClass);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> tClass) throws SQLException {
+ return tClass.isInstance(this);
+ }
+
+ public void abort(Executor executor) throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("abort not supported");
+ }
+
+ public int getNetworkTimeout() throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("getNetworkTimeout not supported");
+ }
+
+ public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("setNetworkTimeout not supported");
+ }
+
+ public String getSchema() throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("getSchema not supported");
+ }
+
+ public void setSchema(String schema) throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("setSchema not supported");
+ }
+}
[2/3] TAJO-176: Implement Tajo JDBC Driver. (Keuntae Park via jihoon)
Posted by ji...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java
new file mode 100644
index 0000000..6c64ac2
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDatabaseMetaData.java
@@ -0,0 +1,1196 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.TajoConstants;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.datum.TextDatum;
+
+import java.sql.*;
+import java.util.*;
+
+/**
+ * TajoDatabaseMetaData.
+ */
+public class TajoDatabaseMetaData implements DatabaseMetaData {
+ private static final char SEARCH_STRING_ESCAPE = '\\';
+
+ private final TajoConnection conn;
+
+ public TajoDatabaseMetaData(TajoConnection conn) {
+ this.conn = conn;
+ }
+
+ @Override
+ public boolean allProceduresAreCallable()
+ throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean allTablesAreSelectable()
+ throws SQLException {
+ return true;
+ }
+
+ @Override
+ public String getURL()
+ throws SQLException {
+ return conn.getUri();
+ }
+
+ @Override
+ public String getUserName()
+ throws SQLException {
+ return "tajo";
+ }
+
+ @Override
+ public boolean isReadOnly()
+ throws SQLException {
+ return true;
+ }
+
+ @Override
+ public String getDatabaseProductName()
+ throws SQLException {
+ return "Tajo";
+ }
+
+ @Override
+ public String getDatabaseProductVersion()
+ throws SQLException {
+ //TODO get from tajo master
+ return TajoConstants.TAJO_VERSION;
+ }
+
+ @Override
+ public String getDriverName()
+ throws SQLException {
+ return "tajo";
+ }
+
+ @Override
+ public String getDriverVersion()
+ throws SQLException {
+ return TajoDriver.MAJOR_VERSION + "." + TajoDriver.MINOR_VERSION;
+ }
+
+ @Override
+ public int getDriverMajorVersion() {
+ return TajoDriver.MAJOR_VERSION;
+ }
+
+ @Override
+ public int getDriverMinorVersion() {
+ return TajoDriver.MINOR_VERSION;
+ }
+
+ @Override
+ public String getIdentifierQuoteString()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getIdentifierQuoteString not supported");
+ }
+
+ @Override
+ public String getSQLKeywords()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getSQLKeywords not supported");
+ }
+
+ @Override
+ public String getNumericFunctions()
+ throws SQLException {
+ return "";
+ }
+
+ @Override
+ public String getStringFunctions()
+ throws SQLException {
+ return "";
+ }
+
+ @Override
+ public String getSystemFunctions()
+ throws SQLException {
+ return "";
+ }
+
+ @Override
+ public String getTimeDateFunctions()
+ throws SQLException {
+ return "";
+ }
+
+ @Override
+ public String getSearchStringEscape()
+ throws SQLException {
+ return "\\";
+ }
+
+ @Override
+ public String getExtraNameCharacters()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getExtraNameCharacters not supported");
+ }
+
+ @Override
+ public String getSchemaTerm()
+ throws SQLException {
+ return "";
+ }
+
+ @Override
+ public String getProcedureTerm()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getProcedureTerm not supported");
+ }
+
+ @Override
+ public String getCatalogTerm()
+ throws SQLException {
+ return "database";
+ }
+
+ @Override
+ public String getCatalogSeparator()
+ throws SQLException {
+ return ".";
+ }
+
+ @Override
+ public int getMaxBinaryLiteralLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxBinaryLiteralLength not supported");
+ }
+
+ @Override
+ public int getMaxCharLiteralLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxCharLiteralLength not supported");
+ }
+
+ @Override
+ public int getMaxColumnNameLength()
+ throws SQLException {
+ return 128;
+ }
+
+ @Override
+ public int getMaxColumnsInGroupBy()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxColumnsInGroupBy not supported");
+ }
+
+ @Override
+ public int getMaxColumnsInIndex()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxColumnsInIndex not supported");
+ }
+
+ @Override
+ public int getMaxColumnsInOrderBy()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxColumnsInOrderBy not supported");
+ }
+
+ @Override
+ public int getMaxColumnsInSelect()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxColumnsInSelect not supported");
+ }
+
+ @Override
+ public int getMaxColumnsInTable()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxColumnsInTable not supported");
+ }
+
+ @Override
+ public int getMaxConnections()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxConnections not supported");
+ }
+
+ @Override
+ public int getMaxCursorNameLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxCursorNameLength not supported");
+ }
+
+ @Override
+ public int getMaxIndexLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxIndexLength not supported");
+ }
+
+ @Override
+ public int getMaxSchemaNameLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxSchemaNameLength not supported");
+ }
+
+ @Override
+ public int getMaxProcedureNameLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxProcedureNameLength not supported");
+ }
+
+ @Override
+ public int getMaxCatalogNameLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxCatalogNameLength not supported");
+ }
+
+ @Override
+ public int getMaxRowSize()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxRowSize not supported");
+ }
+
+ @Override
+ public boolean doesMaxRowSizeIncludeBlobs()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("doesMaxRowSizeIncludeBlobs not supported");
+ }
+
+ @Override
+ public int getMaxStatementLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxStatementLength not supported");
+ }
+
+ @Override
+ public int getMaxStatements()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxStatements not supported");
+ }
+
+ @Override
+ public int getMaxTableNameLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxTableNameLength not supported");
+ }
+
+ @Override
+ public int getMaxTablesInSelect()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxTablesInSelect not supported");
+ }
+
+ @Override
+ public int getMaxUserNameLength()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxUserNameLength not supported");
+ }
+
+ @Override
+ public int getDefaultTransactionIsolation()
+ throws SQLException {
+ return Connection.TRANSACTION_NONE;
+ }
+
+ @Override
+ public boolean dataDefinitionCausesTransactionCommit()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("dataDefinitionCausesTransactionCommit not supported");
+ }
+
+ @Override
+ public boolean dataDefinitionIgnoredInTransactions()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("dataDefinitionIgnoredInTransactions not supported");
+ }
+
+ @Override
+ public ResultSet getProcedures(String catalog, String schemaPattern, String procedureNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("stored procedures not supported");
+ }
+
+ @Override
+ public ResultSet getProcedureColumns(String catalog, String schemaPattern, String procedureNamePattern, String columnNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("stored procedures not supported");
+ }
+
+ /**
+ * Convert a pattern containing JDBC catalog search wildcards into
+ * Java regex patterns.
+ *
+ * @param pattern input which may contain '%' or '_' wildcard characters, or
+ * these characters escaped using {@link #getSearchStringEscape()}.
+ * @return replace %/_ with regex search characters, also handle escaped
+ * characters.
+ */
+ private String convertPattern(final String pattern) {
+ if (pattern == null) {
+ return ".*";
+ } else {
+ StringBuilder result = new StringBuilder(pattern.length());
+
+ boolean escaped = false;
+ for (int i = 0, len = pattern.length(); i < len; i++) {
+ char c = pattern.charAt(i);
+ if (escaped) {
+ if (c != SEARCH_STRING_ESCAPE) {
+ escaped = false;
+ }
+ result.append(c);
+ } else {
+ if (c == SEARCH_STRING_ESCAPE) {
+ escaped = true;
+ continue;
+ } else if (c == '%') {
+ result.append(".*");
+ } else if (c == '_') {
+ result.append('.');
+ } else {
+ result.append(Character.toLowerCase(c));
+ }
+ }
+ }
+
+ return result.toString();
+ }
+ }
+
+ @Override
+ public ResultSet getTables(String catalog, String schemaPattern, String tableNamePattern, String[] types)
+ throws SQLException {
+ try {
+ final List<MetaDataTuple> resultTables = new ArrayList<MetaDataTuple>();
+ final String resultCatalog;
+ if (catalog == null) {
+ resultCatalog = "default";
+ } else {
+ resultCatalog = catalog;
+ }
+
+ String regtableNamePattern = convertPattern(tableNamePattern);
+ try {
+ TajoClient tajoClient = conn.getTajoClient();
+ List<String> tableNames = tajoClient.getTableList();
+ for (String eachTableName: tableNames) {
+ if (eachTableName.matches(regtableNamePattern)) {
+ MetaDataTuple tuple = new MetaDataTuple(5);
+
+ int index = 0;
+ tuple.put(index++, new TextDatum(resultCatalog)); //TABLE_CAT
+ tuple.put(index++, NullDatum.get()); //TABLE_SCHEM
+ tuple.put(index++, new TextDatum(eachTableName));
+ tuple.put(index++, new TextDatum("TABLE")); //TABLE_TYPE
+ tuple.put(index++, NullDatum.get()); //REMARKS
+
+ resultTables.add(tuple);
+ }
+ }
+ Collections.sort(resultTables, new Comparator<MetaDataTuple> () {
+ @Override
+ public int compare(MetaDataTuple table1, MetaDataTuple table2) {
+ return table1.getString(2).compareTo(table2.getString(2));
+ }
+ });
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new SQLException(e);
+ }
+ TajoMetaDataResultSet result = new TajoMetaDataResultSet(
+ Arrays.asList("TABLE_CAT", "TABLE_SCHEM", "TABLE_NAME", "TABLE_TYPE", "REMARKS"),
+ Arrays.asList(Type.VARCHAR, Type.VARCHAR, Type.VARCHAR, Type.VARCHAR, Type.VARCHAR),
+ resultTables);
+
+ return result;
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new SQLException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public ResultSet getSchemas()
+ throws SQLException {
+ return getSchemas(null, null);
+ }
+
+ @Override
+ public ResultSet getCatalogs()
+ throws SQLException {
+ List<MetaDataTuple> columns = new ArrayList<MetaDataTuple>();
+ MetaDataTuple tuple = new MetaDataTuple(1);
+ tuple.put(0, new TextDatum("default"));
+ columns.add(tuple);
+
+ ResultSet result = new TajoMetaDataResultSet(
+ Arrays.asList("TABLE_CAT")
+ , Arrays.asList(Type.VARCHAR)
+ , columns);
+
+ return result;
+ }
+
+ @Override
+ public ResultSet getTableTypes()
+ throws SQLException {
+ List<MetaDataTuple> columns = new ArrayList<MetaDataTuple>();
+ MetaDataTuple tuple = new MetaDataTuple(1);
+ tuple.put(0, new TextDatum("TABLE"));
+ columns.add(tuple);
+
+ ResultSet result = new TajoMetaDataResultSet(
+ Arrays.asList("TABLE_TYPE")
+ , Arrays.asList(Type.VARCHAR)
+ , columns);
+
+ return result;
+ }
+
+ @Override
+ public ResultSet getUDTs(String catalog, String schemaPattern, String typeNamePattern, int[] types)
+ throws SQLException {
+ List<MetaDataTuple> columns = new ArrayList<MetaDataTuple>();
+
+ return new TajoMetaDataResultSet(
+ Arrays.asList("TYPE_CAT", "TYPE_SCHEM", "TYPE_NAME", "CLASS_NAME", "DATA_TYPE"
+ , "REMARKS", "BASE_TYPE")
+ , Arrays.asList(Type.VARCHAR, Type.VARCHAR, Type.VARCHAR, Type.VARCHAR, Type.INT4, Type.VARCHAR, Type.INT4)
+ , columns);
+ }
+
+ @Override
+ public ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern)
+ throws SQLException {
+ List<MetaDataTuple> columns = new ArrayList<MetaDataTuple>();
+ try {
+ if (catalog == null) {
+ catalog = "default";
+ }
+
+ String regtableNamePattern = convertPattern(tableNamePattern);
+ String regcolumnNamePattern = convertPattern(columnNamePattern);
+
+ List<String> tables = conn.getTajoClient().getTableList();
+ for (String table: tables) {
+ if (table.matches(regtableNamePattern)) {
+ TableDesc tableDesc = conn.getTajoClient().getTableDesc(table);
+ int pos = 0;
+ for (Column column: tableDesc.getSchema().getColumns()) {
+ if (column.getColumnName().matches(regcolumnNamePattern)) {
+ MetaDataTuple tuple = new MetaDataTuple(22);
+
+ int index = 0;
+ tuple.put(index++, new TextDatum(catalog)); //TABLE_CAT
+ tuple.put(index++, NullDatum.get()); //TABLE_SCHEM
+ tuple.put(index++, new TextDatum(table)); //TABLE_NAME
+ tuple.put(index++, new TextDatum(column.getColumnName())); //COLUMN_NAME
+ tuple.put(index++, new TextDatum("" + TajoDriver.tajoTypeToSqlType(column.getDataType()))); //TODO DATA_TYPE
+ tuple.put(index++, new TextDatum(TajoDriver.toSqlType(column.getDataType()))); //TYPE_NAME
+ tuple.put(index++, new TextDatum("0")); //COLUMN_SIZE
+ tuple.put(index++, new TextDatum("0")); //BUFFER_LENGTH
+ tuple.put(index++, new TextDatum("0")); //DECIMAL_DIGITS
+ tuple.put(index++, new TextDatum("0")); //NUM_PREC_RADIX
+ tuple.put(index++, new TextDatum("" + DatabaseMetaData.columnNullable)); //NULLABLE
+ tuple.put(index++, NullDatum.get()); //REMARKS
+ tuple.put(index++, NullDatum.get()); //COLUMN_DEF
+ tuple.put(index++, NullDatum.get()); //SQL_DATA_TYPE
+ tuple.put(index++, NullDatum.get()); //SQL_DATETIME_SUB
+ tuple.put(index++, new TextDatum("0")); //CHAR_OCTET_LENGTH
+ tuple.put(index++, new TextDatum("" + pos)); //ORDINAL_POSITION
+ tuple.put(index++, new TextDatum("YES")); //IS_NULLABLE
+ tuple.put(index++, NullDatum.get()); //SCOPE_CATLOG
+ tuple.put(index++, NullDatum.get()); //SCOPE_SCHEMA
+ tuple.put(index++, NullDatum.get()); //SCOPE_TABLE
+ tuple.put(index++, new TextDatum("0")); //SOURCE_DATA_TYPE
+ columns.add(tuple);
+ }
+ pos++;
+ }
+ }
+ }
+
+ return new TajoMetaDataResultSet(
+ Arrays.asList("TABLE_CAT", "TABLE_SCHEM", "TABLE_NAME", "COLUMN_NAME", "DATA_TYPE"
+ , "TYPE_NAME", "COLUMN_SIZE", "BUFFER_LENGTH", "DECIMAL_DIGITS", "NUM_PREC_RADIX"
+ , "NULLABLE", "REMARKS", "COLUMN_DEF", "SQL_DATA_TYPE", "SQL_DATETIME_SUB"
+ , "CHAR_OCTET_LENGTH", "ORDINAL_POSITION", "IS_NULLABLE", "SCOPE_CATLOG", "SCOPE_SCHEMA"
+ , "SCOPE_TABLE", "SOURCE_DATA_TYPE")
+ , Arrays.asList(Type.VARCHAR, Type.VARCHAR, Type.VARCHAR, Type.VARCHAR, Type.INT4
+ , Type.VARCHAR, Type.INT4, Type.INT4, Type.INT4, Type.INT4
+ , Type.INT4, Type.VARCHAR, Type.VARCHAR, Type.INT4, Type.INT4
+ , Type.INT4, Type.INT4, Type.VARCHAR, Type.VARCHAR, Type.VARCHAR
+ , Type.VARCHAR, Type.INT4)
+ , columns);
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new SQLException(e);
+ }
+ }
+
+ @Override
+ public ResultSet getColumnPrivileges(String catalog, String schema, String table, String columnNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("privileges not supported");
+ }
+
+ @Override
+ public ResultSet getTablePrivileges(String catalog, String schemaPattern, String tableNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("privileges not supported");
+ }
+
+ @Override
+ public ResultSet getBestRowIdentifier(String catalog, String schema, String table, int scope, boolean nullable)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("row identifiers not supported");
+ }
+
+ @Override
+ public ResultSet getVersionColumns(String catalog, String schema, String table)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("version columns not supported");
+ }
+
+ @Override
+ public ResultSet getPrimaryKeys(String catalog, String schema, String table)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("primary keys not supported");
+ }
+
+ @Override
+ public ResultSet getImportedKeys(String catalog, String schema, String table)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("imported keys not supported");
+ }
+
+ @Override
+ public ResultSet getExportedKeys(String catalog, String schema, String table)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("exported keys not supported");
+ }
+
+ @Override
+ public ResultSet getCrossReference(String parentCatalog, String parentSchema, String parentTable, String foreignCatalog, String foreignSchema, String foreignTable)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("cross reference not supported");
+ }
+
+ @Override
+ public ResultSet getTypeInfo()
+ throws SQLException {
+ throw new UnsupportedOperationException("getTypeInfo not supported");
+ }
+
+ @Override
+ public ResultSet getIndexInfo(String catalog, String schema, String table, boolean unique, boolean approximate)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("indexes not supported");
+ }
+
+ @Override
+ public boolean deletesAreDetected(int type)
+ throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean insertsAreDetected(int type)
+ throws SQLException {
+ return false;
+ }
+
+ @Override
+ public Connection getConnection()
+ throws SQLException {
+ return conn;
+ }
+
+ @Override
+ public ResultSet getSuperTypes(String catalog, String schemaPattern, String typeNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("type hierarchies not supported");
+ }
+
+ @Override
+ public ResultSet getSuperTables(String catalog, String schemaPattern, String tableNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("type hierarchies not supported");
+ }
+
+ @Override
+ public ResultSet getAttributes(String catalog, String schemaPattern, String typeNamePattern, String attributeNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("user-defined types not supported");
+ }
+
+ @Override
+ public int getResultSetHoldability()
+ throws SQLException {
+ return ResultSet.HOLD_CURSORS_OVER_COMMIT;
+ }
+
+ @Override
+ public int getDatabaseMajorVersion()
+ throws SQLException {
+ return TajoDriver.MAJOR_VERSION;
+ }
+
+ @Override
+ public int getDatabaseMinorVersion()
+ throws SQLException {
+ return TajoDriver.MINOR_VERSION;
+ }
+
+ @Override
+ public int getJDBCMajorVersion()
+ throws SQLException {
+ return TajoDriver.JDBC_VERSION_MAJOR;
+ }
+
+ @Override
+ public int getJDBCMinorVersion()
+ throws SQLException {
+ return TajoDriver.JDBC_VERSION_MINOR;
+ }
+
+ @Override
+ public int getSQLStateType()
+ throws SQLException {
+ return DatabaseMetaData.sqlStateSQL;
+ }
+
+ @Override
+ public RowIdLifetime getRowIdLifetime()
+ throws SQLException {
+ return RowIdLifetime.ROWID_UNSUPPORTED;
+ }
+
+ @Override
+ public ResultSet getSchemas(String catalog, String schemaPattern)
+ throws SQLException {
+ return new TajoMetaDataResultSet(
+ Arrays.asList("TABLE_SCHEM", "TABLE_CATALOG"),
+ Arrays.asList(Type.VARCHAR, Type.VARCHAR),
+ null);
+ }
+
+ @Override
+ public boolean autoCommitFailureClosesAllResultSets()
+ throws SQLException {
+ return false;
+ }
+
+ @Override
+ public ResultSet getClientInfoProperties()
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getClientInfoProperties not supported");
+ }
+
+ @Override
+ public ResultSet getFunctions(String catalog, String schemaPattern, String functionNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getFunctions not supported");
+ }
+
+ @Override
+ public ResultSet getFunctionColumns(String catalog, String schemaPattern, String functionNamePattern, String columnNamePattern)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("getFunctionColumns not supported");
+ }
+
+ @Override
+ public boolean isCatalogAtStart() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean locatorsUpdateCopy() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean nullPlusNonNullIsNull() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean nullsAreSortedAtEnd() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean nullsAreSortedAtStart() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean nullsAreSortedHigh() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean nullsAreSortedLow() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean othersDeletesAreVisible(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean othersInsertsAreVisible(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean othersUpdatesAreVisible(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean ownDeletesAreVisible(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean ownInsertsAreVisible(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean ownUpdatesAreVisible(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean storesLowerCaseIdentifiers() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean storesLowerCaseQuotedIdentifiers() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean storesMixedCaseIdentifiers() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean storesMixedCaseQuotedIdentifiers() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean storesUpperCaseIdentifiers() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean storesUpperCaseQuotedIdentifiers() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsANSI92EntryLevelSQL() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsANSI92FullSQL() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsANSI92IntermediateSQL() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsAlterTableWithAddColumn() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsAlterTableWithDropColumn() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsBatchUpdates() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsCatalogsInDataManipulation() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsCatalogsInIndexDefinitions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsCatalogsInPrivilegeDefinitions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsCatalogsInProcedureCalls() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsCatalogsInTableDefinitions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsColumnAliasing() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean supportsConvert() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsConvert(int fromType, int toType) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsCoreSQLGrammar() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsCorrelatedSubqueries() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsDataDefinitionAndDataManipulationTransactions()
+ throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsDataManipulationTransactionsOnly() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsDifferentTableCorrelationNames() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsExpressionsInOrderBy() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsExtendedSQLGrammar() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsFullOuterJoins() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsGetGeneratedKeys() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsGroupBy() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean supportsGroupByBeyondSelect() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsGroupByUnrelated() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsIntegrityEnhancementFacility() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsLikeEscapeClause() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsLimitedOuterJoins() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMinimumSQLGrammar() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMixedCaseIdentifiers() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMixedCaseQuotedIdentifiers() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMultipleOpenResults() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMultipleResultSets() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsMultipleTransactions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsNamedParameters() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsNonNullableColumns() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsOpenCursorsAcrossCommit() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsOpenCursorsAcrossRollback() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsOpenStatementsAcrossCommit() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsOpenStatementsAcrossRollback() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsOrderByUnrelated() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsOuterJoins() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsPositionedDelete() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsPositionedUpdate() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsResultSetConcurrency(int type, int concurrency)
+ throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsResultSetHoldability(int holdability)
+ throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsResultSetType(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSavepoints() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSchemasInDataManipulation() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSchemasInIndexDefinitions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSchemasInPrivilegeDefinitions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSchemasInProcedureCalls() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSchemasInTableDefinitions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSelectForUpdate() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsStatementPooling() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsStoredFunctionsUsingCallSyntax() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsStoredProcedures() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSubqueriesInComparisons() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSubqueriesInExists() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSubqueriesInIns() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsSubqueriesInQuantifieds() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsTableCorrelationNames() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsTransactionIsolationLevel(int level)
+ throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsTransactions() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsUnion() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean supportsUnionAll() throws SQLException {
+ return true;
+ }
+
+ @Override
+ public boolean updatesAreDetected(int type) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean usesLocalFilePerTable() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean usesLocalFiles() throws SQLException {
+ return false;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> T unwrap(Class<T> iface)
+ throws SQLException {
+ if (isWrapperFor(iface)) {
+ return (T) this;
+ }
+ throw new SQLFeatureNotSupportedException("No wrapper for " + iface);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface)
+ throws SQLException {
+ return iface.isInstance(this);
+ }
+
+ public boolean generatedKeyAlwaysReturned() throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("generatedKeyAlwaysReturned not supported");
+ }
+
+ public ResultSet getPseudoColumns(String catalog, String schemaPattern,
+ String tableNamePattern, String columnNamePattern) throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("getPseudoColumns not supported");
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDriver.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDriver.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDriver.java
new file mode 100644
index 0000000..ca0502f
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoDriver.java
@@ -0,0 +1,233 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.exception.UnsupportedException;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.sql.*;
+import java.util.Properties;
+import java.util.logging.Logger;
+
+public class TajoDriver implements Driver, Closeable {
+ public static final int MAJOR_VERSION = 1;
+ public static final int MINOR_VERSION = 0;
+
+ public static final int JDBC_VERSION_MAJOR = 4;
+ public static final int JDBC_VERSION_MINOR = 0;
+
+ public static final String TAJO_JDBC_URL_PREFIX = "jdbc:tajo://";
+
+ protected static TajoConf jdbcTajoConf;
+
+ static {
+ try {
+ java.sql.DriverManager.registerDriver(new TajoDriver());
+ } catch (SQLException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+ public TajoDriver() {
+ jdbcTajoConf = new TajoConf();
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+
+ @Override
+ public Connection connect(String url, Properties properties) throws SQLException {
+ return new TajoConnection(url, properties);
+ }
+
+ @Override
+ public boolean acceptsURL(String url) throws SQLException {
+ return url.startsWith(TAJO_JDBC_URL_PREFIX);
+ }
+
+ @Override
+ public DriverPropertyInfo[] getPropertyInfo(String s, Properties properties) throws SQLException {
+ return new DriverPropertyInfo[0];
+ }
+
+ @Override
+ public int getMajorVersion() {
+ return MAJOR_VERSION;
+ }
+
+ @Override
+ public int getMinorVersion() {
+ return MINOR_VERSION;
+ }
+
+ @Override
+ public boolean jdbcCompliant() {
+ return false;
+ }
+
+ public static String toSqlType(TajoDataTypes.DataType type) {
+ switch (type.getType()) {
+ case BOOLEAN:
+ return "boolean";
+ case INT1:
+ return "tinyint";
+ case INT2:
+ return "smallint";
+ case INT4:
+ return "integer";
+ case INT8:
+ return "bigint";
+ case FLOAT4:
+ return "float";
+ case FLOAT8:
+ return "float8";
+ case DECIMAL:
+ return "numeric";
+ case VARBINARY:
+ return "bytea";
+ case CHAR:
+ return "character";
+ case DATE:
+ return "date";
+ case VARCHAR:
+ return "varchar";
+ case TEXT:
+ return "varchar";
+ default:
+ throw new UnsupportedException("Unrecognized column type:" + type);
+ }
+ }
+
+ public static int tajoTypeToSqlType(TajoDataTypes.DataType type) throws SQLException {
+ switch (type.getType()) {
+ case BOOLEAN:
+ return Types.BOOLEAN;
+ case INT1:
+ return Types.TINYINT;
+ case INT2:
+ return Types.SMALLINT;
+ case INT4:
+ return Types.INTEGER;
+ case INT8:
+ return Types.BIGINT;
+ case FLOAT4:
+ return Types.FLOAT;
+ case FLOAT8:
+ return Types.DOUBLE;
+ case DECIMAL:
+ return Types.DECIMAL;
+ case DATE:
+ return Types.TIMESTAMP;
+ case VARCHAR:
+ return Types.VARCHAR;
+ case TEXT:
+ return Types.VARCHAR;
+ default:
+ throw new SQLException("Unrecognized column type: " + type);
+ }
+ }
+
+ static int columnDisplaySize(int columnType) throws SQLException {
+ // according to hiveTypeToSqlType possible options are:
+ switch(columnType) {
+ case Types.BOOLEAN:
+ return columnPrecision(columnType);
+ case Types.VARCHAR:
+ return Integer.MAX_VALUE; // hive has no max limit for strings
+ case Types.TINYINT:
+ case Types.SMALLINT:
+ case Types.INTEGER:
+ case Types.BIGINT:
+ return columnPrecision(columnType) + 1; // allow +/-
+ case Types.TIMESTAMP:
+ return columnPrecision(columnType);
+ // see http://download.oracle.com/javase/6/docs/api/constant-values.html#java.lang.Float.MAX_EXPONENT
+ case Types.FLOAT:
+ return 24; // e.g. -(17#).e-###
+ // see http://download.oracle.com/javase/6/docs/api/constant-values.html#java.lang.Double.MAX_EXPONENT
+ case Types.DOUBLE:
+ return 25; // e.g. -(17#).e-####
+ case Types.DECIMAL:
+ return Integer.MAX_VALUE;
+ default:
+ throw new SQLException("Invalid column type: " + columnType);
+ }
+ }
+
+ static int columnPrecision(int columnType) throws SQLException {
+ // according to hiveTypeToSqlType possible options are:
+ switch(columnType) {
+ case Types.BOOLEAN:
+ return 1;
+ case Types.VARCHAR:
+ return Integer.MAX_VALUE; // hive has no max limit for strings
+ case Types.TINYINT:
+ return 3;
+ case Types.SMALLINT:
+ return 5;
+ case Types.INTEGER:
+ return 10;
+ case Types.BIGINT:
+ return 19;
+ case Types.FLOAT:
+ return 7;
+ case Types.DOUBLE:
+ return 15;
+ case Types.TIMESTAMP:
+ return 29;
+ case Types.DECIMAL:
+ return Integer.MAX_VALUE;
+ default:
+ throw new SQLException("Invalid column type: " + columnType);
+ }
+ }
+
+ static int columnScale(int columnType) throws SQLException {
+ // according to hiveTypeToSqlType possible options are:
+ switch(columnType) {
+ case Types.BOOLEAN:
+ case Types.VARCHAR:
+ case Types.TINYINT:
+ case Types.SMALLINT:
+ case Types.INTEGER:
+ case Types.BIGINT:
+ return 0;
+ case Types.FLOAT:
+ return 7;
+ case Types.DOUBLE:
+ return 15;
+ case Types.TIMESTAMP:
+ return 9;
+ case Types.DECIMAL:
+ return Integer.MAX_VALUE;
+ default:
+ throw new SQLException("Invalid column type: " + columnType);
+ }
+ }
+
+ public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("getParentLogger not supported");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoMetaDataResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoMetaDataResultSet.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoMetaDataResultSet.java
new file mode 100644
index 0000000..1e75424
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoMetaDataResultSet.java
@@ -0,0 +1,77 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.storage.Tuple;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.List;
+
+public class TajoMetaDataResultSet extends TajoResultSetBase {
+ private List<MetaDataTuple> values;
+
+ public TajoMetaDataResultSet(List<String> columns, List<Type> types, List<MetaDataTuple> values) {
+ init();
+ schema = new Schema();
+
+ int index = 0;
+ if(columns != null) {
+ for(String columnName: columns) {
+ schema.addColumn(columnName, types.get(index++));
+ }
+ }
+ this.values = values;
+ totalRow = values == null ? 0 : values.size();
+ }
+
+ @Override
+ protected Tuple nextTuple() throws IOException {
+ if(curRow >= totalRow) {
+ return null;
+ }
+ return values.get(curRow);
+ }
+
+ @Override
+ public void close() throws SQLException {
+ }
+
+ @Override
+ public String getString(int fieldId) throws SQLException {
+ Datum datum = cur.get(fieldId - 1);
+ if(datum == null) {
+ return null;
+ }
+
+ return datum.asChars();
+ }
+
+ @Override
+ public String getString(String name) throws SQLException {
+ Datum datum = cur.get(findColumn(name));
+ if(datum == null) {
+ return null;
+ }
+ return datum.asChars();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java
new file mode 100644
index 0000000..a6e3bbf
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoPreparedStatement.java
@@ -0,0 +1,660 @@
+/**
+ * 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.tajo.jdbc;
+
+import org.apache.tajo.client.TajoClient;
+
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.net.URL;
+import java.sql.*;
+import java.util.Calendar;
+import java.util.HashMap;
+
+/**
+ * TajoPreparedStatement.
+ *
+ */
+public class TajoPreparedStatement implements PreparedStatement {
+ private final String sql;
+ private TajoClient tajoClient;
+ /**
+ * save the SQL parameters {paramLoc:paramValue}
+ */
+ private final HashMap<Integer, String> parameters=new HashMap<Integer, String>();
+
+ /**
+ * We need to keep a reference to the result set to support the following:
+ * <code>
+ * statement.execute(String sql);
+ * statement.getResultSet();
+ * </code>.
+ */
+ private ResultSet resultSet = null;
+
+ /**
+ * Add SQLWarnings to the warningChain if needed.
+ */
+ private SQLWarning warningChain = null;
+
+ /**
+ * Keep state so we can fail certain calls made after close().
+ */
+ private boolean isClosed = false;
+
+ /**
+ * keep the current ResultRet update count
+ */
+ private final int updateCount = 0;
+
+ /**
+ *
+ */
+ public TajoPreparedStatement(TajoClient tajoClient,
+ String sql) {
+ this.tajoClient = tajoClient;
+ this.sql = sql;
+ }
+
+ @Override
+ public void addBatch() throws SQLException {
+ throw new SQLFeatureNotSupportedException("addBatch");
+ }
+
+ @Override
+ public void clearParameters() throws SQLException {
+ this.parameters.clear();
+ }
+
+ @Override
+ public boolean execute() throws SQLException {
+ ResultSet rs = executeImmediate(sql);
+ return rs != null;
+ }
+
+ @Override
+ public ResultSet executeQuery() throws SQLException {
+ return executeImmediate(sql);
+ }
+
+ @Override
+ public int executeUpdate() throws SQLException {
+ executeImmediate(sql);
+ return updateCount;
+ }
+
+ protected ResultSet executeImmediate(String sql) throws SQLException {
+ if (isClosed) {
+ throw new SQLFeatureNotSupportedException("Can't execute after statement has been closed");
+ }
+
+ try {
+ if (sql.contains("?")) {
+ sql = updateSql(sql, parameters);
+ }
+ resultSet = tajoClient.executeQueryAndGetResult(sql);
+ } catch (Exception e) {
+ throw new SQLFeatureNotSupportedException(e.getMessage(), e);
+ }
+ return resultSet;
+ }
+
+ /**
+ * update the SQL string with parameters set by setXXX methods of {@link java.sql.PreparedStatement}
+ *
+ * @param sql
+ * @param parameters
+ * @return updated SQL string
+ */
+ private String updateSql(final String sql, HashMap<Integer, String> parameters) {
+
+ StringBuffer newSql = new StringBuffer(sql);
+
+ int paramLoc = 1;
+ while (getCharIndexFromSqlByParamLocation(sql, '?', paramLoc) > 0) {
+ // check the user has set the needs parameters
+ if (parameters.containsKey(paramLoc)) {
+ int tt = getCharIndexFromSqlByParamLocation(newSql.toString(), '?', 1);
+ newSql.deleteCharAt(tt);
+ newSql.insert(tt, parameters.get(paramLoc));
+ }
+ paramLoc++;
+ }
+
+ return newSql.toString();
+
+ }
+
+ /**
+ * Get the index of given char from the SQL string by parameter location
+ * </br> The -1 will be return, if nothing found
+ *
+ * @param sql
+ * @param cchar
+ * @param paramLoc
+ * @return
+ */
+ private int getCharIndexFromSqlByParamLocation(final String sql, final char cchar, final int paramLoc) {
+ int signalCount = 0;
+ int charIndex = -1;
+ int num = 0;
+ for (int i = 0; i < sql.length(); i++) {
+ char c = sql.charAt(i);
+ if (c == '\'' || c == '\\')// record the count of char "'" and char "\"
+ {
+ signalCount++;
+ } else if (c == cchar && signalCount % 2 == 0) {// check if the ? is really the parameter
+ num++;
+ if (num == paramLoc) {
+ charIndex = i;
+ break;
+ }
+ }
+ }
+ return charIndex;
+ }
+
+ @Override
+ public ResultSetMetaData getMetaData() throws SQLException {
+ if(resultSet != null) {
+ return resultSet.getMetaData();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public ParameterMetaData getParameterMetaData() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getParameterMetaData not supported");
+ }
+
+ @Override
+ public void setArray(int i, Array x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setArray not supported");
+ }
+
+ @Override
+ public void setAsciiStream(int parameterIndex, InputStream x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setAsciiStream not supported");
+ }
+
+ @Override
+ public void setAsciiStream(int parameterIndex, InputStream x, int length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setAsciiStream not supported");
+ }
+
+ @Override
+ public void setAsciiStream(int parameterIndex, InputStream x, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setAsciiStream not supported");
+ }
+
+ @Override
+ public void setBigDecimal(int parameterIndex, BigDecimal x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBigDecimal not supported");
+ }
+
+ @Override
+ public void setBinaryStream(int parameterIndex, InputStream x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBinaryStream not supported");
+ }
+
+ @Override
+ public void setBinaryStream(int parameterIndex, InputStream x, int length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBinaryStream not supported");
+ }
+
+ @Override
+ public void setBinaryStream(int parameterIndex, InputStream x, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBinaryStream not supported");
+ }
+
+ @Override
+ public void setBlob(int i, Blob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBlob not supported");
+ }
+
+ @Override
+ public void setBlob(int parameterIndex, InputStream inputStream) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBlob not supported");
+ }
+
+ @Override
+ public void setBlob(int parameterIndex, InputStream inputStream, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBlob not supported");
+ }
+
+ @Override
+ public void setBoolean(int parameterIndex, boolean x) throws SQLException {
+ this.parameters.put(parameterIndex, "" + x);
+ }
+
+ @Override
+ public void setByte(int parameterIndex, byte x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setByte not supported");
+ }
+
+ @Override
+ public void setBytes(int parameterIndex, byte[] x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setBytes not supported");
+ }
+
+ @Override
+ public void setCharacterStream(int parameterIndex, Reader reader) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setCharacterStream not supported");
+ }
+
+ @Override
+ public void setCharacterStream(int parameterIndex, Reader reader, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setCharacterStream not supported");
+ }
+
+ @Override
+ public void setCharacterStream(int parameterIndex, Reader reader, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setCharacterStream not supported");
+ }
+
+ @Override
+ public void setClob(int i, Clob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setClob not supported");
+ }
+
+ @Override
+ public void setClob(int parameterIndex, Reader reader) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setClob not supported");
+ }
+
+ @Override
+ public void setClob(int parameterIndex, Reader reader, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setClob not supported");
+ }
+
+ @Override
+ public void setDate(int parameterIndex, Date x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setDate not supported");
+ }
+
+ @Override
+ public void setDate(int parameterIndex, Date x, Calendar cal) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setDate not supported");
+ }
+
+ @Override
+ public void setDouble(int parameterIndex, double x) throws SQLException {
+ this.parameters.put(parameterIndex,"" + x);
+ }
+
+ @Override
+ public void setFloat(int parameterIndex, float x) throws SQLException {
+ this.parameters.put(parameterIndex,"" + x);
+ }
+
+ @Override
+ public void setInt(int parameterIndex, int x) throws SQLException {
+ this.parameters.put(parameterIndex,"" + x);
+ }
+
+ @Override
+ public void setLong(int parameterIndex, long x) throws SQLException {
+ this.parameters.put(parameterIndex,"" + x);
+ }
+
+ @Override
+ public void setNCharacterStream(int parameterIndex, Reader value) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNCharacterStream not supported");
+ }
+
+ @Override
+ public void setNCharacterStream(int parameterIndex, Reader value, long length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNCharacterStream not supported");
+ }
+
+ @Override
+ public void setNClob(int parameterIndex, NClob value) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNClob not supported");
+ }
+
+ @Override
+ public void setNClob(int parameterIndex, Reader reader) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNClob not supported");
+ }
+
+ @Override
+ public void setNClob(int parameterIndex, Reader reader, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNClob not supported");
+ }
+
+ @Override
+ public void setNString(int parameterIndex, String value) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNString not supported");
+ }
+
+ @Override
+ public void setNull(int parameterIndex, int sqlType) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNull not supported");
+ }
+
+ @Override
+ public void setNull(int paramIndex, int sqlType, String typeName) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setNull not supported");
+ }
+
+ @Override
+ public void setObject(int parameterIndex, Object x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setObject not supported");
+ }
+
+ @Override
+ public void setObject(int parameterIndex, Object x, int targetSqlType)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setObject not supported");
+ }
+
+ @Override
+ public void setObject(int parameterIndex, Object x, int targetSqlType, int scale)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setObject not supported");
+ }
+
+ @Override
+ public void setRef(int i, Ref x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setRef not supported");
+ }
+
+ @Override
+ public void setRowId(int parameterIndex, RowId x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setRowId not supported");
+ }
+
+ @Override
+ public void setSQLXML(int parameterIndex, SQLXML xmlObject) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setSQLXML not supported");
+ }
+
+ @Override
+ public void setShort(int parameterIndex, short x) throws SQLException {
+ this.parameters.put(parameterIndex,"" + x);
+ }
+
+ @Override
+ public void setString(int parameterIndex, String x) throws SQLException {
+ x=x.replace("'", "\\'");
+ this.parameters.put(parameterIndex,"'" + x +"'");
+ }
+
+ @Override
+ public void setTime(int parameterIndex, Time x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTime not supported");
+ }
+
+ @Override
+ public void setTime(int parameterIndex, Time x, Calendar cal) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTime not supported");
+ }
+
+ @Override
+ public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTimestamp not supported");
+ }
+
+ @Override
+ public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setTimestamp not supported");
+ }
+
+ @Override
+ public void setURL(int parameterIndex, URL x) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setURL not supported");
+ }
+
+ @Override
+ public void setUnicodeStream(int parameterIndex, InputStream x, int length)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException("setUnicodeStream not supported");
+ }
+
+ @Override
+ public void addBatch(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("addBatch not supported");
+ }
+
+ @Override
+ public void cancel() throws SQLException {
+ throw new SQLFeatureNotSupportedException("cancel not supported");
+ }
+
+ @Override
+ public void clearBatch() throws SQLException {
+ throw new SQLFeatureNotSupportedException("clearBatch not supported");
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ warningChain=null;
+ }
+
+ public void closeOnCompletion() throws SQLException {
+ // JDK 1.7
+ throw new SQLFeatureNotSupportedException("closeOnCompletion");
+ }
+
+ @Override
+ public void close() throws SQLException {
+ if (resultSet!=null) {
+ resultSet.close();
+ resultSet = null;
+ }
+ isClosed = true;
+ }
+
+ @Override
+ public boolean execute(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("execute(sql) not supported");
+ }
+
+ @Override
+ public boolean execute(String sql, int autoGeneratedKeys) throws SQLException {
+ throw new SQLFeatureNotSupportedException("execute(sql) not supported");
+ }
+
+ @Override
+ public boolean execute(String sql, int[] columnIndexes) throws SQLException {
+ throw new SQLFeatureNotSupportedException("execute(sql) not supported");
+ }
+
+ @Override
+ public boolean execute(String sql, String[] columnNames) throws SQLException {
+ throw new SQLFeatureNotSupportedException("execute(sql) not supported");
+ }
+
+ @Override
+ public int[] executeBatch() throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeBatch not supported");
+ }
+
+ @Override
+ public ResultSet executeQuery(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeQuery(sql) not supported");
+ }
+
+ @Override
+ public int executeUpdate(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeUpdate not supported");
+ }
+
+ @Override
+ public int executeUpdate(String sql, int autoGeneratedKeys) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeUpdate not supported");
+ }
+
+ @Override
+ public int executeUpdate(String sql, int[] columnIndexes) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeUpdate not supported");
+ }
+
+ @Override
+ public int executeUpdate(String sql, String[] columnNames) throws SQLException {
+ throw new SQLFeatureNotSupportedException("executeUpdate not supported");
+ }
+
+ @Override
+ public Connection getConnection() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getConnection not supported");
+ }
+
+ @Override
+ public int getFetchDirection() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getFetchDirection not supported");
+ }
+
+ @Override
+ public int getFetchSize() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getFetchSize not supported");
+ }
+
+ @Override
+ public ResultSet getGeneratedKeys() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getGeneratedKeys not supported");
+ }
+
+ @Override
+ public int getMaxFieldSize() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxFieldSize not supported");
+ }
+
+ @Override
+ public int getMaxRows() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMaxRows not supported");
+ }
+
+ @Override
+ public boolean getMoreResults() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMoreResults not supported");
+ }
+
+ @Override
+ public boolean getMoreResults(int current) throws SQLException {
+ throw new SQLFeatureNotSupportedException("getMoreResults not supported");
+ }
+
+ @Override
+ public int getQueryTimeout() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getQueryTimeout not supported");
+ }
+
+ @Override
+ public ResultSet getResultSet() throws SQLException {
+ return this.resultSet;
+ }
+
+ @Override
+ public int getResultSetConcurrency() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getResultSetConcurrency not supported");
+ }
+
+ @Override
+ public int getResultSetHoldability() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getResultSetHoldability not supported");
+ }
+
+ @Override
+ public int getResultSetType() throws SQLException {
+ throw new SQLFeatureNotSupportedException("getResultSetType not supported");
+ }
+
+ @Override
+ public int getUpdateCount() throws SQLException {
+ return updateCount;
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ return warningChain;
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return isClosed;
+ }
+
+ public boolean isCloseOnCompletion() throws SQLException {
+ //JDK 1.7
+ throw new SQLFeatureNotSupportedException("isCloseOnCompletion not supported");
+ }
+
+ @Override
+ public boolean isPoolable() throws SQLException {
+ throw new SQLFeatureNotSupportedException("isPoolable not supported");
+ }
+
+ @Override
+ public void setCursorName(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setCursorName not supported");
+ }
+
+ @Override
+ public void setEscapeProcessing(boolean enable) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setEscapeProcessing not supported");
+ }
+
+ @Override
+ public void setFetchDirection(int direction) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setFetchDirection not supported");
+ }
+
+ @Override
+ public void setFetchSize(int rows) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setFetchSize not supported");
+ }
+
+ @Override
+ public void setMaxFieldSize(int max) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setMaxFieldSize not supported");
+ }
+
+ @Override
+ public void setMaxRows(int max) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setMaxRows not supported");
+ }
+
+ @Override
+ public void setPoolable(boolean poolable) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setPoolable not supported");
+ }
+
+ @Override
+ public void setQueryTimeout(int seconds) throws SQLException {
+ throw new SQLFeatureNotSupportedException("setQueryTimeout not supported");
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ throw new SQLFeatureNotSupportedException("isWrapperFor not supported");
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ throw new SQLFeatureNotSupportedException("unwrap not supported");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/342fd47f/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
new file mode 100644
index 0000000..a50ed2c
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/jdbc/TajoResultSet.java
@@ -0,0 +1,150 @@
+/**
+ * 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.tajo.jdbc;
+
+import com.google.common.collect.Lists;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.tajo.QueryId;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.storage.MergeScanner;
+import org.apache.tajo.storage.Scanner;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.storage.fragment.FileFragment;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+
+public class TajoResultSet extends TajoResultSetBase {
+ private FileSystem fs;
+ private Scanner scanner;
+ private TajoClient tajoClient;
+ QueryId queryId;
+
+ public TajoResultSet(TajoClient tajoClient, QueryId queryId) {
+ this.tajoClient = tajoClient;
+ this.queryId = queryId;
+ init();
+ }
+
+ public TajoResultSet(TajoClient tajoClient, QueryId queryId,
+ Configuration conf, TableDesc desc) throws IOException {
+ this.schema = desc.getSchema();
+ this.tajoClient = tajoClient;
+ this.queryId = queryId;
+ if(desc != null) {
+ fs = desc.getPath().getFileSystem(conf);
+ this.totalRow = desc.getStats() != null ? desc.getStats().getNumRows() : 0;
+
+ Collection<FileFragment> frags = getFragments(desc.getMeta(), desc.getPath());
+ scanner = new MergeScanner(conf, schema, desc.getMeta(), frags);
+ }
+ init();
+ }
+
+ @Override
+ protected void init() {
+ cur = null;
+ curRow = 0;
+ }
+
+ class FileNameComparator implements Comparator<FileStatus> {
+
+ @Override
+ public int compare(FileStatus f1, FileStatus f2) {
+ return f2.getPath().getName().compareTo(f1.getPath().getName());
+ }
+ }
+
+ private Collection<FileFragment> getFragments(TableMeta meta, Path tablePath)
+ throws IOException {
+ List<FileFragment> fraglist = Lists.newArrayList();
+ FileStatus[] files = fs.listStatus(tablePath, new PathFilter() {
+ @Override
+ public boolean accept(Path path) {
+ return path.getName().charAt(0) != '.';
+ }
+ });
+ Arrays.sort(files, new FileNameComparator());
+
+ String tbname = tablePath.getName();
+ for (int i = 0; i < files.length; i++) {
+ if (files[i].getLen() == 0) {
+ continue;
+ }
+ fraglist.add(new FileFragment(tbname + "_" + i, files[i].getPath(), 0l, files[i].getLen()));
+ }
+ return fraglist;
+ }
+
+ @Override
+ public void close() throws SQLException {
+ try {
+ if(tajoClient != null) {
+ this.tajoClient.closeQuery(queryId);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ try {
+ if(scanner != null) {
+ this.scanner.close();
+ }
+ //TODO clean temp result file
+ cur = null;
+ curRow = -1;
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ public void beforeFirst() throws SQLException {
+ try {
+ if(scanner != null) {
+ scanner.reset();
+ }
+ init();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+
+ @Override
+ protected Tuple nextTuple() throws IOException {
+ if(scanner == null) {
+ return null;
+ }
+ return scanner.next();
+ }
+
+ public boolean hasResult() {
+ return scanner != null;
+ }
+}