Chinaunix首页 | 论坛 | 博客

代码路上

首页　| 　博文目录　| 　关于我

hibean2014

  • 博客访问： 31026
  • 博文数量： 6
  • 博客积分： 0
  • 博客等级： 民兵
  • 技术积分： 91
  • 用 户 组： 普通用户
  • 注册时间： 2014-10-19 20:11
文章分类

全部博文（6）

文章存档

2016年（1）

2014年（5）

我的朋友
最近访客
推荐博文
相关博文
HSQLDB源码阅读 （三）命令的处理

分类： Java

2014-10-21 21:04:05

我们先说说上篇结尾的思考题，如果各个命令/请求参数信息截然不同的话，那都用一个Result来封装，我觉得是不太妥当的。如果差异这么大，我觉得还不如用不同的类来表示不同的命令/请求，就像在web中，有不同的HttpServletRequest的实现。这HSQLDB里，请求响应都是用Result表示，而在web里则分别用HttpServletRequest、HttpServletResponse表示。

这篇我们将继续深入阅读源码，了解命令的处理过程。

点击(此处)折叠或打开
  1. while (keepAlive) {
  2.                     msgType = dataInput.readByte();

  4.                     if (msgType < ResultConstants.MODE_UPPER_LIMIT) {
  5.                         receiveResult(msgType);
  6.                     } else {
  7.                         receiveOdbcPacket((char) msgType);
  8.                     }
  9.                 }
这是在ServerConnection.run方法中一个片段，这里是命令处理的主要地方。我们这里仍然只分析HSQL_STREAM_PROTOCOL的处理过程。
进入到receiveResult方法中一探究竟：

点击(此处)折叠或打开

  1. private void receiveResult(int resultMode) throws CleanExit, IOException {

  3.         boolean terminate = false;
  4.         Result resultIn = Result.newResult(session, resultMode, dataInput,
  5.                                            rowIn);

  7.         resultIn.readLobResults(session, dataInput, rowIn);
  8.         server.printRequest(mThread, resultIn);

  10.         Result resultOut = null;

  12.         switch (resultIn.getType()) {

  14.             case ResultConstants.CONNECT : {
  15.                 resultOut = setDatabase(resultIn);

  17.                 break;
  18.             }
  19.             case ResultConstants.DISCONNECT : {
  20.                 resultOut = Result.updateZeroResult;
  21.                 terminate = true;

  23.                 break;
  24.             }
  25.             case ResultConstants.RESETSESSION : {
  26.                 session.resetSession();

  28.                 resultOut = Result.updateZeroResult;

  30.                 break;
  31.             }
  32.             case ResultConstants.EXECUTE_INVALID : {
  33.                 resultOut =
  34.                     Result.newErrorResult(Error.error(ErrorCode.X_07502));

  36.                 break;
  37.             }
  38.             default : {
  39.                 resultOut = session.execute(resultIn);

  41.                 break;
  42.             }
  43.         }

  45.         resultOut.write(session, dataOutput, rowOut);
  46.         rowOut.reset(mainBuffer);
  47.         rowIn.resetRow(mainBuffer.length);

  49.         if (terminate) {
  50.             throw cleanExit;
  51.         }
  52.     }
我们看其它类型的处理都比较简单，我们要关注就是这个default的处理，继续进入到session.execute方法中。

点击(此处)折叠或打开

  1. public synchronized Result execute(Result cmd) {

  3.         if (isClosed) {
  4.             return Result.newErrorResult(Error.error(ErrorCode.X_08503));
  5.         }

  7.         sessionContext.currentMaxRows = 0;
  8.         isBatch = false;

  10.         JavaSystem.gc();

  12.         switch (cmd.mode) {

  14.             case ResultConstants.LARGE_OBJECT_OP : {
  15.                 return performLOBOperation((ResultLob) cmd);
  16.             }
  17.             case ResultConstants.EXECUTE : {
  18.                 int maxRows = cmd.getUpdateCount();

  20.                 if (maxRows == -1) {
  21.                     sessionContext.currentMaxRows = 0;
  22.                 } else {
  23.                     sessionContext.currentMaxRows = maxRows;
  24.                 }

  26.                 Statement cs = cmd.statement;

  28.                 if (cs == null
  29.                         || cs.compileTimestamp
  30.                            < database.schemaManager.schemaChangeTimestamp) {
  31.                     long csid = cmd.getStatementID();

  33.                     cs = statementManager.getStatement(this, csid);

  35.                     cmd.setStatement(cs);

  37.                     if (cs == null) {

  39.                         // invalid sql has been removed already
  40.                         return Result.newErrorResult(
  41.                             Error.error(ErrorCode.X_07502));
  42.                     }
  43.                 }

  45.                 Object[] pvals = (Object[]) cmd.valueData;
  46.                 Result result = executeCompiledStatement(cs, pvals,
  47.                     cmd.queryTimeout);

  49.                 result = performPostExecute(cmd, result);

  51.                 return result;
  52.             }
  53.             case ResultConstants.BATCHEXECUTE : {
  54.                 isBatch = true;

  56.                 Result result = executeCompiledBatchStatement(cmd);

  58.                 result = performPostExecute(cmd, result);

  60.                 return result;
  61.             }
  62.             case ResultConstants.EXECDIRECT : {
  63.                 Result result = executeDirectStatement(cmd);

  65.                 result = performPostExecute(cmd, result);

  67.                 return result;
  68.             }
  69.             case ResultConstants.BATCHEXECDIRECT : {
  70.                 isBatch = true;

  72.                 Result result = executeDirectBatchStatement(cmd);

  74.                 result = performPostExecute(cmd, result);

  76.                 return result;
  77.             }
  78.             case ResultConstants.PREPARE : {
  79.                 Statement cs;

  81.                 try {
  82.                     cs = statementManager.compile(this, cmd);
  83.                 } catch (Throwable t) {
  84.                     String errorString = cmd.getMainString();

  86.                     if (database.getProperties().getErrorLevel()
  87.                             == HsqlDatabaseProperties.NO_MESSAGE) {
  88.                         errorString = null;
  89.                     }

  91.                     return Result.newErrorResult(t, errorString);
  92.                 }

  94.                 Result result = Result.newPrepareResponse(cs);

  96.                 if (cs.getType() == StatementTypes.SELECT_CURSOR
  97.                         || cs.getType() == StatementTypes.CALL) {
  98.                     sessionData.setResultSetProperties(cmd, result);
  99.                 }

  101.                 result = performPostExecute(cmd, result);

  103.                 return result;
  104.             }
  105.             case ResultConstants.CLOSE_RESULT : {
  106.                 closeNavigator(cmd.getResultId());

  108.                 return Result.updateZeroResult;
  109.             }
  110.             case ResultConstants.UPDATE_RESULT : {
  111.                 Result result = this.executeResultUpdate(cmd);

  113.                 result = performPostExecute(cmd, result);

  115.                 return result;
  116.             }
  117.             case ResultConstants.FREESTMT : {
  118.                 statementManager.freeStatement(cmd.getStatementID());

  120.                 return Result.updateZeroResult;
  121.             }
  122.             case ResultConstants.GETSESSIONATTR : {
  123.                 int id = cmd.getStatementType();

  125.                 return getAttributesResult(id);
  126.             }
  127.             case ResultConstants.SETSESSIONATTR : {
  128.                 return setAttributes(cmd);
  129.             }
  130.             case ResultConstants.ENDTRAN : {
  131.                 switch (cmd.getActionType()) {

  133.                     case ResultConstants.TX_COMMIT :
  134.                         try {
  135.                             commit(false);
  136.                         } catch (Throwable t) {
  137.                             return Result.newErrorResult(t);
  138.                         }
  139.                         break;

  141.                     case ResultConstants.TX_COMMIT_AND_CHAIN :
  142.                         try {
  143.                             commit(true);
  144.                         } catch (Throwable t) {
  145.                             return Result.newErrorResult(t);
  146.                         }
  147.                         break;

  149.                     case ResultConstants.TX_ROLLBACK :
  150.                         rollback(false);
  151.                         break;

  153.                     case ResultConstants.TX_ROLLBACK_AND_CHAIN :
  154.                         rollback(true);
  155.                         break;

  157.                     case ResultConstants.TX_SAVEPOINT_NAME_RELEASE :
  158.                         try {
  159.                             String name = cmd.getMainString();

  161.                             releaseSavepoint(name);
  162.                         } catch (Throwable t) {
  163.                             return Result.newErrorResult(t);
  164.                         }
  165.                         break;

  167.                     case ResultConstants.TX_SAVEPOINT_NAME_ROLLBACK :
  168.                         try {
  169.                             rollbackToSavepoint(cmd.getMainString());
  170.                         } catch (Throwable t) {
  171.                             return Result.newErrorResult(t);
  172.                         }
  173.                         break;

  175.                     case ResultConstants.PREPARECOMMIT :
  176.                         try {
  177.                             prepareCommit();
  178.                         } catch (Throwable t) {
  179.                             return Result.newErrorResult(t);
  180.                         }
  181.                         break;
  182.                 }

  184.                 return Result.updateZeroResult;
  185.             }
  186.             case ResultConstants.SETCONNECTATTR : {
  187.                 switch (cmd.getConnectionAttrType()) {

  189.                     case ResultConstants.SQL_ATTR_SAVEPOINT_NAME :
  190.                         try {
  191.                             savepoint(cmd.getMainString());
  192.                         } catch (Throwable t) {
  193.                             return Result.newErrorResult(t);
  194.                         }

  196.                     // case ResultConstants.SQL_ATTR_AUTO_IPD
  197.                     // - always true
  198.                     // default: throw - case never happens
  199.                 }

  201.                 return Result.updateZeroResult;
  202.             }
  203.             case ResultConstants.REQUESTDATA : {
  204.                 return sessionData.getDataResultSlice(cmd.getResultId(),
  205.                                                       cmd.getUpdateCount(),
  206.                                                       cmd.getFetchSize());
  207.             }
  208.             case ResultConstants.DISCONNECT : {
  209.                 close();

  211.                 return Result.updateZeroResult;
  212.             }
  213.             default : {
  214.                 return Result.newErrorResult(
  215.                     Error.runtimeError(ErrorCode.U_S0500, "Session"));
  216.             }
  217.         }
  218.     }
很多种命令，看起来相当复杂。我们挑其中比较简单的

点击(此处)折叠或打开

  1. case ResultConstants.EXECDIRECT : {
  2.                 Result result = executeDirectStatement(cmd);

  4.                 result = performPostExecute(cmd, result);

  6.                 return result;
  7.             }
当Result的mode为EXECDIRECT时，表示直接执行sql语句。（与之相对应的，我们知道有一种prepare之后再执行sql的方式）
我们先了解其处理过程，再回头来看其它命令，就更容易理解了。
进入executeDirectStatement：

点击(此处)折叠或打开

  1. public Result executeDirectStatement(Result cmd) {

  3.         String sql = cmd.getMainString();
  4.         HsqlArrayList list;
  5.         int maxRows = cmd.getUpdateCount();

  7.         if (maxRows == -1) {
  8.             sessionContext.currentMaxRows = 0;
  9.         } else if (sessionMaxRows == 0) {
  10.             sessionContext.currentMaxRows = maxRows;
  11.         } else {
  12.             sessionContext.currentMaxRows = sessionMaxRows;
  13.             sessionMaxRows = 0;
  14.         }

  16.         try {
  17.             list = parser.compileStatements(sql, cmd);
  18.         } catch (Throwable e) {
  19.             return Result.newErrorResult(e);
  20.         }

  22.         Result result = null;
  23.         boolean recompile = false;
  24.         HsqlName originalSchema = getCurrentSchemaHsqlName();

  26.         for (int i = 0; i < list.size(); i++) {
  27.             Statement cs = (Statement) list.get(i);

  29.             if (i > 0) {
  30.                 if (cs.getCompileTimestamp()
  31.                         > database.txManager.getGlobalChangeTimestamp()) {
  32.                     recompile = true;
  33.                 }

  35.                 if (cs.getSchemaName() != null
  36.                         && cs.getSchemaName() != originalSchema) {
  37.                     recompile = true;
  38.                 }
  39.             }

  41.             if (recompile) {
  42.                 cs = compileStatement(cs.getSQL(), cmd.getExecuteProperties());
  43.             }

  45.             cs.setGeneratedColumnInfo(cmd.getGeneratedResultType(),
  46.                                       cmd.getGeneratedResultMetaData());

  48.             result = executeCompiledStatement(cs, ValuePool.emptyObjectArray,
  49.                                               cmd.queryTimeout);

  51.             if (result.mode == ResultConstants.ERROR) {
  52.                 break;
  53.             }
  54.         }

  56.         return result;
  57.     }
sql语句是字符串，所以需要解析其涵义，hsql先将sql语句进行编译：parser.compileStatements(sql, cmd);（我们下一章再去关注，这个sql语句的编译过程）。返回的是HsqlArrayList——存Statement的List，意味着可能一个命令中包含多个执行的sql语句，然后循环执行该list中的Statement对象——executeCompiledStatement。

  1. result = executeCompiledStatement(cs, ValuePool.emptyObjectArray,
  2.                                               cmd.queryTimeout);
再进入executeCompiledStatement

点击(此处)折叠或打开

  1. public Result executeCompiledStatement(Statement cs, Object[] pvals,
  2.                                            int timeout) {

  4.         Result r;

  6.         if (abortTransaction) {
  7.             rollbackNoCheck(false);

  9.             return Result.newErrorResult(Error.error(ErrorCode.X_40001));
  10.         }

  12.         if (sessionContext.depth > 0) {
  13.             if (sessionContext.noSQL.booleanValue()
  14.                     || cs.isAutoCommitStatement()) {
  15.                 return Result.newErrorResult(Error.error(ErrorCode.X_46000));
  16.             }
  17.         }

  19.         if (cs.isAutoCommitStatement()) {
  20.             if (isReadOnly()) {
  21.                 return Result.newErrorResult(Error.error(ErrorCode.X_25006));
  22.             }

  24.             try {

  26.                 /** special autocommit for backward compatibility */
  27.                 commit(false);
  28.             } catch (HsqlException e) {
  29.                 database.logger.logInfoEvent("Exception at commit");
  30.             }
  31.         }

  33.         sessionContext.currentStatement = cs;

  35.         boolean isTX = cs.isTransactionStatement();

  37.         if (!isTX) {
  38.             actionTimestamp =
  39.                 database.txManager.getNextGlobalChangeTimestamp();

  41.             sessionContext.setDynamicArguments(pvals);

  43.             // statements such as DISCONNECT may close the session
  44.             if (database.logger.getSqlEventLogLevel()
  45.                     >= SimpleLog.LOG_NORMAL) {
  46.                 database.logger.logStatementEvent(this, cs, pvals,
  47.                                                   Result.updateZeroResult,
  48.                                                   SimpleLog.LOG_NORMAL);
  49.             }

  51.             r = cs.execute(this);
  52.             sessionContext.currentStatement = null;

  54.             return r;
  55.         }

  57.         while (true) {
  58.             actionIndex = rowActionList.size();

  60.             database.txManager.beginAction(this, cs);

  62.             cs = sessionContext.currentStatement;

  64.             if (cs == null) {
  65.                 return Result.newErrorResult(Error.error(ErrorCode.X_07502));
  66.             }

  68.             if (abortTransaction) {
  69.                 rollbackNoCheck(false);

  71.                 sessionContext.currentStatement = null;

  73.                 return Result.newErrorResult(Error.error(ErrorCode.X_40001));
  74.             }

  76.             timeoutManager.startTimeout(timeout);

  78.             try {
  79.                 latch.await();
  80.             } catch (InterruptedException e) {
  81.                 abortTransaction = true;
  82.             }

  84.             boolean abortAction = timeoutManager.endTimeout();

  86.             if (abortAction) {
  87.                 r = Result.newErrorResult(Error.error(ErrorCode.X_40502));

  89.                 endAction(r);

  91.                 break;
  92.             }

  94.             if (abortTransaction) {
  95.                 rollbackNoCheck(false);

  97.                 sessionContext.currentStatement = null;

  99.                 return Result.newErrorResult(Error.error(ErrorCode.X_40001));
  100.             }

  102.             database.txManager.beginActionResume(this);

  104.             // tempActionHistory.add("sql execute " + cs.sql + " " + actionTimestamp + " " + rowActionList.size());
  105.             sessionContext.setDynamicArguments(pvals);

  107.             r = cs.execute(this);

  109.             if (database.logger.getSqlEventLogLevel()
  110.                     >= SimpleLog.LOG_NORMAL) {
  111.                 database.logger.logStatementEvent(this, cs, pvals, r,
  112.                                                   SimpleLog.LOG_NORMAL);
  113.             }

  115.             lockStatement = sessionContext.currentStatement;

  117.             // tempActionHistory.add("sql execute end " + actionTimestamp + " " + rowActionList.size());
  118.             endAction(r);

  120.             if (abortTransaction) {
  121.                 rollbackNoCheck(false);

  123.                 sessionContext.currentStatement = null;

  125.                 return Result.newErrorResult(Error.error(r.getException(),
  126.                         ErrorCode.X_40001, null));
  127.             }

  129.             if (redoAction) {
  130.                 redoAction = false;

  132.                 try {
  133.                     latch.await();
  134.                 } catch (InterruptedException e) {
  135.                     abortTransaction = true;
  136.                 }
  137.             } else {
  138.                 break;
  139.             }
  140.         }

  142.         if (sessionContext.depth == 0
  143.                 && (sessionContext.isAutoCommit.booleanValue()
  144.                     || cs.isAutoCommitStatement())) {
  145.             try {
  146.                 if (r.mode == ResultConstants.ERROR) {
  147.                     rollbackNoCheck(false);
  148.                 } else {
  149.                     commit(false);
  150.                 }
  151.             } catch (Exception e) {
  152.                 sessionContext.currentStatement = null;

  154.                 return Result.newErrorResult(Error.error(ErrorCode.X_40001,
  155.                         e));
  156.             }
  157.         }

  159.         sessionContext.currentStatement = null;

  161.         return r;
  162.     }
又是一个好长的函数~
我们忽略掉其中的错误检查部分，看函数的主体内容，其实就一个：

  1. r = cs.execute(this);
其它代码主要都在处理事务的东西。
而cs.execute则是一个抽象方法

点击(此处)折叠或打开

  1. public abstract Result execute(Session session);
它的具体代码，由Statement的子类来实现。

代码分析到现在，我们来理清一下命令的处理过程：
1. 大多数命令，都会进入Session.execute(Result)进行处理
2. 在Session.execute中，根据命令的类型，进行分别的处理
3. 对于Execute的命令（如DirectExecute)，需要Statement对象来执行
4. Statement由parser编译而来——编译器根据语句内容，生成了不同的Statement子类的实例
5. 执行Statement.execute(Session)方法，返回Result对象

理清命令处理的思路之后，再回头来看Session.execute(Result cmd)的其它命令的处理，过程就变得清晰一点了。
其中BATCHEXECDIRECT和EXECDIRECT是类似的，比较让人困惑的是EXECUTE

点击(此处)折叠或打开

  1. case ResultConstants.EXECUTE : {
  2.                 int maxRows = cmd.getUpdateCount();

  4.                 if (maxRows == -1) {
  5.                     sessionContext.currentMaxRows = 0;
  6.                 } else {
  7.                     sessionContext.currentMaxRows = maxRows;
  8.                 }

  10.                 Statement cs = cmd.statement;

  12.                 if (cs == null
  13.                         || cs.compileTimestamp
  14.                            < database.schemaManager.schemaChangeTimestamp) {
  15.                     long csid = cmd.getStatementID();

  17.                     cs = statementManager.getStatement(this, csid);

  19.                     cmd.setStatement(cs);

  21.                     if (cs == null) {

  23.                         // invalid sql has been removed already
  24.                         return Result.newErrorResult(
  25.                             Error.error(ErrorCode.X_07502));
  26.                     }
  27.                 }

  29.                 Object[] pvals = (Object[]) cmd.valueData;
  30.                 Result result = executeCompiledStatement(cs, pvals,
  31.                     cmd.queryTimeout);

  33.                 result = performPostExecute(cmd, result);

  35.                 return result;
  36.             }
我在之前看这个代码的时候也很困惑，它的Statement居然是根据getStatementID取出来的。
而且在之前通信协议的解析中[Result.newResult]

点击(此处)折叠或打开

  1. case ResultConstants.EXECUTE :
  2.                 result.updateCount = in.readInt();
  3.                 result.fetchSize = in.readInt();
  4.                 result.statementID = in.readLong();
  5.                 result.rsProperties = in.readByte();
  6.                 result.queryTimeout = in.readShort();

  8.                 Statement statement =
  9.                     session.statementManager.getStatement(session,
  10.                         result.statementID);

  12.                 if (statement == null) {

  14.                     // invalid statement
  15.                     result.mode = ResultConstants.EXECUTE_INVALID;
  16.                     result.valueData = ValuePool.emptyObjectArray;

  18.                     break;
  19.                 }

  21.                 result.statement = statement;
  22.                 result.metaData = result.statement.getParametersMetaData();
  23.                 result.valueData = readSimple(in, result.metaData);
  24.                 break;
这里也是通过statementManager.getStatement取出来的。那这个Statement是什么时候存进去的呢？
秘密就在于Prepare

点击(此处)折叠或打开

  1. case ResultConstants.PREPARE : {
  2.                 Statement cs;

  4.                 try {
  5.                     cs = statementManager.compile(this, cmd);
  6.                 } catch (Throwable t) {
  7.                     String errorString = cmd.getMainString();

  9.                     if (database.getProperties().getErrorLevel()
  10.                             == HsqlDatabaseProperties.NO_MESSAGE) {
  11.                         errorString = null;
  12.                     }

  14.                     return Result.newErrorResult(t, errorString);
  15.                 }

  17.                 Result result = Result.newPrepareResponse(cs);

  19.                 if (cs.getType() == StatementTypes.SELECT_CURSOR
  20.                         || cs.getType() == StatementTypes.CALL) {
  21.                     sessionData.setResultSetProperties(cmd, result);
  22.                 }

  24.                 result = performPostExecute(cmd, result);

  26.                 return result;
  27.             }
也就是在这里，hsql将sql进行了预编译，然后生成一个statementId返回给客户端，客户端在下次请求执行的时候，必须将这个statementId带上，然后hsql从statementManager中取出之前编译好的sql语句（即statement对象）。

所以，我们知道，为什么对于需要多次执行的sql语句，为什么prepare的性能要高一些，因为executeDirect的语句需要每次编译，而prepare的sql语句只需要编译一次。

这里我们就只解释这些命令了，其它命令相对比较简单，就不逐一进行说明了，有兴趣的自己查看源码。

好，本文就到这，下回我们分析sql语句的编译过程。

[相关源码文件]
Result.java
Session.java
StatementManager.java











阅读(2491) | 评论(0) | 转发(0) |
0

上一篇：HSQLDB源码阅读 （二）通信协议的处理

下一篇：HSQLDB源码阅读 （四）SQL编译（上）

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6