hbase support

This commit is contained in:
温少
2012-05-21 05:24:43 +08:00
parent 647fc890aa
commit e538dd1909
11 changed files with 307 additions and 41 deletions
@@ -0,0 +1,59 @@
package com.alibaba.druid.common.jdbc;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
public class LocalResultSet extends ResultSetBase {
private int rowIndex = -1;
private List<Object[]> rows = new ArrayList<Object[]>();
public LocalResultSet(Statement statement){
super(statement);
}
public List<Object[]> getRows() {
return rows;
}
@Override
public synchronized boolean next() throws SQLException {
if (closed) {
throw new SQLException();
}
if (rowIndex < rows.size() - 1) {
rowIndex++;
return true;
}
return false;
}
public Object getObjectInternal(int columnIndex) {
Object[] row = rows.get(rowIndex);
Object obj = row[columnIndex - 1];
return obj;
}
@Override
public synchronized boolean previous() throws SQLException {
if (closed) {
throw new SQLException();
}
if (rowIndex > 0) {
rowIndex--;
return true;
}
return false;
}
@Override
public void updateObject(int columnIndex, Object x) throws SQLException {
Object[] row = rows.get(rowIndex);
row[columnIndex - 1] = x;
}
}
@@ -11,6 +11,7 @@ import java.sql.Date;
import java.sql.NClob;
import java.sql.Ref;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.RowId;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
@@ -24,14 +25,15 @@ import java.util.Map;
public abstract class ResultSetBase implements ResultSet {
protected boolean closed = false;
protected boolean wasNull = false;
private SQLWarning warning;
private String cursorName;
private int fetchSize = 0;
private int fetchDirection = 0;
protected boolean closed = false;
protected boolean wasNull = false;
private SQLWarning warning;
private String cursorName;
private int fetchSize = 0;
private int fetchDirection = 0;
private Statement statement;
private Statement statement;
protected ResultSetMetaData metaData;
public ResultSetBase(Statement statement){
super();
@@ -302,7 +304,14 @@ public abstract class ResultSetBase implements ResultSet {
return wasNull;
}
public abstract Object getObjectInternal(int columnIndex) throws SQLException;
public Object getObjectInternal(int columnIndex) throws SQLException {
if (this.getMetaData() != null) {
String columName = this.getMetaData().getColumnName(columnIndex);
return getObject(columName);
}
return null;
}
@Override
public Object getObject(int columnIndex) throws SQLException {
@@ -340,7 +349,7 @@ public abstract class ResultSetBase implements ResultSet {
@Override
public BigDecimal getBigDecimal(String columnLabel) throws SQLException {
return getBigDecimal(Integer.parseInt(columnLabel));
return getBigDecimal(findColumn(columnLabel));
}
@Override
@@ -623,7 +632,7 @@ public abstract class ResultSetBase implements ResultSet {
@Override
public void updateObject(String columnLabel, Object x) throws SQLException {
updateObject(Integer.parseInt(columnLabel), x);
updateObject(findColumn(columnLabel), x);
}
@Override
@@ -966,82 +975,82 @@ public abstract class ResultSetBase implements ResultSet {
@Override
public String getString(String columnLabel) throws SQLException {
return getString(Integer.parseInt(columnLabel));
return getString(findColumn(columnLabel));
}
@Override
public boolean getBoolean(String columnLabel) throws SQLException {
return getBoolean(Integer.parseInt(columnLabel));
return getBoolean(findColumn(columnLabel));
}
@Override
public byte getByte(String columnLabel) throws SQLException {
return getByte(Integer.parseInt(columnLabel));
return getByte(findColumn(columnLabel));
}
@Override
public short getShort(String columnLabel) throws SQLException {
return getShort(Integer.parseInt(columnLabel));
return getShort(findColumn(columnLabel));
}
@Override
public int getInt(String columnLabel) throws SQLException {
return getInt(Integer.parseInt(columnLabel));
return getInt(findColumn(columnLabel));
}
@Override
public long getLong(String columnLabel) throws SQLException {
return getLong(Integer.parseInt(columnLabel));
return getLong(findColumn(columnLabel));
}
@Override
public float getFloat(String columnLabel) throws SQLException {
return getFloat(Integer.parseInt(columnLabel));
return getFloat(findColumn(columnLabel));
}
@Override
public double getDouble(String columnLabel) throws SQLException {
return getDouble(Integer.parseInt(columnLabel));
return getDouble(findColumn(columnLabel));
}
@Override
public BigDecimal getBigDecimal(String columnLabel, int scale) throws SQLException {
return getBigDecimal(Integer.parseInt(columnLabel), scale);
return getBigDecimal(findColumn(columnLabel), scale);
}
@Override
public byte[] getBytes(String columnLabel) throws SQLException {
return getBytes(Integer.parseInt(columnLabel));
return getBytes(findColumn(columnLabel));
}
@Override
public Date getDate(String columnLabel) throws SQLException {
return getDate(Integer.parseInt(columnLabel));
return getDate(findColumn(columnLabel));
}
@Override
public Time getTime(String columnLabel) throws SQLException {
return getTime(Integer.parseInt(columnLabel));
return getTime(findColumn(columnLabel));
}
@Override
public Timestamp getTimestamp(String columnLabel) throws SQLException {
return getTimestamp(Integer.parseInt(columnLabel));
return getTimestamp(findColumn(columnLabel));
}
@Override
public InputStream getAsciiStream(String columnLabel) throws SQLException {
return getAsciiStream(Integer.parseInt(columnLabel));
return getAsciiStream(findColumn(columnLabel));
}
@Override
public InputStream getUnicodeStream(String columnLabel) throws SQLException {
return getUnicodeStream(Integer.parseInt(columnLabel));
return getUnicodeStream(findColumn(columnLabel));
}
@Override
public InputStream getBinaryStream(String columnLabel) throws SQLException {
return getBinaryStream(Integer.parseInt(columnLabel));
return getBinaryStream(findColumn(columnLabel));
}
@Override
@@ -1157,4 +1166,13 @@ public abstract class ResultSetBase implements ResultSet {
return 0;
}
@Override
public ResultSetMetaData getMetaData() throws SQLException {
if (closed) {
throw new SQLException("resultSet closed");
}
return metaData;
}
}
@@ -16,7 +16,7 @@ import com.alibaba.druid.common.jdbc.ResultSetMetaDataBase.ColumnMetaData;
import com.alibaba.druid.hbase.mapping.HMapping;
import com.alibaba.druid.util.JdbcUtils;
public class HBaseResultSet extends ResultSetBase {
public class HBaseResultSet extends ResultSetBase implements HResultSet {
private HStatementInterface statement;
private ResultScanner scanner;
@@ -24,8 +24,6 @@ public class HBaseResultSet extends ResultSetBase {
private Result result;
private byte[] family = Bytes.toBytes("d");
private HResultSetMetaData metaData;
private HMapping mapping;
public HBaseResultSet(HStatementInterface statement, HTableInterface htable, ResultScanner scanner){
@@ -60,7 +58,7 @@ public class HBaseResultSet extends ResultSetBase {
@Override
public HResultSetMetaData getMetaData() throws SQLException {
return metaData;
return (HResultSetMetaData) metaData;
}
public void setMetaData(HResultSetMetaData metaData) {
@@ -78,7 +76,7 @@ public class HBaseResultSet extends ResultSetBase {
@Override
public Object getObjectInternal(int columnIndex) throws SQLException {
ColumnMetaData column = this.metaData.getColumn(columnIndex);
ColumnMetaData column = this.getMetaData().getColumn(columnIndex);
return getObjectInternal(column.getColumnName(), column.getColumnType());
}
@@ -129,7 +127,7 @@ public class HBaseResultSet extends ResultSetBase {
@Override
public byte[] getBytes(int columnIndex) throws SQLException {
ColumnMetaData column = this.metaData.getColumn(columnIndex);
ColumnMetaData column = this.getMetaData().getColumn(columnIndex);
return getBytes(column.getColumnName());
}
@@ -1,10 +1,12 @@
package com.alibaba.druid.hbase;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
@@ -77,5 +79,9 @@ public class HEngine {
}
return tablePool.getTable(tableName);
}
public HBaseAdmin getHBaseAdmin() throws IOException {
return new HBaseAdmin(this.config);
}
}
@@ -10,7 +10,9 @@ import com.alibaba.druid.common.jdbc.PreparedStatementBase;
import com.alibaba.druid.common.jdbc.ResultSetMetaDataBase.ColumnMetaData;
import com.alibaba.druid.hbase.exec.ExecutePlan;
import com.alibaba.druid.hbase.exec.InsertExecutePlan;
import com.alibaba.druid.hbase.exec.ShowTablesPlan;
import com.alibaba.druid.hbase.exec.SingleTableQueryExecutePlan;
import com.alibaba.druid.hbase.hbql.ast.HBQLShowStatement;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
@@ -113,6 +115,9 @@ public class HPreparedStatement extends PreparedStatementBase implements Prepare
}
this.executePlan = insertExecutePlan;
} else if (sqlStmt instanceof HBQLShowStatement) {
ShowTablesPlan showTablePlan = new ShowTablesPlan();
this.executePlan = showTablePlan;
} else {
throw new SQLException("TODO");
}
@@ -0,0 +1,8 @@
package com.alibaba.druid.hbase;
import java.sql.ResultSet;
public interface HResultSet extends ResultSet {
}
@@ -3,11 +3,11 @@ package com.alibaba.druid.hbase.exec;
import java.sql.SQLException;
import com.alibaba.druid.hbase.HPreparedStatement;
import com.alibaba.druid.hbase.HBaseResultSet;
import com.alibaba.druid.hbase.HResultSet;
public interface ExecutePlan {
HBaseResultSet executeQuery(HPreparedStatement statement) throws SQLException;
HResultSet executeQuery(HPreparedStatement statement) throws SQLException;
boolean execute(HPreparedStatement statement) throws SQLException;
}
@@ -3,12 +3,12 @@ package com.alibaba.druid.hbase.exec;
import java.sql.SQLException;
import com.alibaba.druid.hbase.HPreparedStatement;
import com.alibaba.druid.hbase.HBaseResultSet;
import com.alibaba.druid.hbase.HResultSet;
public class ExecutePlanAdapter implements ExecutePlan {
@Override
public HBaseResultSet executeQuery(HPreparedStatement statement) throws SQLException {
public HResultSet executeQuery(HPreparedStatement statement) throws SQLException {
throw new UnsupportedOperationException();
}
@@ -0,0 +1,166 @@
package com.alibaba.druid.hbase.exec;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import com.alibaba.druid.common.jdbc.ResultSetBase;
import com.alibaba.druid.common.jdbc.ResultSetMetaDataBase.ColumnMetaData;
import com.alibaba.druid.hbase.HPreparedStatement;
import com.alibaba.druid.hbase.HResultSet;
import com.alibaba.druid.hbase.HResultSetMetaData;
public class ShowTablesPlan implements ExecutePlan {
@Override
public HResultSet executeQuery(HPreparedStatement statement) throws SQLException {
try {
HBaseAdmin admin = statement.getConnection().getEngine().getHBaseAdmin();
HTableDescriptor[] tables = admin.listTables();
return new ShowTableResultSet(statement, tables);
} catch (SQLException e) {
throw e;
} catch (Exception e) {
throw new SQLException("executeQuery error", e);
}
}
@Override
public boolean execute(HPreparedStatement statement) throws SQLException {
HResultSet resultSet = executeQuery(statement);
statement.setResultSet(resultSet);
return true;
}
public static class ShowTableResultSet extends ResultSetBase implements HResultSet {
private final HTableDescriptor[] tables;
private int rowIndex = -1;
public ShowTableResultSet(Statement statement, HTableDescriptor[] tables){
super(statement);
this.tables = tables;
HResultSetMetaData metaData = new HResultSetMetaData();
{
ColumnMetaData column = new ColumnMetaData();
column.setColumnName("name");
column.setColumnType(Types.VARCHAR);
column.setColumnTypeName("VARCHAR");
metaData.getColumns().add(column);
}
{
ColumnMetaData column = new ColumnMetaData();
column.setColumnName("owner");
column.setColumnType(Types.VARCHAR);
column.setColumnTypeName("VARCHAR");
metaData.getColumns().add(column);
}
{
ColumnMetaData column = new ColumnMetaData();
column.setColumnName("familys");
column.setColumnType(Types.VARCHAR);
column.setColumnTypeName("VARCHAR");
metaData.getColumns().add(column);
}
{
ColumnMetaData column = new ColumnMetaData();
column.setColumnName("maxFileSize");
column.setColumnType(Types.INTEGER);
column.setColumnTypeName("int");
metaData.getColumns().add(column);
}
this.metaData = metaData;
}
@Override
public synchronized boolean next() throws SQLException {
if (closed) {
throw new SQLException();
}
if (rowIndex < tables.length - 1) {
rowIndex++;
return true;
}
return false;
}
@Override
public synchronized boolean previous() throws SQLException {
if (closed) {
throw new SQLException();
}
if (rowIndex > 0) {
rowIndex--;
return true;
}
return false;
}
public String getString(String columnName) {
return (String) getObject(columnName);
}
public Object getObject(String columnName) {
HTableDescriptor table = tables[rowIndex];
if ("name".equals(columnName)) {
return table.getNameAsString();
}
if ("owner".equals(columnName)) {
return table.getOwnerString();
}
if ("familys".equals(columnName)) {
StringBuffer buf = new StringBuffer();
for (HColumnDescriptor column : table.getFamilies()) {
if (buf.length() != 0) {
buf.append(',');
}
buf.append(column.getNameAsString());
}
return buf.toString();
}
if ("maxFileSize".equals(columnName)) {
table.getMaxFileSize();
}
return null;
}
@Override
public int findColumn(String columnName) throws SQLException {
if ("name".equals(columnName)) {
return 1;
}
if ("owner".equals(columnName)) {
return 2;
}
if ("familys".equals(columnName)) {
return 3;
}
if ("maxFileSize".equals(columnName)) {
return 4;
}
return 0;
}
@Override
public void updateObject(int columnIndex, Object x) throws SQLException {
}
}
}
@@ -29,10 +29,9 @@ public class MockResultSet extends ResultSetBase implements ResultSet {
private int rowIndex = -1;
private List<Object[]> rows = new ArrayList<Object[]>();
private MockResultSetMetaData metaData = new MockResultSetMetaData();
public MockResultSet(Statement statement){
super(statement);
super.metaData = new MockResultSetMetaData();
}
public List<Object[]> getRows() {
@@ -59,11 +58,11 @@ public class MockResultSet extends ResultSetBase implements ResultSet {
throw new SQLException("resultSet closed");
}
return metaData;
return (MockResultSetMetaData) metaData;
}
public MockResultSetMetaData getMockMetaData() throws SQLException {
return metaData;
return (MockResultSetMetaData) metaData;
}
public Object getObjectInternal(int columnIndex) {