sqlqry.pas

仿sql查询分析器

      if (SSErrInfo.lNative = 0) or (SSErrInfo.lNative = 5701) then
      begin // 应该是PRINT 返回的消息, 或更改数据库信息
        if SSErrInfo.lNative = 0 then
          ErrMsg := String(SSErrInfo.pwszMessage)
        else
          ErrMsg := '';
        InfoMsg := True;
      end
      else
      begin
        ErrMsg := Format('服务器: 消息 %d, 级别 %d, 状态 %d',
                      [SSErrInfo.lNative, SSErrInfo.bClass, SSErrInfo.bState]);
        ProcName := String(SSErrInfo.pwszProcedure);
        if ProcName <> '' then
          ErrMsg := ErrMsg + Format(', 过程 %s', [ProcName]);
        ErrMsg := ErrMsg + Format(', 行 %d', [SSErrInfo.wLineNumber]);
        ErrMsg := ErrMsg + #13#10 + String(SSErrInfo.pwszMessage);

        // 11 可能是连接中断问题, 但未见正式的文档说明.
        if SSErrInfo.lNative = 11 then ConnClosed := True;
      end;

      Result := True;
      ActiveX.CoTaskMemFree(SSErrInfo);
      ActiveX.CoTaskMemFree(pStrBuf);
    end;
  end;

  function GetErrInfo(ErrInfo: IErrorInfo): string;
  var
    Msg: WideString;
  begin
    if ErrInfo.GetDescription(Msg) = S_OK then
      Result := Format('%s', [string(Msg)])
    else
      Result := '';
  end;
begin
  InfoMsg := False;
  ConnClosed := False;
  Result := 0;

  GetErrorInfo(0, OleErr);
  if OleErr <> nil then
  begin
    if Supports(OleErr, IErrorRecords, ErrRecords) then
    begin
      hr := ErrRecords.GetRecordCount(RecCnt);
      if hr <> S_OK then Exit;

      for I := 0 to RecCnt - 1 do
      begin
        if ErrRecords.GetCustomErrorObject(I, ISQLServerErrorInfo, IUnknown(SSqlErr)) = S_OK then
        begin
          if SSqlErr = nil then Continue;
          if not GetSSqlErrInfo(SSqlErr, ErrorMsg) then
          begin
            ErrorMsg := GetErrInfo(OleErr);
          end;
        end
        else
        begin
          ErrorMsg := GetErrInfo(OleErr);
        end;
      end;
    end
    else
    begin
      ErrorMsg := GetErrInfo(OleErr);
    end;

    if InfoMsg then
      Result := 2 else
    if ConnClosed then
      Result := 3 else
      Result := 1;
  end;
end;

procedure InitConnection(Conn: _Connection);
var
  RecsAffected: OleVariant;
begin
  Conn.Execute('set showplan_text off', RecsAffected, -1);
  Conn.Execute('SET NOEXEC OFF SET PARSEONLY OFF', RecsAffected, -1);
  Conn.Execute('set showplan_all off', RecsAffected, -1);
end;

function IsConnectionActive(Conn: _Connection): Boolean;
begin
  Result := (Conn.State and adStateOpen) = adStateOpen;
end;

type
  TExecuteThread = class(TThread)
  private
    FConnection: _Connection;
    FMain: TSqlQryFrm;
    FSQL: string;
    FParseOnly: Boolean;

    FCommand: CommandStd;
    FCanceled: Boolean;

    FInfoMsg: Boolean;  // 当前是否有消息性错误.

    FConnClosed: Boolean; // 是否遇到连接错误.
    FHasErrors: Boolean;
    FTimeCost: TList;
    FMsgList: TStringList;
    FDataList: TList;

    FLock: TRTLCriticalSection;

    procedure AddMsg(AMsg: string);
    procedure AddRecordset(Rst: _Recordset);
    procedure ProcessResults;
  public
    constructor Create;
    destructor Destroy; override;
    procedure Execute; override;
    procedure Cancel;
    procedure Lock;
    procedure Unlock;
  end;

{ TExecuteThread }

procedure TExecuteThread.AddMsg(AMsg: string);
begin
  FMsgList.Add(AMsg);
end;

procedure TExecuteThread.AddRecordset(Rst: _Recordset);
var
  RowData: TRowDataList;
  bNormal: Boolean;
begin
  RowData := TRowDataList.Create;
  try
    bNormal := RowData.ImportRecordset(Rst, FCanceled);
    if not bNormal and (RowData.Data.Count = 0) then
      FreeAndNil(RowData)
    else
      FDataList.Add(Pointer(RowData));

    if (RowData <> nil) and bNormal then
      AddMsg(Format(#13#10'（所影响的行数为 %d 行）'#13#10, [RowData.Data.Count]));
  except
    on E: Exception do
    begin
      RowData.Free;
      AddMsg(E.Message);
    end;
  end;
end;

procedure TExecuteThread.Cancel;
begin
  Lock;
  try
    FCanceled := True;
    if FCommand <> nil then FCommand.Cancel;
  finally
    Unlock;
  end;
end;

constructor TExecuteThread.Create;
begin
  inherited Create(True); // Create suspended

  InitializeCriticalSection(FLock);
  FTimeCost := TList.Create;
  FDataList := TList.Create;
  FMsgList := TStringList.Create;
end;

destructor TExecuteThread.Destroy;
begin
  FTimeCost.Free;
  FDataList.Free;
  FMsgList.Free;

  Lock;
  Unlock;
  DeleteCriticalSection(FLock);

  inherited;
end;

procedure TExecuteThread.Execute;
var
  Rst, Rst2: _Recordset;
  I, OldTick, NewTick, Affected: Integer;
  VarRecsAffected, Params: OleVariant;
  sqls: TStringDynArray;
  hr: HResult;
begin
  if FSQL = '' then Exit;

  CoInitialize(Nil);
  try

    SetLength(sqls, 0);
    sqls := SplitSQL(FSQL);

    FMain.FExecuting := True;

    // 如果连接已经关闭, 重新建立.
    if not IsConnectionActive(FConnection) then
    begin
      (FConnection as ConnectionStd).Open('', '', '', -1);
      ProcessResults;
    end;

    // 连接无法建立. 退出.
    if not IsConnectionActive(FConnection) then Exit;

    // 建立Command 对象
    FCommand := CoCommand.Create as CommandStd;
    FCommand.Set_ActiveConnection(FConnection);

    if FParseOnly then
    begin
      hr := (FConnection as ConnectionStd).Execute('set parseonly on',
                  VarRecsAffected, adExecuteNoRecords, Rst2);
      if hr <> S_OK then Exit;
    end;

    for I := 0 to Length(sqls) - 1 do
    begin
      if FCanceled or FConnClosed then Break;

      OldTick := GetTickCount;
      try
        FCommand.Set_CommandText(sqls[I]);
        Params := EmptyParam;
        hr := FCommand.Execute(VarRecsAffected, Params, 0, Rst);

        NewTick := GetTickCount;
        FTimeCost.Add(Pointer(NewTick - OldTick));

        if hr <> S_OK then
        begin
          if LoWord(hr) <> adErrOperationCancelled then FHasErrors := True;
          ProcessResults;
          Continue;
        end;
      except
        On E: Exception do begin
          NewTick := GetTickCount;
          FTimeCost.Add(Pointer(NewTick - OldTick));
          AddMsg(E.Message);
          FHasErrors := True;
          Continue;
        end;
      end;

      while (Rst <> nil) and (not FCanceled) and (not FConnClosed) do
      begin
        ProcessResults;

        Affected := VarRecsAffected;
        if (Rst.State and adStateOpen) = adStateOpen then
        begin
          AddRecordset(Rst);
        end
        else if not FInfoMsg and (Affected >= 0) then
        begin
          AddMsg(Format(#13#10'（所影响的行数为 %d 行）'#13#10, [Affected]));
        end;

        if FCanceled then Break;

        hr := RecordsetStd(Rst).NextRecordset(VarRecsAffected, Rst2);
        if hr <> S_OK then
          if LoWord(hr) <> adErrOperationCancelled then FHasErrors := True;
        Rst := Rst2;
        Rst2 := nil;
      end;
    end;

    // 有时候释放Recordset要花很长时间, 特别SELECT带有image类型的字段
    Rst := nil;

    Lock;
    try
      FCommand := nil;
    finally
      Unlock;
    end;

    if FParseOnly then
    begin
      hr := (FConnection as ConnectionStd).Execute('set parseonly off',
                    VarRecsAffected, adExecuteNoRecords, Rst2);
      if hr <> S_OK then Exit;
    end;

    if FCanceled then
    begin
      AddMsg('用户已取消查询');
    end;

    // 清除TSqlQryFrm中的线程变量.
    Self.Synchronize(FMain.ClearVarRef);

  finally
    CoUninitialize;
  end;
end;

procedure TExecuteThread.Lock;
begin
  EnterCriticalSection(FLock);
end;

procedure TExecuteThread.ProcessResults;
var
  err_type: Integer;
  ErrMsg: string;
begin
  err_type := ProcessResult(ErrMsg);
  if ErrMsg <> '' then AddMsg(ErrMsg);
  FInfoMsg := err_type = 2;
  FConnClosed := err_type = 3;
end;

procedure TExecuteThread.Unlock;
begin
  LeaveCriticalSection(FLock);
end;

{ TSqlQryFrm }

procedure TSqlQryFrm.CancelExecute;
begin
  FCanceling := True;
  if FExecThread <> nil then
  begin
    TExecuteThread(FExecThread).Cancel;
    SBar.SimpleText := '正在取消批查询, 请等待...';
  end;
end;

procedure TSqlQryFrm.ClearResults;
var
  I: Integer;
  Ctrl: TControl;
begin
  Pgc1.ActivePageIndex := 0;
  for I := PnlResult.ControlCount - 1 downto 0 do
  begin
    Ctrl := PnlResult.Controls[I];
    Ctrl.Free;
  end;
  PnlResult.Align := alTop;
  PnlResult.Height := SbxResult.ClientHeight;

  MoMsg.Clear;
end;

procedure TSqlQryFrm.Execute;
var
  sql: string;
begin
  if Self.Executing or (MoSql.Text = '') then Exit;

  if MoSql.SelLength = 0 then
    sql := MoSql.Text
  else
    sql := MoSql.SelText;

  // 清除结果.
  ClearResults;

  FCanceling := False;

  if FConnClosed and IsConnectionActive(FConnection) then
  begin
    // 即使是执行过程中, 遇到通讯问题, State仍然不会变成adStateClosed
    // 所以必须重新关闭, 再建立联接
    FConnection.Close;
    FConnClosed := False;
  end;

  SBar.SimpleText := '正在执行批查询...';

  FExecThread := TExecuteThread.Create;
  FExecThread.FreeOnTerminate := True;
  FExecThread.OnTerminate := OnThreadTerminate;
  TExecuteThread(FExecThread).FConnection := FConnection;
  TExecuteThread(FExecThread).FMain := Self;
  TExecuteThread(FExecThread).FSQL := sql;
  TExecuteThread(FExecThread).FParseOnly := FParseOnly;
  FExecThread.Resume;
end;

procedure TSqlQryFrm.ParseSQL;
begin
  if Self.Executing then Exit;

  FParseOnly := True;
  try Execute; except end;