Merge Whale

bee-engine/pom.xml

@@ -16,6 +16,26 @@
 			<groupId>com.site.common</groupId>
 			<artifactId>lookup</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.alibaba.cobar</groupId>
+			<artifactId>cobar-server</artifactId>
+			<version>1.3.0</version>
+		</dependency>
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+			<version>1.6.6</version>
+		</dependency>
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-log4j12</artifactId>
+			<version>1.6.6</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.commons</groupId>
+			<artifactId>commons-lang3</artifactId>
+			<version>3.1</version>
+		</dependency>
 	</dependencies>
 </project>
 

bee-engine/src/main/java/com/dianping/bee/server/Startup.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-15
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software s the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import org.apache.log4j.helpers.LogLog;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public final class Startup {
+	private static final String dateFormat = "yyyy-MM-dd HH:mm:ss";
+
+	public static void main(String[] args) {
+		try {
+			// init
+			WhaleServer server = WhaleServer.getInstance();
+			server.beforeStart(dateFormat);
+
+			// startup
+			server.startup();
+		} catch (Throwable e) {
+			SimpleDateFormat sdf = new SimpleDateFormat(dateFormat);
+			LogLog.error(sdf.format(new Date()) + " startup error", e);
+			System.exit(-1);
+		}
+	}
+
+}

bee-engine/src/main/java/com/dianping/bee/server/WhaleServer.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-15
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server;
+
+import java.io.IOException;
+
+import org.apache.log4j.Logger;
+
+import com.alibaba.cobar.CobarConfig;
+import com.alibaba.cobar.CobarServer;
+import com.alibaba.cobar.config.model.SystemConfig;
+import com.alibaba.cobar.net.NIOAcceptor;
+import com.dianping.whale.server.config.WhaleServerConfig;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class WhaleServer {
+
+	public static final String NAME = "Whale";
+
+	public static final String VERSION = "0.0.1";
+
+	private static final CobarServer COBAR_INSTANCE = CobarServer.getInstance();
+
+	private static final WhaleServer WHALE_INSTANCE = new WhaleServer();
+
+	private static final Logger LOGGER = Logger.getLogger(WhaleServer.class);
+
+	public static final WhaleServer getInstance() {
+		return WHALE_INSTANCE;
+	}
+
+	private final WhaleServerConfig whaleConfig;
+
+	private WhaleServer() {
+		this.whaleConfig = new WhaleServerConfig();
+	}
+
+	public void beforeStart(String dateFormat) {
+		COBAR_INSTANCE.beforeStart(dateFormat);
+	}
+
+	/**
+	 * @return
+	 */
+	public CobarConfig getConfig() {
+		return COBAR_INSTANCE.getConfig();
+	}
+
+	public void startup() throws IOException {
+		COBAR_INSTANCE.startup();
+
+		whaleConfig.load();
+		SystemConfig system = COBAR_INSTANCE.getConfig().getSystem();
+		// startup server
+		WhaleServerConnectionFactory sf = new WhaleServerConnectionFactory();
+		sf.setCharset(system.getCharset());
+		sf.setIdleTimeout(system.getIdleTimeout());
+
+		NIOAcceptor server = new NIOAcceptor(NAME + "Server", whaleConfig.getWhalePort(), sf);
+		server.setProcessors(COBAR_INSTANCE.getProcessors());
+		server.start();
+
+	}
+
+}

bee-engine/src/main/java/com/dianping/bee/server/WhaleServerConnection.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-17
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server;
+
+import java.nio.channels.SocketChannel;
+import java.util.Set;
+
+import com.alibaba.cobar.ErrorCode;
+import com.alibaba.cobar.net.util.MySQLMessage;
+import com.alibaba.cobar.protocol.mysql.OkPacket;
+import com.alibaba.cobar.server.ServerConnection;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class WhaleServerConnection extends ServerConnection {
+
+	/**
+	 * @param channel
+	 */
+	public WhaleServerConnection(SocketChannel channel) {
+		super(channel);
+	}
+
+	 public void initDB(byte[] data) {
+       MySQLMessage mm = new MySQLMessage(data);
+       mm.position(5);
+       String db = mm.readString();
+
+       // 检查schema是否已经设置
+//       if (schema != null) {
+//           if (schema.equals(db)) {
+//               write(writeToBuffer(OkPacket.OK, allocate()));
+//           } else {
+//               writeErrMessage(ErrorCode.ER_DBACCESS_DENIED_ERROR, "Not allowed to change the database!");
+//           }
+//           return;
+//       }
+
+       // 检查schema的有效性
+       if (db == null || !privileges.schemaExists(db)) {
+           writeErrMessage(ErrorCode.ER_BAD_DB_ERROR, "Unknown database '" + db + "'");
+           return;
+       }
+       if (!privileges.userExists(user, host)) {
+           writeErrMessage(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user '" + user + "'");
+           return;
+       }
+       Set<String> schemas = privileges.getUserSchemas(user);
+       if (schemas == null || schemas.size() == 0 || schemas.contains(db)) {
+           this.schema = db;
+           write(writeToBuffer(OkPacket.OK, allocate()));
+       } else {
+           String s = "Access denied for user '" + user + "' to database '" + db + "'";
+           writeErrMessage(ErrorCode.ER_DBACCESS_DENIED_ERROR, s);
+       }
+   }
+}

bee-engine/src/main/java/com/dianping/bee/server/WhaleServerConnectionFactory.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-17
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server;
+
+import java.nio.channels.SocketChannel;
+
+import com.alibaba.cobar.CobarPrivileges;
+import com.alibaba.cobar.CobarServer;
+import com.alibaba.cobar.config.model.SystemConfig;
+import com.alibaba.cobar.net.FrontendConnection;
+import com.alibaba.cobar.net.factory.FrontendConnectionFactory;
+import com.alibaba.cobar.server.ServerConnection;
+import com.alibaba.cobar.server.session.BlockingSession;
+import com.alibaba.cobar.server.session.NonBlockingSession;
+import com.dianping.whale.cobar.server.WhaleServerQueryHandler;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class WhaleServerConnectionFactory extends FrontendConnectionFactory {
+	@Override
+	protected FrontendConnection getConnection(SocketChannel channel) {
+		SystemConfig sys = CobarServer.getInstance().getConfig().getSystem();
+		ServerConnection c = new WhaleServerConnection(channel);
+		c.setPrivileges(new CobarPrivileges());
+		c.setQueryHandler(new WhaleServerQueryHandler(c));
+		c.setTxIsolation(sys.getTxIsolation());
+		c.setSession(new BlockingSession(c));
+		c.setSession2(new NonBlockingSession(c));
+		return c;
+	}
+}

bee-engine/src/main/java/com/dianping/bee/server/config/WhaleServerConfig.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-17
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server.config;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+
+import com.alibaba.cobar.config.loader.xml.XMLServerLoader;
+import com.alibaba.cobar.config.util.ConfigException;
+import com.alibaba.cobar.config.util.ConfigUtil;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class WhaleServerConfig {
+
+	private static final int DEFAULT_WHALE_PORT = 7066;
+
+	private int whalePort;
+
+	public WhaleServerConfig() {
+		whalePort = DEFAULT_WHALE_PORT;
+	}
+
+	public int getWhalePort() {
+		return this.whalePort;
+	}
+
+	public void load() {
+		InputStream dtd = null;
+		InputStream xml = null;
+		try {
+			dtd = XMLServerLoader.class.getResourceAsStream("/server.dtd");
+			xml = XMLServerLoader.class.getResourceAsStream("/server.xml");
+			Element root = ConfigUtil.getDocument(dtd, xml).getDocumentElement();
+			NodeList list = root.getElementsByTagName("system");
+			for (int i = 0, n = list.getLength(); i < n; i++) {
+				Node node = list.item(i);
+				if (node instanceof Element) {
+					Map<String, Object> props = ConfigUtil.loadElements((Element) node);
+					if (props.containsKey("whale")) {
+						this.whalePort = Integer.parseInt(props.get("whale").toString());
+						break;
+					}
+				}
+			}
+		} catch (ConfigException e) {
+			throw e;
+		} catch (Throwable e) {
+			throw new ConfigException(e);
+		} finally {
+			if (dtd != null) {
+				try {
+					dtd.close();
+				} catch (IOException e) {
+				}
+			}
+			if (xml != null) {
+				try {
+					xml.close();
+				} catch (IOException e) {
+				}
+			}
+		}
+	}
+
+}

bee-engine/src/main/java/com/dianping/bee/server/mysql/SelectHandler.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-14
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server.mysql;
+
+import com.alibaba.cobar.parser.util.ParseUtil;
+import com.alibaba.cobar.server.ServerConnection;
+import com.alibaba.cobar.server.parser.ServerParseSelect;
+import com.alibaba.cobar.server.response.SelectDatabase;
+import com.alibaba.cobar.server.response.SelectIdentity;
+import com.alibaba.cobar.server.response.SelectLastInsertId;
+import com.alibaba.cobar.server.response.SelectUser;
+import com.alibaba.cobar.server.response.SelectVersion;
+import com.alibaba.cobar.server.response.SelectVersionComment;
+import com.dianping.whale.cobar.response.SelectResponse;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class SelectHandler {
+
+	public static void handle(String stmt, ServerConnection c, int offs) {
+		int offset = offs;
+		switch (ServerParseSelect.parse(stmt, offs)) {
+		case ServerParseSelect.VERSION_COMMENT:
+			SelectVersionComment.response(c);
+			break;
+		case ServerParseSelect.DATABASE:
+			SelectDatabase.response(c);
+			break;
+		case ServerParseSelect.USER:
+			SelectUser.response(c);
+			break;
+		case ServerParseSelect.VERSION:
+			SelectVersion.response(c);
+			break;
+		case ServerParseSelect.LAST_INSERT_ID:
+			// offset = ParseUtil.move(stmt, 0, "select".length());
+			loop: for (; offset < stmt.length(); ++offset) {
+				switch (stmt.charAt(offset)) {
+				case ' ':
+					continue;
+				case '/':
+				case '#':
+					offset = ParseUtil.comment(stmt, offset);
+					continue;
+				case 'L':
+				case 'l':
+					break loop;
+				}
+			}
+			offset = ServerParseSelect.indexAfterLastInsertIdFunc(stmt, offset);
+			offset = ServerParseSelect.skipAs(stmt, offset);
+			SelectLastInsertId.response(c, stmt, offset);
+			break;
+		case ServerParseSelect.IDENTITY:
+			// offset = ParseUtil.move(stmt, 0, "select".length());
+			loop: for (; offset < stmt.length(); ++offset) {
+				switch (stmt.charAt(offset)) {
+				case ' ':
+					continue;
+				case '/':
+				case '#':
+					offset = ParseUtil.comment(stmt, offset);
+					continue;
+				case '@':
+					break loop;
+				}
+			}
+			int indexOfAtAt = offset;
+			offset += 2;
+			offset = ServerParseSelect.indexAfterIdentity(stmt, offset);
+			String orgName = stmt.substring(indexOfAtAt, offset);
+			offset = ServerParseSelect.skipAs(stmt, offset);
+			SelectIdentity.response(c, stmt, offset, orgName);
+			break;
+		default:
+			SelectResponse.response(c, stmt);
+		}
+	}
+}

bee-engine/src/main/java/com/dianping/bee/server/mysql/SelectResponse.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-14
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server.mysql;
+
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.log4j.Logger;
+
+import com.alibaba.cobar.ErrorCode;
+import com.alibaba.cobar.Fields;
+import com.alibaba.cobar.config.model.DataSourceConfig;
+import com.alibaba.cobar.config.model.SchemaConfig;
+import com.alibaba.cobar.net.util.PacketUtil;
+import com.alibaba.cobar.protocol.mysql.EOFPacket;
+import com.alibaba.cobar.protocol.mysql.FieldPacket;
+import com.alibaba.cobar.protocol.mysql.ResultSetHeaderPacket;
+import com.alibaba.cobar.protocol.mysql.RowDataPacket;
+import com.alibaba.cobar.server.ServerConnection;
+import com.alibaba.cobar.util.IntegerUtil;
+import com.alibaba.cobar.util.StringUtil;
+import com.dianping.whale.engine.IQueryInterface;
+import com.dianping.whale.server.WhaleServer;
+import com.dianping.whale.storage.RowSet;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class SelectResponse {
+
+	private static final Logger LOGGER = Logger.getLogger(SelectResponse.class);
+
+	private static final Map<String, IQueryInterface> queryInterfaces = new HashMap<String, IQueryInterface>();
+
+	/**
+	 * find Query Engine by schema
+	 * 
+	 * @param c
+	 * @return
+	 */
+	private static IQueryInterface getQueryInterface(ServerConnection c) {
+		String schema = c.getSchema();
+		SchemaConfig schemaConfig = WhaleServer.getInstance().getConfig().getSchemas().get(schema);
+		if (schemaConfig != null) {
+			String dataNode = schemaConfig.getDataNode();
+			String dataSourceName = WhaleServer.getInstance().getConfig().getDataNodes().get(dataNode).getSource()
+			      .getName();
+			DataSourceConfig dataSourceConfig = WhaleServer.getInstance().getConfig().getDataSources().get(dataSourceName);
+			String type = dataSourceConfig.getType();
+			if ("whale".equals(type)) {
+				String classPath = dataSourceConfig.getSqlMode();
+				IQueryInterface queryInstance = queryInterfaces.get(classPath);
+				if (queryInstance == null) {
+					try {
+						queryInstance = (IQueryInterface) Class.forName(classPath).newInstance();
+						queryInterfaces.put(classPath, queryInstance);
+					} catch (Exception e) {
+						LOGGER.warn("Instance Query Interface Failed", e);
+						return null;
+					}
+				}
+				return queryInstance;
+			} else if ("mysql".equals(type)) {
+			}
+		}
+		return null;
+	}
+
+	/**
+	 * 
+	 * @param rowData
+	 * @param rowType
+	 * @param charset
+	 * @return
+	 */
+	private static RowDataPacket getRow(Object[] rowData, int[] rowType, String charset) {
+		RowDataPacket row = new RowDataPacket(rowData.length);
+		for (int i = 0; i < rowData.length; i++) {
+			switch (rowType[i]) {
+			case Fields.FIELD_TYPE_STRING:
+				row.add(StringUtil.encode(rowData[i].toString(), charset));
+				break;
+			case Fields.FIELD_TYPE_INT24:
+				row.add(IntegerUtil.toBytes(Integer.parseInt(rowData[i].toString())));
+				break;
+			case Fields.FIELD_TYPE_DECIMAL:
+			case Fields.FIELD_TYPE_TINY:
+			case Fields.FIELD_TYPE_SHORT:
+			case Fields.FIELD_TYPE_LONG:
+			case Fields.FIELD_TYPE_FLOAT:
+			case Fields.FIELD_TYPE_DOUBLE:
+			case Fields.FIELD_TYPE_NULL:
+			case Fields.FIELD_TYPE_TIMESTAMP:
+			case Fields.FIELD_TYPE_LONGLONG:
+			case Fields.FIELD_TYPE_DATE:
+			case Fields.FIELD_TYPE_TIME:
+			case Fields.FIELD_TYPE_DATETIME:
+			case Fields.FIELD_TYPE_YEAR:
+			case Fields.FIELD_TYPE_NEWDATE:
+			case Fields.FIELD_TYPE_VARCHAR:
+			case Fields.FIELD_TYPE_BIT:
+			case Fields.FIELD_TYPE_NEW_DECIMAL:
+			case Fields.FIELD_TYPE_ENUM:
+			case Fields.FIELD_TYPE_SET:
+			case Fields.FIELD_TYPE_TINY_BLOB:
+			case Fields.FIELD_TYPE_MEDIUM_BLOB:
+			case Fields.FIELD_TYPE_LONG_BLOB:
+			case Fields.FIELD_TYPE_BLOB:
+			case Fields.FIELD_TYPE_VAR_STRING:
+			case Fields.FIELD_TYPE_GEOMETRY:
+			default:
+				row.add(StringUtil.encode(rowData[i].toString(), charset));
+			}
+		}
+		return row;
+	}
+
+	/**
+	 * 
+	 * @param c
+	 * @param stmt
+	 */
+	public static void response(ServerConnection c, String stmt) {
+		IQueryInterface queryInterface = getQueryInterface(c);
+		if (queryInterface == null) {
+			LOGGER.warn("No Query Interface Found");
+			c.writeErrMessage(ErrorCode.ER_BAD_DB_ERROR, stmt);
+			return;
+		}
+
+		RowSet result = null;
+		try {
+			result = queryInterface.query(stmt);
+		} catch (SQLException e) {
+			LOGGER.warn(stmt, e);
+			c.writeErrMessage(ErrorCode.ER_PARSE_ERROR, stmt);
+			return;
+		}
+
+		byte packetId = 0;
+		EOFPacket eof = new EOFPacket();
+		ByteBuffer buffer = c.allocate();
+		// write header
+		int fieldCount = result.getMetaData().getColumnCount();
+		ResultSetHeaderPacket header = PacketUtil.getHeader(fieldCount);
+		header.packetId = ++packetId;
+		buffer = header.write(buffer, c);
+
+		// write fields
+		int columnIndex = 0;
+		FieldPacket[] fields = new FieldPacket[fieldCount];
+		for (int i = 0; i < fieldCount; i++) {
+			fields[columnIndex] = PacketUtil.getField(result.getMetaData().getColumnNames()[i], result.getMetaData()
+			      .getColumnTypes()[i]);
+			fields[columnIndex++].packetId = ++packetId;
+		}
+		eof.packetId = ++packetId;
+
+		for (FieldPacket field : fields) {
+			buffer = field.write(buffer, c);
+		}
+		buffer = eof.write(buffer, c);
+
+		// write rows
+		for (int rowIndex = 0; rowIndex < result.getData().length; rowIndex++) {
+			RowDataPacket row = getRow(result.getData()[rowIndex], result.getMetaData().getColumnTypes(), c.getCharset());
+			row.packetId = ++packetId;
+			buffer = row.write(buffer, c);
+		}
+
+		EOFPacket lastEof = new EOFPacket();
+		lastEof.packetId = ++packetId;
+		buffer = lastEof.write(buffer, c);
+		c.write(buffer);
+	}
+}

bee-engine/src/main/java/com/dianping/bee/server/mysql/WhaleServerQueryHandler.java

+/**
+ * Project: bee-engine
+ * 
+ * File Created at 2012-8-15
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.server.mysql;
+
+import org.apache.log4j.Logger;
+
+import com.alibaba.cobar.ErrorCode;
+import com.alibaba.cobar.net.handler.FrontendQueryHandler;
+import com.alibaba.cobar.server.ServerConnection;
+import com.alibaba.cobar.server.handler.BeginHandler;
+import com.alibaba.cobar.server.handler.ExplainHandler;
+import com.alibaba.cobar.server.handler.KillHandler;
+import com.alibaba.cobar.server.handler.SavepointHandler;
+import com.alibaba.cobar.server.handler.SetHandler;
+import com.alibaba.cobar.server.handler.ShowHandler;
+import com.alibaba.cobar.server.handler.StartHandler;
+import com.alibaba.cobar.server.handler.UseHandler;
+import com.alibaba.cobar.server.parser.ServerParse;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class WhaleServerQueryHandler implements FrontendQueryHandler {
+	private static final Logger LOGGER = Logger.getLogger(WhaleServerQueryHandler.class);
+
+	private final ServerConnection source;
+
+	public WhaleServerQueryHandler(ServerConnection source) {
+		this.source = source;
+	}
+
+	@Override
+	public void query(String sql) {
+		ServerConnection c = this.source;
+		if (LOGGER.isDebugEnabled()) {
+			LOGGER.debug(new StringBuilder().append(c).append(sql).toString());
+		}
+		int rs = ServerParse.parse(sql);
+		switch (rs & 0xff) {
+		case ServerParse.EXPLAIN:
+			ExplainHandler.handle(sql, c, rs >>> 8);
+			break;
+		case ServerParse.SET:
+			SetHandler.handle(sql, c, rs >>> 8);
+			break;
+		case ServerParse.SHOW:
+			ShowHandler.handle(sql, c, rs >>> 8);
+			break;
+		case ServerParse.SELECT:
+			SelectHandler.handle(sql, c, rs >>> 8);
+			break;
+		case ServerParse.START:
+			StartHandler.handle(sql, c, rs >>> 8);
+			break;
+		case ServerParse.BEGIN:
+			BeginHandler.handle(sql, c);
+			break;
+		case ServerParse.SAVEPOINT:
+			SavepointHandler.handle(sql, c);
+			break;
+		case ServerParse.KILL:
+			KillHandler.handle(sql, rs >>> 8, c);
+			break;
+		case ServerParse.KILL_QUERY:
+			c.writeErrMessage(ErrorCode.ER_UNKNOWN_COM_ERROR, "Unsupported command");
+			break;
+		case ServerParse.USE:
+			UseHandler.handle(sql, c, rs >>> 8);
+			break;
+		case ServerParse.COMMIT:
+			c.commit();
+			break;
+		case ServerParse.ROLLBACK:
+			c.rollback();
+			break;
+		default:
+			c.execute(sql, rs);
+		}
+	}
+
+}

bee-engine/src/test/java/com/dianping/bee/engine/spi/impl/Department.java

+/**
+ * Project: whale-engine
+ * 
+ * File Created at 2012-8-15
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.engine.spi.impl;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class Department {
+
+	private Integer Id;
+
+	private String name;
+
+	public Integer getId() {
+		return Id;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setId(Integer id) {
+		Id = id;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+}

bee-engine/src/test/java/com/dianping/bee/engine/spi/impl/SampleDB1.java

+/**
+ * Project: whale-engine
+ * 
+ * File Created at 2012-8-14
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.engine.spi.impl;
+
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.lang3.RandomStringUtils;
+
+import com.alibaba.cobar.Fields;
+import com.alibaba.cobar.parser.ast.expression.primary.Identifier;
+import com.alibaba.cobar.parser.ast.stmt.SQLStatement;
+import com.alibaba.cobar.parser.recognizer.SQLParserDelegate;
+import com.dianping.bee.engine.visitor.DefaultVisitor;
+import com.dianping.whale.engine.IQueryInterface;
+import com.dianping.whale.storage.RowSet;
+import com.dianping.whale.storage.RowSetMetaData;
+import com.dianping.whale.storage.model.Department;
+import com.dianping.whale.storage.model.User;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class SampleDB1 implements IQueryInterface {
+
+	private static final SampleDB1 INSTANCE = new SampleDB1();
+
+	private static final int SAMPLE_DATA_BASE_SIZE = 10;
+
+	private static Map<String, Object[][]> tableData;
+
+	private static Map<String, String[]> columnMeta;
+
+	private static final String TABLE_USER_NAME = "USER";
+
+	private static final String TABLE_DEPARTMENT_NAME = "DEPARTMENT";
+
+	static {
+		loadSampleData();
+	}
+
+	public static SampleDB1 getInstance() {
+		return INSTANCE;
+	}
+
+	private static void loadSampleData() {
+		tableData = new HashMap<String, Object[][]>();
+		tableData.put(TABLE_USER_NAME, new Object[SAMPLE_DATA_BASE_SIZE * 5][]);
+		tableData.put(TABLE_DEPARTMENT_NAME, new Object[SAMPLE_DATA_BASE_SIZE][]);
+
+		columnMeta = new HashMap<String, String[]>();
+		columnMeta.put(TABLE_USER_NAME, new String[] { "ID", "NAME", "ADDRESS", "DEPARTMENTID" });
+		for (int i = 0; i < SAMPLE_DATA_BASE_SIZE * 5; i++) {
+			User user = new User();
+			user.setId(i + 1);
+			user.setName(RandomStringUtils.randomAlphabetic(5));
+			user.setAddress(RandomStringUtils.randomAlphabetic(10));
+			user.setDepartmentId(i % SAMPLE_DATA_BASE_SIZE + 1);
+
+			Object[] row = new Object[4];
+			row[0] = user.getId();
+			row[1] = user.getName();
+			row[2] = user.getAddress();
+			row[3] = user.getDepartmentId();
+			tableData.get(TABLE_USER_NAME)[i] = row;
+		}
+
+		columnMeta.put(TABLE_DEPARTMENT_NAME, new String[] { "ID", "NAME" });
+		for (int i = 0; i < SAMPLE_DATA_BASE_SIZE; i++) {
+			Department dept = new Department();
+			dept.setId(i + 1);
+			dept.setName(RandomStringUtils.randomAlphabetic(3));
+
+			Object[] row = new Object[2];
+			row[0] = dept.getId();
+			row[1] = dept.getName();
+			tableData.get(TABLE_DEPARTMENT_NAME)[i] = row;
+		}
+	}
+
+	public RowSet query(String sql) throws SQLException {
+		SQLStatement statement = null;
+		statement = SQLParserDelegate.parse(sql);
+
+		DefaultVisitor visitor = new DefaultVisitor();
+		statement.accept(visitor);
+		Map<String, Identifier> tables = visitor.getTables();
+		Identifier table = tables.values().iterator().next();
+		String tableName = table.getIdText().toUpperCase();
+
+		Object[][] queryData = tableData.get(tableName);
+
+		RowSet rowSet = new RowSet();
+		rowSet.setData(queryData);
+
+		String[] columnNames = columnMeta.get(tableName);
+		int[] columnTypes = new int[columnNames.length];
+		for (int i = 0; i < columnNames.length; i++) {
+			columnTypes[i] = Fields.FIELD_TYPE_STRING;
+		}
+
+		RowSetMetaData metaData = new RowSetMetaData();
+		metaData.setColumnNames(columnNames);
+		metaData.setColumnTypes(columnTypes);
+		rowSet.setMetaData(metaData);
+
+		return rowSet;
+	}
+
+}

bee-engine/src/test/java/com/dianping/bee/engine/spi/impl/SampleDB2.java

+/**
+ * Project: whale-engine
+ * 
+ * File Created at 2012-8-14
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.engine.spi.impl;
+
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.lang3.RandomStringUtils;
+
+import com.alibaba.cobar.Fields;
+import com.alibaba.cobar.parser.ast.expression.primary.Identifier;
+import com.alibaba.cobar.parser.ast.stmt.SQLStatement;
+import com.alibaba.cobar.parser.recognizer.SQLParserDelegate;
+import com.dianping.bee.engine.visitor.DefaultVisitor;
+import com.dianping.whale.engine.IQueryInterface;
+import com.dianping.whale.storage.RowSet;
+import com.dianping.whale.storage.RowSetMetaData;
+import com.dianping.whale.storage.model.Department;
+import com.dianping.whale.storage.model.User;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class SampleDB2 implements IQueryInterface {
+
+	private static final SampleDB2 INSTANCE = new SampleDB2();
+
+	private static final int SAMPLE_DATA_BASE_SIZE = 5;
+
+	private static Map<String, Object[][]> tableData;
+
+	private static Map<String, String[]> columnMeta;
+
+	private static final String TABLE_USER_NAME = "USER";
+
+	private static final String TABLE_DEPARTMENT_NAME = "DEPARTMENT";
+
+	static {
+		loadSampleData();
+	}
+
+	public static SampleDB2 getInstance() {
+		return INSTANCE;
+	}
+
+	private static void loadSampleData() {
+		tableData = new HashMap<String, Object[][]>();
+		tableData.put(TABLE_USER_NAME, new Object[SAMPLE_DATA_BASE_SIZE * 5][]);
+		tableData.put(TABLE_DEPARTMENT_NAME, new Object[SAMPLE_DATA_BASE_SIZE][]);
+
+		columnMeta = new HashMap<String, String[]>();
+		columnMeta.put(TABLE_USER_NAME, new String[] { "ID", "NAME", "ADDRESS", "DEPARTMENTID" });
+		for (int i = 0; i < SAMPLE_DATA_BASE_SIZE * 5; i++) {
+			User user = new User();
+			user.setId(i + 1);
+			user.setName(RandomStringUtils.randomAlphabetic(5));
+			user.setAddress(RandomStringUtils.randomAlphabetic(10));
+			user.setDepartmentId(i % SAMPLE_DATA_BASE_SIZE + 1);
+
+			Object[] row = new Object[4];
+			row[0] = user.getId();
+			row[1] = user.getName();
+			row[2] = user.getAddress();
+			row[3] = user.getDepartmentId();
+			tableData.get(TABLE_USER_NAME)[i] = row;
+		}
+
+		columnMeta.put(TABLE_DEPARTMENT_NAME, new String[] { "ID", "NAME" });
+		for (int i = 0; i < SAMPLE_DATA_BASE_SIZE; i++) {
+			Department dept = new Department();
+			dept.setId(i + 1);
+			dept.setName(RandomStringUtils.randomAlphabetic(3));
+
+			Object[] row = new Object[2];
+			row[0] = dept.getId();
+			row[1] = dept.getName();
+			tableData.get(TABLE_DEPARTMENT_NAME)[i] = row;
+		}
+	}
+
+	public RowSet query(String sql) throws SQLException {
+		SQLStatement statement = null;
+		statement = SQLParserDelegate.parse(sql);
+
+		DefaultVisitor visitor = new DefaultVisitor();
+		statement.accept(visitor);
+		Map<String, Identifier> tables = visitor.getTables();
+		Identifier table = tables.values().iterator().next();
+		String tableName = table.getIdText().toUpperCase();
+
+		Object[][] queryData = tableData.get(tableName);
+
+		RowSet rowSet = new RowSet();
+		rowSet.setData(queryData);
+
+		String[] columnNames = columnMeta.get(tableName);
+		int[] columnTypes = new int[columnNames.length];
+		for (int i = 0; i < columnNames.length; i++) {
+			columnTypes[i] = Fields.FIELD_TYPE_STRING;
+		}
+
+		RowSetMetaData metaData = new RowSetMetaData();
+		metaData.setColumnNames(columnNames);
+		metaData.setColumnTypes(columnTypes);
+		rowSet.setMetaData(metaData);
+
+		return rowSet;
+	}
+}

bee-engine/src/test/java/com/dianping/bee/engine/spi/impl/User.java

+/**
+ * Project: whale-engine
+ * 
+ * File Created at 2012-8-17
+ * 
+ * Copyright 2012 dianping.com.
+ * All rights reserved.
+ *
+ * This software is the confidential and proprietary information of
+ * Dianping Company. ("Confidential Information").  You shall not
+ * disclose such Confidential Information and shall use it only in
+ * accordance with the terms of the license agreement you entered into
+ * with dianping.com.
+ */
+package com.dianping.bee.engine.spi.impl;
+
+/**
+ * @author <a href="mailto:yiming.liu@dianping.com">Yiming Liu</a>
+ */
+public class User {
+
+	private Integer Id;
+
+	private String name;
+
+	private String address;
+
+	private Integer departmentId;
+
+	public String getAddress() {
+		return address;
+	}
+
+	public Integer getDepartmentId() {
+		return departmentId;
+	}
+
+	public Integer getId() {
+		return Id;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setAddress(String address) {
+		this.address = address;
+	}
+
+	public void setDepartmentId(Integer departmentId) {
+		this.departmentId = departmentId;
+	}
+
+	public void setId(Integer id) {
+		Id = id;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+}