hbase support

This commit is contained in:
温少
2012-05-19 08:19:51 +08:00
5 changed files with 79 additions and 50 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;
@@ -8,7 +9,6 @@ import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.NClob;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLClientInfoException;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
@@ -16,22 +16,14 @@ import java.sql.SQLXML;
import java.sql.Savepoint;
import java.sql.Statement;
import java.sql.Struct;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.HTableInterface;
import com.alibaba.druid.common.jdbc.ConnectionBase;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.druid.sql.ast.statement.SQLSelectQueryBlock;
import com.alibaba.druid.sql.ast.statement.SQLSelectStatement;
public class HBaseConnection extends ConnectionBase implements Connection {
@@ -244,32 +236,9 @@ public class HBaseConnection extends ConnectionBase implements Connection {
return null;
}
public ResultSet executeQuery(HBaseStatementInterface jdbcHbaseStmt, String sql, List<Object> parameters) throws SQLException {
try {
String dbType = this.getConnectProperties().getProperty("dbType");
List<SQLStatement> stmtList = SQLUtils.parseStatements(sql, dbType);
if (stmtList.size() != 1) {
throw new SQLException("not support multi-statement");
}
SQLSelectStatement stmt = (SQLSelectStatement) stmtList.get(0);
SQLSelectQueryBlock selectQueryBlock = (SQLSelectQueryBlock) stmt.getSelect().getQuery();
SQLExprTableSource tableSource = (SQLExprTableSource) selectQueryBlock.getFrom();
String tableName = ((SQLIdentifierExpr) tableSource.getExpr()).getName();
HTable htable = new HTable(config, tableName);
Scan scan = new Scan();
ResultScanner scanner = htable.getScanner(scan);
return new HBaseResultSet(jdbcHbaseStmt, scanner);
} catch (SQLException ex) {
throw ex;
} catch (Exception ex) {
throw new SQLException("executeQuery error", ex);
}
public HTableInterface getHTable(String tableName) throws IOException {
return new HTable(config, tableName);
}
}
@@ -3,8 +3,17 @@ package com.alibaba.druid.hbase;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import com.alibaba.druid.common.jdbc.PreparedStatementBase;
import com.alibaba.druid.hbase.exec.ExecutePlan;
import com.alibaba.druid.hbase.exec.SingleTableQueryExecutePlan;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.druid.sql.ast.statement.SQLSelectQueryBlock;
import com.alibaba.druid.sql.ast.statement.SQLSelectStatement;
public class HBasePreparedStatement extends PreparedStatementBase implements PreparedStatement, HBaseStatementInterface {
@@ -13,10 +22,38 @@ public class HBasePreparedStatement extends PreparedStatementBase implements Pre
private HBaseConnection hbaseConnection;
public HBasePreparedStatement(HBaseConnection conn, String sql){
private ExecutePlan executePlan;
public HBasePreparedStatement(HBaseConnection conn, String sql) throws SQLException{
super(conn);
this.sql = sql;
this.hbaseConnection = conn;
init();
}
public void init() throws SQLException {
String dbType = this.hbaseConnection.getConnectProperties().getProperty("dbType");
List<SQLStatement> stmtList = SQLUtils.parseStatements(sql, dbType);
if (stmtList.size() != 1) {
throw new SQLException("not support multi-statement");
}
SQLSelectStatement stmt = (SQLSelectStatement) stmtList.get(0);
SQLSelectQueryBlock selectQueryBlock = (SQLSelectQueryBlock) stmt.getSelect().getQuery();
SQLExprTableSource tableSource = (SQLExprTableSource) selectQueryBlock.getFrom();
String tableName = ((SQLIdentifierExpr) tableSource.getExpr()).getName();
SingleTableQueryExecutePlan singleTableQueryExecuetePlan = new SingleTableQueryExecutePlan();
singleTableQueryExecuetePlan.setTableName(tableName);
this.executePlan = singleTableQueryExecuetePlan;
}
public ExecutePlan getExecutePlan() {
return executePlan;
}
public String[] getColumnNames() {
@@ -38,7 +75,7 @@ public class HBasePreparedStatement extends PreparedStatementBase implements Pre
@Override
public ResultSet executeQuery() throws SQLException {
return hbaseConnection.executeQuery(this, sql, getParameters());
return this.executePlan.executeScan(this);
}
@Override
@@ -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,9 +1,10 @@
package com.alibaba.druid.hbase.exec;
import java.util.List;
import java.sql.SQLException;
import com.alibaba.druid.hbase.HBaseConnection;
import com.alibaba.druid.hbase.HBasePreparedStatement;
import com.alibaba.druid.hbase.HBaseResultSet;
public interface ExecutePlan {
void execute(HBaseConnection connection, List<Object> paramerers);
HBaseResultSet executeScan(HBasePreparedStatement statement) throws SQLException;
}
@@ -1,9 +1,14 @@
package com.alibaba.druid.hbase.exec;
import java.util.List;
import java.sql.SQLException;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import com.alibaba.druid.hbase.HBaseConnection;
import com.alibaba.druid.hbase.HBasePreparedStatement;
import com.alibaba.druid.hbase.HBaseResultSet;
public class SingleTableQueryExecutePlan implements ExecutePlan {
@@ -18,8 +23,20 @@ public class SingleTableQueryExecutePlan implements ExecutePlan {
}
@Override
public void execute(HBaseConnection connection, List<Object> paramerers) {
public HBaseResultSet executeScan(HBasePreparedStatement statement) throws SQLException {
try {
HBaseConnection connection = statement.getConnection();
HTableInterface htable = connection.getHTable(tableName);
Scan scan = new Scan();
ResultScanner scanner = htable.getScanner(scan);
return new HBaseResultSet(statement, htable, scanner);
} catch (SQLException e) {
throw e;
} catch (Exception e) {
throw new SQLException("executeQuery error", e);
}
}
}