hbase support

This commit is contained in:
温少
2012-05-18 23:58:58 +08:00
4 changed files with 36 additions and 13 deletions
@@ -1,5 +1,6 @@
package com.alibaba.druid.hbase;
import java.io.IOException;
import java.sql.Array;
import java.sql.Blob;
import java.sql.CallableStatement;
@@ -22,6 +23,7 @@ import java.util.Properties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
@@ -244,6 +246,10 @@ public class HBaseConnection extends ConnectionBase implements Connection {
return null;
}
public HTableInterface getHTable(String tableName) throws IOException {
return new HTable(config, tableName);
}
public ResultSet executeQuery(HBaseStatementInterface jdbcHbaseStmt, String sql, List<Object> parameters) throws SQLException {
try {
@@ -260,12 +266,12 @@ public class HBaseConnection extends ConnectionBase implements Connection {
SQLExprTableSource tableSource = (SQLExprTableSource) selectQueryBlock.getFrom();
String tableName = ((SQLIdentifierExpr) tableSource.getExpr()).getName();
HTable htable = new HTable(config, tableName);
HTableInterface htable = getHTable(tableName);
Scan scan = new Scan();
ResultScanner scanner = htable.getScanner(scan);
return new HBaseResultSet(jdbcHbaseStmt, scanner);
return new HBaseResultSet(jdbcHbaseStmt, htable, scanner);
} catch (SQLException ex) {
throw ex;
} catch (Exception ex) {
@@ -5,22 +5,26 @@ import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Bytes;
import com.alibaba.druid.common.jdbc.ResultSetBase;
import com.alibaba.druid.util.JdbcUtils;
public class HBaseResultSet extends ResultSetBase {
private HBaseStatementInterface statement;
private ResultScanner scanner;
private HTableInterface htable;
private Result result;
private byte[] family = Bytes.toBytes("d");
public HBaseResultSet(HBaseStatementInterface statement, ResultScanner scanner){
public HBaseResultSet(HBaseStatementInterface statement, HTableInterface htable, ResultScanner scanner){
super(statement);
this.statement = statement;
this.htable = htable;
this.scanner = scanner;
}
@@ -67,12 +71,12 @@ public class HBaseResultSet extends ResultSetBase {
byte[] bytes = getBytes(columnName);
return Bytes.toString(bytes);
}
public byte[] getBytes(String columnName) throws SQLException {
if ("@ROW".equals(columnName)) {
return result.getRow();
}
byte[] qualifier = Bytes.toBytes(columnName);
byte[] value = result.getValue(family, qualifier);
return value;
@@ -80,8 +84,9 @@ public class HBaseResultSet extends ResultSetBase {
@Override
public void close() throws SQLException {
JdbcUtils.close(scanner);
try {
scanner.close();
htable.close();
} catch (Exception ex) {
throw new SQLException("close error", ex);
}
@@ -1,5 +1,9 @@
package com.alibaba.druid.hbase.exec;
public interface ExecutePlan {
import java.util.List;
import com.alibaba.druid.hbase.HBaseConnection;
public interface ExecutePlan {
void execute(HBaseConnection connection, List<Object> paramerers);
}
@@ -1,17 +1,25 @@
package com.alibaba.druid.hbase.exec;
import org.apache.hadoop.hbase.client.HTableInterface;
import java.util.List;
import com.alibaba.druid.hbase.HBaseConnection;
public class SingleTableQueryExecutePlan implements ExecutePlan {
private HTableInterface htable;
private String tableName;
public HTableInterface getHtable() {
return htable;
public String getTableName() {
return tableName;
}
public void setHtable(HTableInterface htable) {
this.htable = htable;
public void setTableName(String tableName) {
this.tableName = tableName;
}
@Override
public void execute(HBaseConnection connection, List<Object> paramerers) {
}
}