hsqldb源码分析系列2 解析器分析
在server类的handleConnection方法中处理客户端的输入，
首先调用allowConnection方法检测下客户端ip是否合法，ServerAcl类中可以查看，在server启动输入参数中可以指定acl 黑白名单ip文件，检测代码很简单，这里就不说明了，有兴趣的看下ServerAcl类。
protected boolean allowConnection(Socket socket) { if (isShuttingDown) { return false; } return (acl == null) ? true : acl.permitAccess( socket.getInetAddress().getAddress()); }然后通过参数判断server启动模式，是hsql模式还是http模式，其实就是启动端口不一样，然后格式不一样而已。
if (serverProtocol == ServerConstants.SC_PROTOCOL_HSQL) { r = new ServerConnection(s, this); ctn = ((ServerConnection) r).getConnectionThreadName(); } else { r = new WebServerConnection(s, (WebServer) this); ctn = ((WebServerConnection) r).getConnectionThreadName(); } t = new Thread(serverConnectionThreadGroup, r, ctn); t.start();
启动一个新线程处理客户端连接，我们先看看ServerConnection的处理，每个ServerCOnnection会保存一个线程递增的id，然后往server添加当前连接
mThread = mCurrentThread.getAndIncrement(); synchronized (server.serverConnSet) { server.serverConnSet.add(this); }
看看run方法，init是处理第一次连接的请求，后面while循环处理该连接的下一个请求，最后close方法关闭连接。
public void run() { int msgType; init(); if (session != null) { try { while (keepAlive) { msgType = dataInput.readByte(); if (msgType < ResultConstants.MODE_UPPER_LIMIT) { receiveResult(msgType); } else { receiveOdbcPacket((char) msgType); } } ......close();} init方法，handshake 方法是查下当前连接请求时哪种类型的连接
private void init() { runnerThread = Thread.currentThread(); keepAlive = true; try { socket.setTcpNoDelay(true); dataInput = new DataInputStream( new BufferedInputStream(socket.getInputStream())); dataOutput = new DataOutputStream(socket.getOutputStream()); int firstInt = handshake(); switch (streamProtocol) { case HSQL_STREAM_PROTOCOL : if (firstInt != ClientConnection .NETWORK_COMPATIBILITY_VERSION_INT) { if (firstInt == -1900000) { firstInt = -2000000; } String verString = ClientConnection.toNetCompVersionString(firstInt); throw Error.error( null, ErrorCode.SERVER_VERSIONS_INCOMPATIBLE, 0, new String[] { verString, HsqlDatabaseProperties.hsqldb_version }); } Result resultIn = Result.newResult(dataInput, rowIn); resultIn.readAdditionalResults(session, dataInput, rowIn); Result resultOut; resultOut = setDatabase(resultIn); resultOut.write(session, dataOutput, rowOut); break; case ODBC_STREAM_PROTOCOL : odbcConnect(firstInt); break; default : // Protocol detection failures should already have been // handled. keepAlive = false; } } catch (Exception e) { 前四个字节的最高位表示连接类型，这个整数还表示版本号，
int firstInt = dataInput.readInt(); switch (firstInt >> 24) { case 80 : // Empirically server.print( "Rejected attempt from client using hsql HTTP protocol"); return 0; case 0 : // For ODBC protocol, this is the first byte of a 4-byte int // size. The size can never be large enough that the first // byte will be non-zero. streamProtocol = ODBC_STREAM_PROTOCOL; break; default : streamProtocol = HSQL_STREAM_PROTOCOL; // HSQL protocol client } 最重要的是中间的输入解析，查询和结果封装部分。具体看看
第一次的处理session为null，后面会赋值，看看Result resultIn = Result.newResult(dataInput, rowIn);的核心部分，第一个字节=LARGE_OBJECT_OP表示是查询大数据对象内容，后面再看具体字段
public static Result newResult(DataInput dataInput, RowInputBinary in) throws IOException, HsqlException { return newResult(null, dataInput.readByte(), dataInput, in); } public static Result newResult(Session session, int mode, DataInput dataInput, RowInputBinary in) throws IOException, HsqlException { try { if (mode == ResultConstants.LARGE_OBJECT_OP) { return ResultLob.newLob(dataInput, false); } Result result = newResult(session, dataInput, in, mode); return result; } catch (IOException e) { throw Error.error(ErrorCode.X_08000); } } newResult 先根据不同的请求类型或者各种类型需要的一些字段信息，前4个字节是请求长度，把整个长度内容copy到in中，然后可以用封装好的RowInputBinary读取字符串，整数等信息。 private static Result newResult(Session session, DataInput dataInput, RowInputBinary in, int mode) throws IOException, HsqlException { Result result = newResult(mode); int length = dataInput.readInt(); in.resetRow(0, length); byte[] byteArray = in.getBuffer(); final int offset = 4; dataInput.readFully(byteArray, offset, length - offset); switch (mode) { case ResultConstants.GETSESSIONATTR : result.statementReturnType = in.readByte(); break; case ResultConstants.DISCONNECT : case ResultConstants.RESETSESSION : case ResultConstants.STARTTRAN : break; case ResultConstants.PREPARE : result.setStatementType(in.readByte()); result.mainString = in.readString(); result.rsProperties = in.readByte(); result.generateKeys = in.readByte(); if (result.generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_NAMES || result .generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_INDEXES) { result.generatedMetaData = new ResultMetaData(in); } break; case ResultConstants.CLOSE_RESULT : result.id = in.readLong(); break; case ResultConstants.FREESTMT : result.statementID = in.readLong(); break; case ResultConstants.EXECDIRECT : result.updateCount = in.readInt(); result.fetchSize = in.readInt(); result.statementReturnType = in.readByte(); result.mainString = in.readString(); result.rsProperties = in.readByte(); result.queryTimeout = in.readShort(); result.generateKeys = in.readByte(); if (result.generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_NAMES || result .generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_INDEXES) { result.generatedMetaData = new ResultMetaData(in); } break; case ResultConstants.CONNECT : result.databaseName = in.readString(); result.mainString = in.readString(); result.subString = in.readString(); result.zoneString = in.readString(); result.updateCount = in.readInt(); break; case ResultConstants.ERROR : case ResultConstants.WARNING : result.mainString = in.readString(); result.subString = in.readString(); result.errorCode = in.readInt(); break; case ResultConstants.CONNECTACKNOWLEDGE : result.databaseID = in.readInt(); result.sessionID = in.readLong(); result.databaseName = in.readString(); result.mainString = in.readString(); break; case ResultConstants.UPDATECOUNT : result.updateCount = in.readInt(); break; case ResultConstants.ENDTRAN : { int type = in.readInt(); result.setActionType(type); // endtran type switch (type) { case ResultConstants.TX_SAVEPOINT_NAME_RELEASE : case ResultConstants.TX_SAVEPOINT_NAME_ROLLBACK : result.mainString = in.readString(); // savepoint name break; case ResultConstants.TX_COMMIT : case ResultConstants.TX_ROLLBACK : case ResultConstants.TX_COMMIT_AND_CHAIN : case ResultConstants.TX_ROLLBACK_AND_CHAIN : break; default : throw Error.runtimeError(ErrorCode.U_S0500, "Result"); } break; } case ResultConstants.SETCONNECTATTR : { int type = in.readInt(); // attr type result.setConnectionAttrType(type); switch (type) { case ResultConstants.SQL_ATTR_SAVEPOINT_NAME : result.mainString = in.readString(); // savepoint name break; // case ResultConstants.SQL_ATTR_AUTO_IPD : // - always true // default: throw - case never happens default : throw Error.runtimeError(ErrorCode.U_S0500, "Result"); } break; } case ResultConstants.PREPARE_ACK : result.statementReturnType = in.readByte(); result.statementID = in.readLong(); result.rsProperties = in.readByte(); result.metaData = new ResultMetaData(in); result.parameterMetaData = new ResultMetaData(in); break; case ResultConstants.CALL_RESPONSE : result.updateCount = in.readInt(); result.fetchSize = in.readInt(); result.statementID = in.readLong(); result.statementReturnType = in.readByte(); result.rsProperties = in.readByte(); result.metaData = new ResultMetaData(in); result.valueData = readSimple(in, result.metaData); break; case ResultConstants.EXECUTE : result.updateCount = in.readInt(); result.fetchSize = in.readInt(); result.statementID = in.readLong(); result.rsProperties = in.readByte(); result.queryTimeout = in.readShort(); Statement statement = session.statementManager.getStatement(session, result.statementID); if (statement == null) { // invalid statement result.mode = ResultConstants.EXECUTE_INVALID; result.valueData = ValuePool.emptyObjectArray; break; } result.statement = statement; result.metaData = result.statement.getParametersMetaData(); result.valueData = readSimple(in, result.metaData); break; case ResultConstants.UPDATE_RESULT : { result.id = in.readLong(); int type = in.readInt(); result.setActionType(type); result.metaData = new ResultMetaData(in); result.valueData = readSimple(in, result.metaData); break; } case ResultConstants.BATCHEXECRESPONSE : case ResultConstants.BATCHEXECUTE : case ResultConstants.BATCHEXECDIRECT : case ResultConstants.SETSESSIONATTR : { result.updateCount = in.readInt(); result.fetchSize = in.readInt(); result.statementID = in.readLong(); result.queryTimeout = in.readShort(); result.metaData = new ResultMetaData(in); result.navigator.readSimple(in, result.metaData); break; } case ResultConstants.PARAM_METADATA : { result.metaData = new ResultMetaData(in); result.navigator.read(in, result.metaData); break; } case ResultConstants.REQUESTDATA : { result.id = in.readLong(); result.updateCount = in.readInt(); result.fetchSize = in.readInt(); break; } case ResultConstants.DATAHEAD : case ResultConstants.DATA : case ResultConstants.GENERATED : { result.id = in.readLong(); result.updateCount = in.readInt(); result.fetchSize = in.readInt(); result.rsProperties = in.readByte(); result.metaData = new ResultMetaData(in); result.navigator = new RowSetNavigatorClient(); result.navigator.read(in, result.metaData); break; } case ResultConstants.DATAROWS : { result.metaData = new ResultMetaData(in); result.navigator = new RowSetNavigatorClient(); result.navigator.read(in, result.metaData); break; } default : throw Error.runtimeError(ErrorCode.U_S0500, "Result"); } return result; } readAdditionalResults 是读取预取数据，取下一部分数据的处理；
第一次是connect，则会调用setdatabse设置连接的session信息（数据库等）
private Result setDatabase(Result resultIn) { try { String databaseName = resultIn.getDatabaseName(); dbIndex = server.getDBIndex(databaseName); dbID = server.dbID[dbIndex]; user = resultIn.getMainString(); if (!server.isSilent()) { server.printWithThread(mThread + ":Trying to connect user '" + user + "' to DB (" + databaseName + ')'); } session = DatabaseManager.newSession(dbID, user, resultIn.getSubString(), resultIn.getZoneString(), resultIn.getUpdateCount()); if (!server.isSilent()) { server.printWithThread(mThread + ":Connected user '" + user + "'"); } return Result.newConnectionAcknowledgeResponse( session.getDatabase(), session.getId(), session.getDatabase().getDatabaseID()); } catch (HsqlException e) { session = null; return Result.newErrorResult(e); } catch (RuntimeException e) { session = null; return Result.newErrorResult(e); } } DataBaseManager的newsession
public static Session newSession(int dbID, String user, String password, String zoneString, int timeZoneSeconds) { Database db = null; synchronized (databaseIDMap) { db = (Database) databaseIDMap.get(dbID); } if (db == null) { return null; } Session session = db.connect(user, password, zoneString, timeZoneSeconds); session.isNetwork = true; return session; } 最终调用Session类的new Session，默认事务是自动提交的。readonly是对应数据库设置是否为只读 Session(Database db, User user, boolean autocommit, boolean readonly, long id, String zoneString, int timeZoneSeconds) { sessionId = id; database = db; this.user = user; this.sessionUser = user; this.zoneString = zoneString; this.sessionTimeZoneSeconds = timeZoneSeconds; this.timeZoneSeconds = timeZoneSeconds; rowActionList = new HsqlArrayList(32, true); waitedSessions = new OrderedHashSet(); waitingSessions = new OrderedHashSet(); tempSet = new OrderedHashSet(); isolationLevelDefault = database.defaultIsolationLevel; isolationLevel = isolationLevelDefault; txConflictRollback = database.txConflictRollback; isReadOnlyDefault = readonly; isReadOnlyIsolation = isolationLevel == SessionInterface.TX_READ_UNCOMMITTED; sessionContext = new SessionContext(this); sessionContext.isAutoCommit = autocommit ? Boolean.TRUE : Boolean.FALSE; sessionContext.isReadOnly = isReadOnlyDefault ? Boolean.TRUE : Boolean.FALSE; parser = new ParserCommand(this, new Scanner()); setResultMemoryRowCount(database.getResultMaxMemoryRows()); resetSchema(); sessionData = new SessionData(database, this); statementManager = new StatementManager(database); } sessionManager当然会保存当前的session和session连接数
Session s = new Session(db, user, autoCommit, readonly, sessionIdCount, zoneString, timeZoneSeconds); sessionMap.put(sessionIdCount, s); sessionIdCount++;最后write
public void write(SessionInterface session, DataOutputStream dataOut, RowOutputInterface rowOut) throws IOException, HsqlException { rowOut.reset(); rowOut.writeByte(mode); int startPos = rowOut.size(); rowOut.writeSize(0); switch (mode) { case ResultConstants.GETSESSIONATTR : rowOut.writeByte(statementReturnType); break; case ResultConstants.DISCONNECT : case ResultConstants.RESETSESSION : case ResultConstants.STARTTRAN : break; case ResultConstants.PREPARE : rowOut.writeByte(statementReturnType); rowOut.writeString(mainString); rowOut.writeByte(rsProperties); rowOut.writeByte(generateKeys); if (generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_NAMES || generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_INDEXES) { generatedMetaData.write(rowOut); } break; case ResultConstants.FREESTMT : rowOut.writeLong(statementID); break; case ResultConstants.CLOSE_RESULT : rowOut.writeLong(id); break; case ResultConstants.EXECDIRECT : rowOut.writeInt(updateCount); rowOut.writeInt(fetchSize); rowOut.writeByte(statementReturnType); rowOut.writeString(mainString); rowOut.writeByte(rsProperties); rowOut.writeShort(queryTimeout); rowOut.writeByte(generateKeys); if (generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_NAMES || generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_INDEXES) { generatedMetaData.write(rowOut); } break; case ResultConstants.CONNECT : rowOut.writeString(databaseName); rowOut.writeString(mainString); rowOut.writeString(subString); rowOut.writeString(zoneString); rowOut.writeInt(updateCount); break; case ResultConstants.ERROR : case ResultConstants.WARNING : rowOut.writeString(mainString); rowOut.writeString(subString); rowOut.writeInt(errorCode); break; case ResultConstants.CONNECTACKNOWLEDGE : rowOut.writeInt(databaseID); rowOut.writeLong(sessionID); rowOut.writeString(databaseName); rowOut.writeString(mainString); break; case ResultConstants.UPDATECOUNT : rowOut.writeInt(updateCount); break; case ResultConstants.ENDTRAN : { int type = getActionType(); rowOut.writeInt(type); // endtran type switch (type) { case ResultConstants.TX_SAVEPOINT_NAME_RELEASE : case ResultConstants.TX_SAVEPOINT_NAME_ROLLBACK : rowOut.writeString(mainString); // savepoint name break; case ResultConstants.TX_COMMIT : case ResultConstants.TX_ROLLBACK : case ResultConstants.TX_COMMIT_AND_CHAIN : case ResultConstants.TX_ROLLBACK_AND_CHAIN : break; default : throw Error.runtimeError(ErrorCode.U_S0500, "Result"); } break; } case ResultConstants.PREPARE_ACK : rowOut.writeByte(statementReturnType); rowOut.writeLong(statementID); rowOut.writeByte(rsProperties); metaData.write(rowOut); parameterMetaData.write(rowOut); break; case ResultConstants.CALL_RESPONSE : rowOut.writeInt(updateCount); rowOut.writeInt(fetchSize); rowOut.writeLong(statementID); rowOut.writeByte(statementReturnType); rowOut.writeByte(rsProperties); metaData.write(rowOut); writeSimple(rowOut, metaData, (Object[]) valueData); break; case ResultConstants.EXECUTE : rowOut.writeInt(updateCount); rowOut.writeInt(fetchSize); rowOut.writeLong(statementID); rowOut.writeByte(rsProperties); rowOut.writeShort(queryTimeout); writeSimple(rowOut, metaData, (Object[]) valueData); break; case ResultConstants.UPDATE_RESULT : rowOut.writeLong(id); rowOut.writeInt(getActionType()); metaData.write(rowOut); writeSimple(rowOut, metaData, (Object[]) valueData); break; case ResultConstants.BATCHEXECRESPONSE : case ResultConstants.BATCHEXECUTE : case ResultConstants.BATCHEXECDIRECT : case ResultConstants.SETSESSIONATTR : { rowOut.writeInt(updateCount); rowOut.writeInt(fetchSize); rowOut.writeLong(statementID); rowOut.writeShort(queryTimeout); metaData.write(rowOut); navigator.writeSimple(rowOut, metaData); break; } case ResultConstants.PARAM_METADATA : { metaData.write(rowOut); navigator.write(rowOut, metaData); break; } case ResultConstants.SETCONNECTATTR : { int type = getConnectionAttrType(); rowOut.writeInt(type); // attr type / updateCount switch (type) { case ResultConstants.SQL_ATTR_SAVEPOINT_NAME : rowOut.writeString(mainString); // savepoint name break; // case ResultConstants.SQL_ATTR_AUTO_IPD // always true // default: // throw, but case never happens default : throw Error.runtimeError(ErrorCode.U_S0500, "Result"); } break; } case ResultConstants.REQUESTDATA : { rowOut.writeLong(id); rowOut.writeInt(updateCount); rowOut.writeInt(fetchSize); break; } case ResultConstants.DATAROWS : metaData.write(rowOut); navigator.write(rowOut, metaData); break; case ResultConstants.DATAHEAD : case ResultConstants.DATA : case ResultConstants.GENERATED : rowOut.writeLong(id); rowOut.writeInt(updateCount); rowOut.writeInt(fetchSize); rowOut.writeByte(rsProperties); metaData.write(rowOut); navigator.write(rowOut, metaData); break; default : throw Error.runtimeError(ErrorCode.U_S0500, "Result"); } rowOut.writeIntData(rowOut.size() - startPos, startPos); dataOut.write(rowOut.getOutputStream().getBuffer(), 0, rowOut.size()); int count = getLobCount(); Result current = this; for (int i = 0; i < count; i++) { ResultLob lob = current.lobResults; lob.writeBody(session, dataOut); current = current.lobResults; } if (chainedResult == null) { dataOut.writeByte(ResultConstants.NONE); } else { chainedResult.write(session, dataOut, rowOut); } dataOut.flush(); }
然后在server的run方法中循环中处理完连接请求之后就可以执行数据库操作了。msgType大于 MODE_UPPER_LIMIT表示不是hsqldb的包类型而是odbc包类型
while (keepAlive) { msgType = dataInput.readByte(); if (msgType < ResultConstants.MODE_UPPER_LIMIT) { receiveResult(msgType); } else { receiveOdbcPacket((char) msgType); } } private void receiveResult(int resultMode) throws CleanExit, IOException { boolean terminate = false; Result resultIn = Result.newResult(session, resultMode, dataInput, rowIn); resultIn.readLobResults(session, dataInput, rowIn); server.printRequest(mThread, resultIn); Result resultOut = null; switch (resultIn.getType()) { case ResultConstants.CONNECT : { resultOut = setDatabase(resultIn); break; } case ResultConstants.DISCONNECT : { resultOut = Result.updateZeroResult; terminate = true; break; } case ResultConstants.RESETSESSION : { session.resetSession(); resultOut = Result.updateZeroResult; break; } case ResultConstants.EXECUTE_INVALID : { resultOut = Result.newErrorResult(Error.error(ErrorCode.X_07502)); break; } default : { resultOut = session.execute(resultIn); break; } } resultOut.write(session, dataOutput, rowOut); rowOut.reset(mainBuffer); rowIn.resetRow(mainBuffer.length); if (terminate) { throw cleanExit; } } session的excute方法执行的主流程
public synchronized Result execute(Result cmd) { if (isClosed) { return Result.newErrorResult(Error.error(ErrorCode.X_08503)); } sessionContext.currentMaxRows = 0; isBatch = false; JavaSystem.gc(); switch (cmd.mode) { case ResultConstants.LARGE_OBJECT_OP : { return performLOBOperation((ResultLob) cmd); } case ResultConstants.EXECUTE : { int maxRows = cmd.getUpdateCount(); if (maxRows == -1) { sessionContext.currentMaxRows = 0; } else { sessionContext.currentMaxRows = maxRows; } Statement cs = cmd.statement; if (cs == null || cs.compileTimestamp < database.schemaManager.schemaChangeTimestamp) { long csid = cmd.getStatementID(); cs = statementManager.getStatement(this, csid); cmd.setStatement(cs); if (cs == null) { // invalid sql has been removed already return Result.newErrorResult( Error.error(ErrorCode.X_07502)); } } Object[] pvals = (Object[]) cmd.valueData; Result result = executeCompiledStatement(cs, pvals); result = performPostExecute(cmd, result); return result; } case ResultConstants.BATCHEXECUTE : { isBatch = true; Result result = executeCompiledBatchStatement(cmd); result = performPostExecute(cmd, result); return result; } case ResultConstants.EXECDIRECT : { Result result = executeDirectStatement(cmd); result = performPostExecute(cmd, result); return result; } case ResultConstants.BATCHEXECDIRECT : { isBatch = true; Result result = executeDirectBatchStatement(cmd); result = performPostExecute(cmd, result); return result; } case ResultConstants.PREPARE : { Statement cs; try { cs = statementManager.compile(this, cmd); } catch (Throwable t) { String errorString = cmd.getMainString(); if (database.getProperties().getErrorLevel() == HsqlDatabaseProperties.NO_MESSAGE) { errorString = null; } return Result.newErrorResult(t, errorString); } Result result = Result.newPrepareResponse(cs); if (cs.getType() == StatementTypes.SELECT_CURSOR || cs.getType() == StatementTypes.CALL) { sessionData.setResultSetProperties(cmd, result); } result = performPostExecute(cmd, result); return result; } case ResultConstants.CLOSE_RESULT : { closeNavigator(cmd.getResultId()); return Result.updateZeroResult; } case ResultConstants.UPDATE_RESULT : { Result result = this.executeResultUpdate(cmd); result = performPostExecute(cmd, result); return result; } case ResultConstants.FREESTMT : { statementManager.freeStatement(cmd.getStatementID()); return Result.updateZeroResult; } case ResultConstants.GETSESSIONATTR : { int id = cmd.getStatementType(); return getAttributesResult(id); } case ResultConstants.SETSESSIONATTR : { return setAttributes(cmd); } case ResultConstants.ENDTRAN : { switch (cmd.getActionType()) { case ResultConstants.TX_COMMIT : try { commit(false); } catch (Throwable t) { return Result.newErrorResult(t); } break; case ResultConstants.TX_COMMIT_AND_CHAIN : try { commit(true); } catch (Throwable t) { return Result.newErrorResult(t); } break; case ResultConstants.TX_ROLLBACK : rollback(false); break; case ResultConstants.TX_ROLLBACK_AND_CHAIN : rollback(true); break; case ResultConstants.TX_SAVEPOINT_NAME_RELEASE : try { String name = cmd.getMainString(); releaseSavepoint(name); } catch (Throwable t) { return Result.newErrorResult(t); } break; case ResultConstants.TX_SAVEPOINT_NAME_ROLLBACK : try { rollbackToSavepoint(cmd.getMainString()); } catch (Throwable t) { return Result.newErrorResult(t); } break; case ResultConstants.PREPARECOMMIT : try { prepareCommit(); } catch (Throwable t) { return Result.newErrorResult(t); } break; } return Result.updateZeroResult; } case ResultConstants.SETCONNECTATTR : { switch (cmd.getConnectionAttrType()) { case ResultConstants.SQL_ATTR_SAVEPOINT_NAME : try { savepoint(cmd.getMainString()); } catch (Throwable t) { return Result.newErrorResult(t); } // case ResultConstants.SQL_ATTR_AUTO_IPD // - always true // default: throw - case never happens } return Result.updateZeroResult; } case ResultConstants.REQUESTDATA : { return sessionData.getDataResultSlice(cmd.getResultId(), cmd.getUpdateCount(), cmd.getFetchSize()); } case ResultConstants.DISCONNECT : { close(); return Result.updateZeroResult; } default : { return Result.newErrorResult( Error.runtimeError(ErrorCode.U_S0500, "Session")); } } }
比如使用下面方式执行：
String ddl0 = "DROP TABLE ADDRESSBOOK IF EXISTS; DROP TABLE ADDRESSBOOK_CATEGORY IF EXISTS; DROP TABLE USER IF EXISTS;"; String ddl1 = "CREATE TABLE USER(USER_ID INTEGER NOT NULL PRIMARY KEY,LOGIN_ID VARCHAR(128) NOT NULL,USER_NAME VARCHAR(254) DEFAULT ' ' NOT NULL,CREATE_DATE TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,UPDATE_DATE TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,LAST_ACCESS_DATE TIMESTAMP,CONSTRAINT IXUQ_LOGIN_ID0 UNIQUE(LOGIN_ID))"; String result1 = "1"; String result2 = "2"; stmnt.execute(ddl0); stmnt.execute(ddl1);则Result内容如下：
case ResultConstants.EXECDIRECT : result.updateCount = in.readInt(); result.fetchSize = in.readInt(); result.statementReturnType = in.readByte(); result.mainString = in.readString(); result.rsProperties = in.readByte(); result.queryTimeout = in.readShort(); result.generateKeys = in.readByte(); if (result.generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_NAMES || result .generateKeys == ResultConstants .RETURN_GENERATED_KEYS_COL_INDEXES) { result.generatedMetaData = new ResultMetaData(in); } break; mainString就是sql语句，然后就是执行过程了
sesseion的execute方法执行片段为
case ResultConstants.EXECDIRECT : { Result result = executeDirectStatement(cmd); result = performPostExecute(cmd, result); return result; } 来到executeDirectStatement了， public Result executeDirectStatement(Result cmd) { String sql = cmd.getMainString(); HsqlArrayList list; int maxRows = cmd.getUpdateCount(); if (maxRows == -1) { sessionContext.currentMaxRows = 0; } else if (sessionMaxRows == 0) { sessionContext.currentMaxRows = maxRows; } else { sessionContext.currentMaxRows = sessionMaxRows; sessionMaxRows = 0; } try { list = parser.compileStatements(sql, cmd); } catch (Throwable e) { return Result.newErrorResult(e); } Result result = null; for (int i = 0; i < list.size(); i++) { Statement cs = (Statement) list.get(i); cs.setGeneratedColumnInfo(cmd.getGeneratedResultType(), cmd.getGeneratedResultMetaData()); result = executeCompiledStatement(cs, ValuePool.emptyObjectArray); if (result.mode == ResultConstants.ERROR) { break; } } return result; } 终于看到parser.compileStatements(sql, cmd) 解析sql的语法分析过程了，parser是在new session中初始化的，
ParserCommand类的compileStatements方法
HsqlArrayList compileStatements(String sql, Result cmd) { HsqlArrayList list = new HsqlArrayList(); Statement cs = null; reset(sql); while (true) { if (token.tokenType == Tokens.X_ENDPARSE) { break; } try { lastError = null; cs = compilePart(cmd.getExecuteProperties()); } catch (HsqlException e) { if (lastError != null && lastError.getLevel() > e.getLevel()) { throw lastError; } throw e; } if (!cs.isExplain && cs.getParametersMetaData().getColumnCount() > 0) { throw Error.error(ErrorCode.X_42575); } cs.setCompileTimestamp( database.txManager.getGlobalChangeTimestamp()); list.add(cs); } int returnType = cmd.getStatementType(); if (returnType != StatementTypes.RETURN_ANY) { int group = cs.getGroup(); if (group == StatementTypes.X_SQL_DATA) { if (returnType == StatementTypes.RETURN_COUNT) { throw Error.error(ErrorCode.X_07503); } } else if (returnType == StatementTypes.RETURN_RESULT) { throw Error.error(ErrorCode.X_07504); } } return list; } 好吧，compilePart里面支持的所有的DDL数据库操作语言 private Statement compilePart(int props) { Statement cs; compileContext.reset(); setParsePosition(getPosition()); if (token.tokenType == Tokens.X_STARTPARSE) { read(); } switch (token.tokenType) { // DQL case Tokens.WITH : case Tokens.OPENBRACKET : case Tokens.SELECT : case Tokens.TABLE : { cs = compileCursorSpecification(RangeGroup.emptyArray, props, false); break; } case Tokens.VALUES : { RangeGroup[] ranges = new RangeGroup[]{ new RangeGroupSimple( session.sessionContext.sessionVariablesRange) }; compileContext.setOuterRanges(ranges); cs = compileShortCursorSpecification(props); break; } // DML case Tokens.INSERT : { cs = compileInsertStatement(RangeGroup.emptyArray); break; } case Tokens.UPDATE : { cs = compileUpdateStatement(RangeGroup.emptyArray); break; } case Tokens.MERGE : { cs = compileMergeStatement(RangeGroup.emptyArray); break; } case Tokens.DELETE : { cs = compileDeleteStatement(RangeGroup.emptyArray); break; } case Tokens.TRUNCATE : { cs = compileTruncateStatement(); break; } // PROCEDURE case Tokens.CALL : { cs = compileCallStatement( new RangeGroup[]{ new RangeGroupSimple( session.sessionContext .sessionVariablesRange) }, false); break; } // SQL SESSION case Tokens.SET : cs = compileSet(); break; // diagnostic case Tokens.GET : cs = compileGetStatement( session.sessionContext.sessionVariablesRange); break; case Tokens.START : cs = compileStartTransaction(); break; case Tokens.COMMIT : cs = compileCommit(); break; case Tokens.ROLLBACK : cs = compileRollback(); break; case Tokens.SAVEPOINT : cs = compileSavepoint(); break; case Tokens.RELEASE : cs = compileReleaseSavepoint(); break; // DDL case Tokens.CREATE : cs = compileCreate(); break; case Tokens.ALTER : cs = compileAlter(); break; case Tokens.DROP : cs = compileDrop(); break; case Tokens.GRANT : case Tokens.REVOKE : cs = compileGrantOrRevoke(); break; case Tokens.COMMENT : cs = compileComment(); break; // HSQL SESSION case Tokens.LOCK : cs = compileLock(); break; case Tokens.CONNECT : cs = compileConnect(); break; case Tokens.DISCONNECT : cs = compileDisconnect(); break; // HSQL COMMAND case Tokens.SCRIPT : cs = compileScript(); break; case Tokens.SHUTDOWN : cs = compileShutdown(); break; case Tokens.BACKUP : cs = compileBackup(); break; case Tokens.CHECKPOINT : cs = compileCheckpoint(); break; case Tokens.EXPLAIN : { int position = getPosition(); cs = compileExplainPlan(); cs.setSQL(getLastPart(position)); break; } case Tokens.DECLARE : cs = compileDeclare(); break; default : throw unexpectedToken(); } // SET_SESSION_AUTHORIZATION is translated dynamically at runtime for logging switch (cs.type) { // these are set at compile time for logging case StatementTypes.COMMIT_WORK : case StatementTypes.ROLLBACK_WORK : case StatementTypes.SET_USER_PASSWORD : case StatementTypes.EXPLAIN_PLAN : break; default : cs.setSQL(getLastPart()); } if (token.tokenType == Tokens.SEMICOLON) { read(); } else if (token.tokenType == Tokens.X_ENDPARSE) {} return cs; } 我们的语句是create的，我们看看compileCreate方法
StatementSchema compileCreate() { int tableType = TableBase.MEMORY_TABLE; boolean isTable = false; boolean isOrReplace = false; read(); switch (token.tokenType) { case Tokens.GLOBAL : read(); readThis(Tokens.TEMPORARY); readIfThis(Tokens.MEMORY); readThis(Tokens.TABLE); isTable = true; tableType = TableBase.TEMP_TABLE; break; case Tokens.TEMP : read(); readThis(Tokens.TABLE); isTable = true; tableType = TableBase.TEMP_TABLE; break; case Tokens.TEMPORARY : read(); readThis(Tokens.TABLE); isTable = true; tableType = TableBase.TEMP_TABLE; break; case Tokens.MEMORY : read(); readThis(Tokens.TABLE); isTable = true; break; case Tokens.CACHED : read(); readThis(Tokens.TABLE); isTable = true; tableType = TableBase.CACHED_TABLE; break; case Tokens.TEXT : read(); readThis(Tokens.TABLE); isTable = true; tableType = TableBase.TEXT_TABLE; break; case Tokens.TABLE : read(); isTable = true; tableType = database.schemaManager.getDefaultTableType(); break; case Tokens.OR : if (database.sqlSyntaxOra) { read(); readThis(Tokens.REPLACE); switch (token.tokenType) { case Tokens.FUNCTION : case Tokens.PROCEDURE : case Tokens.TRIGGER : case Tokens.TYPE : case Tokens.VIEW : break; default : throw unexpectedToken(Tokens.T_OR); } isOrReplace = true; } default : } if (isTable) { return compileCreateTable(tableType); } switch (token.tokenType) { // other objects case Tokens.ALIAS : return compileCreateAlias(); case Tokens.SEQUENCE : return compileCreateSequence(); case Tokens.SCHEMA : return compileCreateSchema(); case Tokens.TRIGGER : return compileCreateTrigger(isOrReplace); case Tokens.USER : return compileCreateUser(); case Tokens.ROLE : return compileCreateRole(); case Tokens.VIEW : return compileCreateView(false, isOrReplace); case Tokens.DOMAIN : return compileCreateDomain(); case Tokens.TYPE : return compileCreateType(isOrReplace); case Tokens.CHARACTER : return compileCreateCharacterSet(); case Tokens.COLLATION : return compileCreateCollation(); // index case Tokens.UNIQUE : read(); checkIsThis(Tokens.INDEX); return compileCreateIndex(true); case Tokens.INDEX : return compileCreateIndex(false); case Tokens.AGGREGATE : case Tokens.FUNCTION : case Tokens.PROCEDURE : return compileCreateProcedureOrFunction(isOrReplace); default : { throw unexpectedToken(); } } } StatementSchema compileCreateTable(int type) { boolean ifNot = false; if (token.tokenType == Tokens.IF) { int position = getPosition(); read(); if (token.tokenType == Tokens.NOT) { read(); readThis(Tokens.EXISTS); ifNot = true; } else { rewind(position); } } HsqlName name = readNewSchemaObjectName(SchemaObject.TABLE, false); name.setSchemaIfNull(session.getCurrentSchemaHsqlName()); Table table; switch (type) { case TableBase.TEMP_TEXT_TABLE : case TableBase.TEXT_TABLE : { table = new TextTable(database, name, type); break; } default : { table = new Table(database, name, type); } } return compileCreateTableBody(table, ifNot); } 我们进入了Table类，database数据库对象，name表名，type表类型compileCreateTableBody编译表字段内容，
StatementSchema compileCreateTableBody(Table table, boolean ifNot) { HsqlArrayList tempConstraints = new HsqlArrayList(); if (token.tokenType == Tokens.AS) { return readTableAsSubqueryDefinition(table); } int position = getPosition(); readThis(Tokens.OPENBRACKET); { Constraint c = new Constraint(null, null, SchemaObject.ConstraintTypes.TEMP); tempConstraints.add(c); } boolean start = true; boolean startPart = true; boolean end = false; while (!end) { switch (token.tokenType) { case Tokens.LIKE : { ColumnSchema[] likeColumns = readLikeTable(table); for (int i = 0; i < likeColumns.length; i++) { table.addColumn(likeColumns[i]); } start = false; startPart = false; break; } case Tokens.CONSTRAINT : case Tokens.PRIMARY : case Tokens.FOREIGN : case Tokens.UNIQUE : case Tokens.CHECK : if (!startPart) { throw unexpectedToken(); } readConstraint(table, tempConstraints); start = false; startPart = false; break; case Tokens.COMMA : if (startPart) { throw unexpectedToken(); } read(); startPart = true; break; case Tokens.CLOSEBRACKET : read(); end = true; break; default : if (!startPart) { throw unexpectedToken(); } checkIsSchemaObjectName(); HsqlName hsqlName = database.nameManager.newColumnHsqlName(table.getName(), token.tokenString, isDelimitedIdentifier()); read(); ColumnSchema newcolumn = readColumnDefinitionOrNull(table, hsqlName, tempConstraints); if (newcolumn == null) { if (start) { rewind(position); return readTableAsSubqueryDefinition(table); } else { throw Error.error(ErrorCode.X_42000); } } table.addColumn(newcolumn); start = false; startPart = false; } } if (token.tokenType == Tokens.ON) { if (!table.isTemp()) { throw unexpectedToken(); } read(); readThis(Tokens.COMMIT); if (token.tokenType == Tokens.DELETE) {} else if (token.tokenType == Tokens.PRESERVE) { table.persistenceScope = TableBase.SCOPE_SESSION; } read(); readThis(Tokens.ROWS); } OrderedHashSet names = new OrderedHashSet(); names.add(database.getCatalogName()); for (int i = 0; i < tempConstraints.size(); i++) { Constraint c = (Constraint) tempConstraints.get(i); HsqlName name = c.getMainTableName(); if (name != null) { Table t = database.schemaManager.findUserTable(null, name.name, name.schema.name); if (t != null && !t.isTemp()) { names.add(table.getName()); } } } String sql = getLastPart(); Object[] args = new Object[] { table, tempConstraints, null, Boolean.valueOf(ifNot) }; HsqlName[] writeLockNames = new HsqlName[names.size()]; names.toArray(writeLockNames); return new StatementSchema(sql, StatementTypes.CREATE_TABLE, args, null, writeLockNames); } hsqlName是每个字段的名称，ColumnSchema是每个字段的类型还有约束条件， 我们看看login_id字段的ColumnSchema，这里isnullable还是为1是有问题的，这是hsqldb的一个bug，不过说新版本已经修复了，http://sourceforge.net/projects/hsqldb/forums/forum/73674/topic/4647209 。
if (!isGenerated && !isIdentity) { if (database.sqlSyntaxMys) { switch (token.tokenType) { case Tokens.NULL : read(); break; case Tokens.NOT : read(); readThis(Tokens.NULL); isNullable = false; break; default : } } 这个参数是各个数据库的语法有特殊处理，可以参考http://hsqldb.org/doc/2.0/guide/dbproperties-chapt.html。但是这里没有提到对于not null的处理。 约束的话有下面这几种 case Tokens.CONSTRAINT : case Tokens.PRIMARY : case Tokens.FOREIGN : case Tokens.UNIQUE : case Tokens.CHECK : if (!startPart) { throw unexpectedToken(); } readConstraint(table, tempConstraints); start = false; startPart = false; break; Constraint对象包含 哪一列上的哪种约束类型Constraint c = new Constraint(constName, set, SchemaObject.ConstraintTypes.UNIQUE);当然还有
if (token.tokenType == Tokens.ON) { if (!table.isTemp()) { throw unexpectedToken(); } read(); readThis(Tokens.COMMIT); if (token.tokenType == Tokens.DELETE) {} else if (token.tokenType == Tokens.PRESERVE) { table.persistenceScope = TableBase.SCOPE_SESSION; } read(); readThis(Tokens.ROWS); } 接下来就调试执行DDL的查询过程了 result = executeCompiledStatement(cs, ValuePool.emptyObjectArray);