freepascal.compiler/packages/fcl-db/src/sqldb/postgres/pqconnection.pp

unit pqconnection;

{$mode objfpc}{$H+}

{$Define LinkDynamically}

interface

uses
  Classes, SysUtils, sqldb, db, dbconst,bufdataset,
{$IfDef LinkDynamically}
  postgres3dyn;
{$Else}
  postgres3;
{$EndIf}

type
  TPQTrans = Class(TSQLHandle)
    protected
    PGConn        : PPGConn;
  end;

  TPQCursor = Class(TSQLCursor)
    protected
    Statement    : string;
    StmtName     : string;
    tr           : TPQTrans;
    res          : PPGresult;
    CurTuple     : integer;
    FieldBinding : array of integer;
  end;

  EPQDatabaseError = class(EDatabaseError)
    public
      SEVERITY:string;
      SQLSTATE: string;
      MESSAGE_PRIMARY:string;
      MESSAGE_DETAIL:string;
      MESSAGE_HINT:string;
      STATEMENT_POSITION:string;
  end;

  TPQTranConnection = class
  protected
    FPGConn        : PPGConn;
    FTranActive    : boolean
  end;

  { TPQConnection }

  TPQConnection = class (TSQLConnection)
  private
    FConnectionPool      : array of TPQTranConnection;
    FCursorCount         : word;
    FConnectString       : string;
    FIntegerDateTimes    : boolean;
    FVerboseErrors       : Boolean;
    procedure CheckConnectionStatus(var conn: PPGconn);
    procedure CheckResultError(var res: PPGresult; conn:PPGconn; ErrMsg: string);
    function TranslateFldType(res : PPGresult; Tuple : integer; out Size : integer) : TFieldType;
    procedure ExecuteDirectPG(const Query : String);
  protected
    procedure DoInternalConnect; override;
    procedure DoInternalDisconnect; override;
    function GetHandle : pointer; override;

    Function AllocateCursorHandle : TSQLCursor; override;
    Procedure DeAllocateCursorHandle(var cursor : TSQLCursor); override;
    Function AllocateTransactionHandle : TSQLHandle; override;

    procedure PrepareStatement(cursor: TSQLCursor;ATransaction : TSQLTransaction;buf : string; AParams : TParams); override;
    procedure Execute(cursor: TSQLCursor;atransaction:tSQLtransaction; AParams : TParams); override;
    procedure AddFieldDefs(cursor: TSQLCursor; FieldDefs : TfieldDefs); override;
    function Fetch(cursor : TSQLCursor) : boolean; override;
    procedure UnPrepareStatement(cursor : TSQLCursor); override;
    function LoadField(cursor : TSQLCursor;FieldDef : TfieldDef;buffer : pointer; out CreateBlob : boolean) : boolean; override;
    function GetTransactionHandle(trans : TSQLHandle): pointer; override;
    function RollBack(trans : TSQLHandle) : boolean; override;
    function Commit(trans : TSQLHandle) : boolean; override;
    procedure CommitRetaining(trans : TSQLHandle); override;
    function StartdbTransaction(trans : TSQLHandle; AParams : string) : boolean; override;
    procedure RollBackRetaining(trans : TSQLHandle); override;
    procedure UpdateIndexDefs(IndexDefs : TIndexDefs;TableName : string); override;
    function GetSchemaInfoSQL(SchemaType : TSchemaType; SchemaObjectName, SchemaPattern : string) : string; override;
    procedure LoadBlobIntoBuffer(FieldDef: TFieldDef;ABlobBuf: PBufBlobField; cursor: TSQLCursor;ATransaction : TSQLTransaction); override;
    function RowsAffected(cursor: TSQLCursor): TRowsCount; override;
  public
    constructor Create(AOwner : TComponent); override;
    function GetConnectionInfo(InfoType:TConnInfoType): string; override;
    procedure CreateDB; override;
    procedure DropDB; override;
  published
    property DatabaseName;
    property KeepConnection;
    property LoginPrompt;
    property Params;
    property OnLogin;
    Property VerboseErrors : Boolean Read FVerboseErrors Write FVerboseErrors default true;
  end;

  { TPQConnectionDef }

  TPQConnectionDef = Class(TConnectionDef)
    Class Function TypeName : String; override;
    Class Function ConnectionClass : TSQLConnectionClass; override;
    Class Function Description : String; override;
    Class Function DefaultLibraryName : String; override;
    Class Function LoadFunction : TLibraryLoadFunction; override;
    Class Function UnLoadFunction : TLibraryUnLoadFunction; override;
    Class Function LoadedLibraryName: string; override;
  end;

implementation

uses math, strutils, FmtBCD;

ResourceString
  SErrRollbackFailed = 'Rollback transaction failed';
  SErrCommitFailed = 'Commit transaction failed';
  SErrConnectionFailed = 'Connection to database failed';
  SErrTransactionFailed = 'Start of transacion failed';
  SErrClearSelection = 'Clear of selection failed';
  SErrExecuteFailed = 'Execution of query failed';
  SErrFieldDefsFailed = 'Can not extract field information from query';
  SErrFetchFailed = 'Fetch of data failed';
  SErrPrepareFailed = 'Preparation of query failed.';
  SErrUnPrepareFailed = 'Unpreparation of query failed.';

const Oid_Bool     = 16;
      Oid_Bytea    = 17;
      Oid_char     = 18;
      Oid_Text     = 25;
      Oid_Oid      = 26;
      Oid_Name     = 19;
      Oid_Int8     = 20;
      Oid_int2     = 21;
      Oid_Int4     = 23;
      Oid_Float4   = 700;
      Oid_Money    = 790;
      Oid_Float8   = 701;
      Oid_Unknown  = 705;
      Oid_MacAddr  = 829;
      Oid_Inet     = 869;
      Oid_bpchar   = 1042;
      Oid_varchar  = 1043;
      oid_date      = 1082;
      oid_time      = 1083;
      Oid_timeTZ    = 1266;
      Oid_timestamp = 1114;
      Oid_timestampTZ = 1184;
      Oid_interval  = 1186;
      oid_numeric   = 1700;
      Oid_uuid      = 2950;


constructor TPQConnection.Create(AOwner : TComponent);

begin
  inherited;
  FConnOptions := FConnOptions + [sqSupportParams] + [sqEscapeRepeat] + [sqEscapeSlash];
  FieldNameQuoteChars:=DoubleQuotes;
  VerboseErrors:=True;
end;

procedure TPQConnection.CreateDB;

begin
  ExecuteDirectPG('CREATE DATABASE ' +DatabaseName);
end;

procedure TPQConnection.DropDB;

begin
  ExecuteDirectPG('DROP DATABASE ' +DatabaseName);
end;

procedure TPQConnection.ExecuteDirectPG(const query : string);

var ASQLDatabaseHandle    : PPGConn;
    res                   : PPGresult;

begin
  CheckDisConnected;
{$IfDef LinkDynamically}
  InitialisePostgres3;
{$EndIf}

  FConnectString := '';
  if (UserName <> '') then FConnectString := FConnectString + ' user=''' + UserName + '''';
  if (Password <> '') then FConnectString := FConnectString + ' password=''' + Password + '''';
  if (HostName <> '') then FConnectString := FConnectString + ' host=''' + HostName + '''';
  FConnectString := FConnectString + ' dbname=''template1''';
  if (Params.Text <> '') then FConnectString := FConnectString + ' '+Params.Text;

  ASQLDatabaseHandle := PQconnectdb(pchar(FConnectString));

  CheckConnectionStatus(ASQLDatabaseHandle);

  res := PQexec(ASQLDatabaseHandle,pchar(query));

  CheckResultError(res,ASQLDatabaseHandle,SDBCreateDropFailed);

  PQclear(res);
  PQFinish(ASQLDatabaseHandle);
{$IfDef LinkDynamically}
  ReleasePostgres3;
{$EndIf}
end;


function TPQConnection.GetTransactionHandle(trans : TSQLHandle): pointer;
begin
  Result := trans;
end;

function TPQConnection.RollBack(trans : TSQLHandle) : boolean;
var
  res : PPGresult;
  tr  : TPQTrans;
  i   : Integer;
begin
  result := false;

  tr := trans as TPQTrans;

  res := PQexec(tr.PGConn, 'ROLLBACK');

  CheckResultError(res,tr.PGConn,SErrRollbackFailed);

  PQclear(res);
  //make connection available in pool
  for i:=0 to length(FConnectionPool)-1 do
    if FConnectionPool[i].FPGConn=tr.PGConn then
      begin
      FConnectionPool[i].FTranActive:=false;
      break;
      end;
  result := true;
end;

function TPQConnection.Commit(trans : TSQLHandle) : boolean;
var
  res : PPGresult;
  tr  : TPQTrans;
  i   : Integer;
begin
  result := false;

  tr := trans as TPQTrans;

  res := PQexec(tr.PGConn, 'COMMIT');
  CheckResultError(res,tr.PGConn,SErrCommitFailed);

  PQclear(res);
  //make connection available in pool
  for i:=0 to length(FConnectionPool)-1 do
    if FConnectionPool[i].FPGConn=tr.PGConn then
      begin
      FConnectionPool[i].FTranActive:=false;
      break;
      end;
  result := true;
end;

function TPQConnection.StartdbTransaction(trans : TSQLHandle; AParams : string) : boolean;
var
  res : PPGresult;
  tr  : TPQTrans;
  i   : Integer;
begin
  result:=false;
  tr := trans as TPQTrans;

  //find an unused connection in the pool
  i:=0;
  while i<length(FConnectionPool) do
    if (FConnectionPool[i].FPGConn=nil) or not FConnectionPool[i].FTranActive then
      break
    else
      i:=i+1;
  if i=length(FConnectionPool) then //create a new connection
    begin
    tr.PGConn := PQconnectdb(pchar(FConnectString));
    CheckConnectionStatus(tr.PGConn);

    if CharSet <> '' then
      PQsetClientEncoding(tr.PGConn, pchar(CharSet));

    //store the new connection
    SetLength(FConnectionPool,i+1);
    FConnectionPool[i]:=TPQTranConnection.Create;
    FConnectionPool[i].FPGConn:=tr.PGConn;
    FConnectionPool[i].FTranActive:=true;
    end
  else //re-use existing connection
    begin
    tr.PGConn:=FConnectionPool[i].FPGConn;
    FConnectionPool[i].FTranActive:=true;
    end;

  res := PQexec(tr.PGConn, 'BEGIN');
  CheckResultError(res,tr.PGConn,sErrTransactionFailed);

  PQclear(res);
  result := true;
end;

procedure TPQConnection.RollBackRetaining(trans : TSQLHandle);
var
  res : PPGresult;
  tr  : TPQTrans;
begin
  tr := trans as TPQTrans;
  res := PQexec(tr.PGConn, 'ROLLBACK');
  CheckResultError(res,tr.PGConn,SErrRollbackFailed);

  PQclear(res);
  res := PQexec(tr.PGConn, 'BEGIN');
  CheckResultError(res,tr.PGConn,sErrTransactionFailed);

  PQclear(res);
end;

procedure TPQConnection.CommitRetaining(trans : TSQLHandle);
var
  res : PPGresult;
  tr  : TPQTrans;
begin
  tr := trans as TPQTrans;
  res := PQexec(tr.PGConn, 'COMMIT');
  CheckResultError(res,tr.PGConn,SErrCommitFailed);

  PQclear(res);
  res := PQexec(tr.PGConn, 'BEGIN');
  CheckResultError(res,tr.PGConn,sErrTransactionFailed);

  PQclear(res);
end;


procedure TPQConnection.DoInternalConnect;
var ASQLDatabaseHandle   : PPGConn;
begin
{$IfDef LinkDynamically}
  InitialisePostgres3;
{$EndIf}

  inherited DoInternalConnect;

  FConnectString := '';
  if (UserName <> '') then FConnectString := FConnectString + ' user=''' + UserName + '''';
  if (Password <> '') then FConnectString := FConnectString + ' password=''' + Password + '''';
  if (HostName <> '') then FConnectString := FConnectString + ' host=''' + HostName + '''';
  if (DatabaseName <> '') then FConnectString := FConnectString + ' dbname=''' + DatabaseName + '''';
  if (Params.Text <> '') then FConnectString := FConnectString + ' '+Params.Text;

  ASQLDatabaseHandle := PQconnectdb(pchar(FConnectString));
  try
    CheckConnectionStatus(ASQLDatabaseHandle);
  except
    DoInternalDisconnect;
    raise;
  end;

  // This only works for pg>=8.0, so timestamps won't work with earlier versions of pg which are compiled with integer_datetimes on
  if PQparameterStatus<>nil then
    FIntegerDateTimes := PQparameterStatus(ASQLDatabaseHandle,'integer_datetimes') = 'on';

  SetLength(FConnectionPool,1);
  FConnectionPool[0]:=TPQTranConnection.Create;
  FConnectionPool[0].FPGConn:=ASQLDatabaseHandle;
  FConnectionPool[0].FTranActive:=false;
end;

procedure TPQConnection.DoInternalDisconnect;
var i:integer;
begin
  Inherited;
  for i:=0 to length(FConnectionPool)-1 do
    begin
    if assigned(FConnectionPool[i].FPGConn) then
      PQfinish(FConnectionPool[i].FPGConn);
    FConnectionPool[i].Free;
    end;
  Setlength(FConnectionPool,0);
{$IfDef LinkDynamically}
  ReleasePostgres3;
{$EndIf}
end;

procedure TPQConnection.CheckConnectionStatus(var conn: PPGconn);
var sErr: string;
    i: integer;
begin
  if (PQstatus(conn) = CONNECTION_BAD) then
    begin
    sErr := PQerrorMessage(conn);
    //make connection available in pool
    for i:=0 to length(FConnectionPool)-1 do
      if FConnectionPool[i].FPGConn=conn then
        begin
        FConnectionPool[i].FPGConn:=nil;
        FConnectionPool[i].FTranActive:=false;
        break;
        end;
    PQfinish(conn);
    DatabaseError(sErrConnectionFailed + ' (PostgreSQL: ' + sErr + ')', Self);
    end;
end;

procedure TPQConnection.CheckResultError(var res: PPGresult; conn: PPGconn;
  ErrMsg: string);

  Procedure MaybeAdd(Var S : String; Prefix,Msg : String);

  begin
    if (Msg='') then
      exit;
    S:=S+LineEnding+Prefix+': '+Msg;
  end;

var
  E: EPQDatabaseError;
  sErr: string;
  CompName: string;
  SEVERITY: string;
  SQLSTATE: string;
  MESSAGE_PRIMARY: string;
  MESSAGE_DETAIL: string;
  MESSAGE_HINT: string;
  STATEMENT_POSITION: string;
  i:Integer;
begin
  if (PQresultStatus(res) <> PGRES_COMMAND_OK) then
    begin
    SEVERITY:=PQresultErrorField(res,ord('S'));
    SQLSTATE:=PQresultErrorField(res,ord('C'));
    MESSAGE_PRIMARY:=PQresultErrorField(res,ord('M'));
    MESSAGE_DETAIL:=PQresultErrorField(res,ord('D'));
    MESSAGE_HINT:=PQresultErrorField(res,ord('H'));
    STATEMENT_POSITION:=PQresultErrorField(res,ord('P'));
    sErr:=PQresultErrorMessage(res);
    if VerboseErrors then
      begin
      MaybeAdd(sErr,'Severity',SEVERITY);
      MaybeAdd(sErr,'SQL State',SQLSTATE);
      MaybeAdd(sErr,'Primary Error',MESSAGE_PRIMARY);
      MaybeAdd(sErr,'Error Detail',MESSAGE_DETAIL);
      MaybeAdd(sErr,'Hint',MESSAGE_HINT);
      MaybeAdd(sErr,'Character',STATEMENT_POSITION);
      end;
    if (Self.Name='') then CompName := Self.ClassName else CompName := Self.Name;
    E:=EPQDatabaseError.CreateFmt('%s : %s  (PostgreSQL: %s)', [CompName, ErrMsg, sErr]);
    E.SEVERITY:=SEVERITY;
    E.SQLSTATE:=SQLSTATE;
    E.MESSAGE_PRIMARY:=MESSAGE_PRIMARY;
    E.MESSAGE_DETAIL:=MESSAGE_DETAIL;
    E.MESSAGE_HINT:=MESSAGE_HINT;
    E.STATEMENT_POSITION:=STATEMENT_POSITION;

    PQclear(res);
    res:=nil;
    if assigned(conn) then
      begin
      PQFinish(conn);
      //make connection available in pool
      for i:=0 to length(FConnectionPool)-1 do
        if FConnectionPool[i].FPGConn=conn then
          begin
          FConnectionPool[i].FPGConn:=nil;
          FConnectionPool[i].FTranActive:=false;
          break;
          end;
      end;
    raise E;
    end;
end;

function TPQConnection.TranslateFldType(res : PPGresult; Tuple : integer; out Size : integer) : TFieldType;
const VARHDRSZ=sizeof(longint);
var li : longint;
begin
  Size := 0;
  case PQftype(res,Tuple) of
    Oid_varchar,Oid_bpchar,
    Oid_name               : begin
                             Result := ftstring;
                             size := PQfsize(Res, Tuple);
                             if (size = -1) then
                               begin
                               li := PQfmod(res,Tuple);
                               if li = -1 then
                                 size := dsMaxStringSize
                               else
                                 size := (li-VARHDRSZ) and $FFFF;
                               end;
                             if size > MaxSmallint then size := MaxSmallint;
                             end;
//    Oid_text               : Result := ftstring;
    Oid_text               : Result := ftMemo;
    Oid_Bytea              : Result := ftBlob;
    Oid_oid                : Result := ftInteger;
    Oid_int8               : Result := ftLargeInt;
    Oid_int4               : Result := ftInteger;
    Oid_int2               : Result := ftSmallInt;
    Oid_Float4             : Result := ftFloat;
    Oid_Float8             : Result := ftFloat;
    Oid_TimeStamp,
    Oid_TimeStampTZ        : Result := ftDateTime;
    Oid_Date               : Result := ftDate;
    Oid_Interval,
    Oid_Time,
    Oid_TimeTZ             : Result := ftTime;
    Oid_Bool               : Result := ftBoolean;
    Oid_Numeric            : begin
                             Result := ftBCD;
                             li := PQfmod(res,Tuple);
                             if li = -1 then
                               size := 4 // No information about the size available, use the maximum value
                             else
                             // The precision is the high 16 bits, the scale the
                             // low 16 bits with an offset of sizeof(int32).
                               begin
                               size := (li-VARHDRSZ) and $FFFF;
                               if (size > MaxBCDScale) or ((li shr 16)-size > MaxBCDPrecision-MaxBCDScale) then
                                 Result := ftFmtBCD;
                               end;
                             end;
    Oid_Money              : Result := ftCurrency;
    Oid_char               : begin
                             Result := ftFixedChar;
                             Size := 1;
                             end;
    Oid_uuid               : begin
                             Result := ftGuid;
                             Size := 38;
                             end;
    Oid_MacAddr            : begin
                             Result := ftFixedChar;
                             Size := 17;
                             end;
    Oid_Inet               : begin
                             Result := ftString;
                             Size := 39;
                             end;
    Oid_Unknown            : Result := ftUnknown;
  else
    Result := ftUnknown;
  end;
end;

Function TPQConnection.AllocateCursorHandle : TSQLCursor;

begin
  result := TPQCursor.create;
end;

Procedure TPQConnection.DeAllocateCursorHandle(var cursor : TSQLCursor);
begin
  FreeAndNil(cursor);
end;

Function TPQConnection.AllocateTransactionHandle : TSQLHandle;

begin
  result := TPQTrans.create;
end;

procedure TPQConnection.PrepareStatement(cursor: TSQLCursor;ATransaction : TSQLTransaction;buf : string; AParams : TParams);
                          
const TypeStrings : array[TFieldType] of string =
    (
      'Unknown',   // ftUnknown
      'text',      // ftString
      'smallint',  // ftSmallint
      'int',       // ftInteger
      'int',       // ftWord
      'bool',      // ftBoolean
      'float',     // ftFloat
      'money',     // ftCurrency
      'numeric',   // ftBCD
      'date',      // ftDate
      'time',      // ftTime
      'timestamp', // ftDateTime
      'Unknown',   // ftBytes
      'Unknown',   // ftVarBytes
      'Unknown',   // ftAutoInc
      'bytea',     // ftBlob 
      'text',      // ftMemo
      'bytea',     // ftGraphic
      'text',      // ftFmtMemo
      'Unknown',   // ftParadoxOle
      'Unknown',   // ftDBaseOle
      'Unknown',   // ftTypedBinary
      'Unknown',   // ftCursor
      'char',      // ftFixedChar
      'text',      // ftWideString
      'bigint',    // ftLargeint
      'Unknown',   // ftADT
      'Unknown',   // ftArray
      'Unknown',   // ftReference
      'Unknown',   // ftDataSet
      'Unknown',   // ftOraBlob
      'Unknown',   // ftOraClob
      'Unknown',   // ftVariant
      'Unknown',   // ftInterface
      'Unknown',   // ftIDispatch
      'uuid',      // ftGuid
      'Unknown',   // ftTimeStamp
      'numeric',   // ftFMTBcd
      'Unknown',   // ftFixedWideChar
      'Unknown'    // ftWideMemo
    );


var s : string;
    i : integer;

begin
  with (cursor as TPQCursor) do
    begin
    FPrepared := False;
    // Prior to v8 there is no support for cursors and parameters.
    // So that's not supported.
    if FStatementType in [stInsert,stUpdate,stDelete, stSelect] then
      begin
      StmtName := 'prepst'+inttostr(FCursorCount);
      inc(FCursorCount);
      tr := TPQTrans(aTransaction.Handle);
      // Only available for pq 8.0, so don't use it...
      // Res := pqprepare(tr,'prepst'+name+nr,pchar(buf),params.Count,pchar(''));
      s := 'prepare '+StmtName+' ';
      if Assigned(AParams) and (AParams.Count > 0) then
        begin
        s := s + '(';
        for i := 0 to AParams.Count-1 do
          if TypeStrings[AParams[i].DataType] <> 'Unknown' then
            s := s + TypeStrings[AParams[i].DataType] + ','
          else
            begin
            if AParams[i].DataType = ftUnknown then
              begin
              if AParams[i].IsNull then
                s:=s+' unknown ,'
              else
                DatabaseErrorFmt(SUnknownParamFieldType,[AParams[i].Name],self)
              end
            else
              DatabaseErrorFmt(SUnsupportedParameter,[Fieldtypenames[AParams[i].DataType]],self);
            end;
        s[length(s)] := ')';
        buf := AParams.ParseSQL(buf,false,sqEscapeSlash in ConnOptions, sqEscapeRepeat in ConnOptions,psPostgreSQL);
        end;
      s := s + ' as ' + buf;
      if LogEvent(detPrepare) then
        Log(detPrepare,S);
      res := PQexec(tr.PGConn,pchar(s));
      CheckResultError(res,nil,SErrPrepareFailed);
      // if statement is INSERT, UPDATE, DELETE with RETURNING clause, then
      // override the statement type derrived by parsing the query.
      if (FStatementType in [stInsert,stUpdate,stDelete]) and (pos('RETURNING', upcase(s)) > 0) then
        begin
        PQclear(res);
        res := PQdescribePrepared(tr.PGConn,pchar(StmtName));
        if (PQresultStatus(res) = PGRES_COMMAND_OK) and (PQnfields(res) > 0) then
          FStatementType := stSelect;
        end;
      FPrepared := True;
      end
    else
      Statement := AParams.ParseSQL(buf,false,sqEscapeSlash in ConnOptions, sqEscapeRepeat in ConnOptions,psPostgreSQL);
    end;
end;

procedure TPQConnection.UnPrepareStatement(cursor : TSQLCursor);
begin
  with (cursor as TPQCursor) do
    begin
    PQclear(res);
    res:=nil;
    if FPrepared then
      begin
      if PQtransactionStatus(tr.PGConn) <> PQTRANS_INERROR then
        begin
        res := PQexec(tr.PGConn,pchar('deallocate '+StmtName));
        CheckResultError(res,nil,SErrUnPrepareFailed);
        PQclear(res);
        res:=nil;
        end;
      FPrepared := False;
      end;
    end;
end;

procedure TPQConnection.Execute(cursor: TSQLCursor;atransaction:tSQLtransaction;AParams : TParams);

var ar  : array of pchar;
    l,i : integer;
    s   : string;
    lengths,formats : array of integer;
    ParamNames,
    ParamValues : array of string;
    cash: int64;

begin
  with cursor as TPQCursor do
    begin
    PQclear(res);
    if FStatementType in [stInsert,stUpdate,stDelete,stSelect] then
      begin
      if Assigned(AParams) and (AParams.Count > 0) then
        begin
        l:=AParams.Count;
        setlength(ar,l);
        setlength(lengths,l);
        setlength(formats,l);
        for i := 0 to AParams.Count -1 do if not AParams[i].IsNull then
          begin
          case AParams[i].DataType of
            ftDateTime:
              s := FormatDateTime('yyyy-mm-dd hh:nn:ss.zzz', AParams[i].AsDateTime);
            ftDate:
              s := FormatDateTime('yyyy-mm-dd', AParams[i].AsDateTime);
            ftTime:
              s := FormatDateTime('hh:nn:ss.zzz', AParams[i].AsDateTime);
            ftFloat, ftBCD:
              Str(AParams[i].AsFloat, s);
            ftCurrency:
              begin
                cash:=NtoBE(round(AParams[i].AsCurrency*100));
                setlength(s, sizeof(cash));
                Move(cash, s[1], sizeof(cash));
              end;
            ftFmtBCD:
              s := BCDToStr(AParams[i].AsFMTBCD, FSQLFormatSettings);
            else
              s := AParams[i].AsString;
          end; {case}
          GetMem(ar[i],length(s)+1);
          StrMove(PChar(ar[i]),Pchar(s),Length(S)+1);
          lengths[i]:=Length(s);
          if (AParams[i].DataType in [ftBlob,ftMemo,ftGraphic,ftCurrency]) then
            Formats[i]:=1
          else
            Formats[i]:=0;  
          end
        else
          FreeAndNil(ar[i]);
        res := PQexecPrepared(tr.PGConn,pchar(StmtName),AParams.Count,@Ar[0],@Lengths[0],@Formats[0],1);
        for i := 0 to AParams.Count -1 do
          FreeMem(ar[i]);
        end
      else
        res := PQexecPrepared(tr.PGConn,pchar(StmtName),0,nil,nil,nil,1);
      end
    else
      begin
      tr := TPQTrans(aTransaction.Handle);

      if Assigned(AParams) and (AParams.Count > 0) then
        begin
        setlength(ParamNames,AParams.Count);
        setlength(ParamValues,AParams.Count);
        for i := 0 to AParams.Count -1 do
          begin
          ParamNames[AParams.Count-i-1] := '$'+inttostr(AParams[i].index+1);
          ParamValues[AParams.Count-i-1] := GetAsSQLText(AParams[i]);
          end;
        s := stringsreplace(Statement,ParamNames,ParamValues,[rfReplaceAll]);
        end
      else
        s := Statement;
      res := PQexec(tr.PGConn,pchar(s));
      if (PQresultStatus(res) in [PGRES_COMMAND_OK]) then
        begin
        PQclear(res);
        res:=nil;
        end;
      end;

    if assigned(res) and not (PQresultStatus(res) in [PGRES_COMMAND_OK,PGRES_TUPLES_OK]) then
      begin
      // Don't perform the rollback, only make it possible to do a rollback.
      // The other databases also don't do this.
      //atransaction.Rollback;
      CheckResultError(res,nil,SErrExecuteFailed);
      end;

    FSelectable := assigned(res) and (PQresultStatus(res)=PGRES_TUPLES_OK);
    end;
end;


procedure TPQConnection.AddFieldDefs(cursor: TSQLCursor; FieldDefs : TfieldDefs);
var
  i         : integer;
  size      : integer;
  fieldtype : tfieldtype;
  nFields   : integer;

begin
  with cursor as TPQCursor do
    begin
    nFields := PQnfields(Res);
    setlength(FieldBinding,nFields);
    for i := 0 to nFields-1 do
      begin
      fieldtype := TranslateFldType(Res, i,size);
      with TFieldDef.Create(FieldDefs, FieldDefs.MakeNameUnique(PQfname(Res, i)), fieldtype,size, False, (i + 1)) do
        FieldBinding[FieldNo-1] := i;
      end;
    CurTuple := -1;
    end;
end;

function TPQConnection.GetHandle: pointer;
var
  i:integer;
begin
  result:=nil;
  if not Connected then
    exit;
  //Get any handle that is (still) connected
  for i:=0 to length(FConnectionPool)-1 do
    if assigned(FConnectionPool[i].FPGConn) and (PQstatus(FConnectionPool[i].FPGConn)<>CONNECTION_BAD) then
      begin
      Result :=FConnectionPool[i].FPGConn;
      exit;
      end;
  //Nothing connected!! Reconnect
  if assigned(FConnectionPool[0].FPGConn) then
    PQreset(FConnectionPool[0].FPGConn)
  else
    FConnectionPool[0].FPGConn := PQconnectdb(pchar(FConnectString));
  CheckConnectionStatus(FConnectionPool[0].FPGConn);
  if CharSet <> '' then
    PQsetClientEncoding(FConnectionPool[0].FPGConn, pchar(CharSet));
  result:=FConnectionPool[0].FPGConn;
end;

function TPQConnection.Fetch(cursor : TSQLCursor) : boolean;

begin
  with cursor as TPQCursor do
    begin
    inc(CurTuple);
    Result := (PQntuples(res)>CurTuple);
    end;
end;

function TPQConnection.LoadField(cursor : TSQLCursor;FieldDef : TfieldDef;buffer : pointer; out CreateBlob : boolean) : boolean;

const NBASE=10000;
      DAYS_PER_MONTH=30;

type TNumericRecord = record
       Digits : SmallInt;
       Weight : SmallInt;
       Sign   : SmallInt;
       Scale  : Smallint;
     end;
     TIntervalRec = packed record
       time  : int64;
       day   : longint;
       month : longint;
     end;
     TMacAddrRec = packed record
       a, b, c, d, e, f: byte;
     end;
     TInetRec = packed record
       family : byte;
       bits   : byte;
       is_cidr: byte;
       nb     : byte;
       ipaddr : array[1..16] of byte;
     end;

var
  x,i           : integer;
  s             : string;
  li            : Longint;
  CurrBuff      : pchar;
  dbl           : pdouble;
  cur           : currency;
  NumericRecord : ^TNumericRecord;
  guid          : TGUID;
  bcd           : TBCD;
  macaddr       : ^TMacAddrRec;
  inet          : ^TInetRec;

begin
  Createblob := False;
  with cursor as TPQCursor do
    begin
    x := FieldBinding[FieldDef.FieldNo-1];

    // Joost, 5 jan 2006: I disabled the following, since it's useful for
    // debugging, but it also slows things down. In principle things can only go
    // wrong when FieldDefs is changed while the dataset is opened. A user just
    // shoudn't do that. ;) (The same is done in IBConnection)
    //if PQfname(Res, x) <> FieldDef.Name then
    //  DatabaseErrorFmt(SFieldNotFound,[FieldDef.Name],self);

    if pqgetisnull(res,CurTuple,x)=1 then
      result := false
    else
      begin
      CurrBuff := pqgetvalue(res,CurTuple,x);

      result := true;

      case FieldDef.DataType of
        ftInteger, ftSmallint, ftLargeInt :
          case PQfsize(res, x) of  // postgres returns big-endian numbers
            sizeof(int64) : pint64(buffer)^ := BEtoN(pint64(CurrBuff)^); // INT8
            sizeof(integer) : pinteger(buffer)^ := BEtoN(pinteger(CurrBuff)^); // INT4
            sizeof(smallint) : psmallint(buffer)^ := BEtoN(psmallint(CurrBuff)^); // INT2
          end; {case}
        ftFloat :
          case PQfsize(res, x) of  // postgres returns big-endian numbers
            sizeof(int64) :  // FLOAT8
              pint64(buffer)^ := BEtoN(pint64(CurrBuff)^);
            sizeof(integer) :  // FLOAT4
              begin
              li := BEtoN(pinteger(CurrBuff)^);
              pdouble(buffer)^ := psingle(@li)^
              end;
          end; {case}
        ftString, ftFixedChar :
          begin
          case PQftype(res, x) of
            Oid_MacAddr:
            begin
              macaddr := Pointer(CurrBuff);
              li := FormatBuf(Buffer^, FieldDef.Size, '%.2x:%.2x:%.2x:%.2x:%.2x:%.2x', 29,
                    [macaddr^.a,macaddr^.b,macaddr^.c,macaddr^.d,macaddr^.e,macaddr^.f]);
            end;
            Oid_Inet:
            begin
              inet := Pointer(CurrBuff);
              if inet^.nb = 4 then
                li := FormatBuf(Buffer^, FieldDef.Size, '%d.%d.%d.%d', 11,
                      [inet^.ipaddr[1],inet^.ipaddr[2],inet^.ipaddr[3],inet^.ipaddr[4]])
              else if inet^.nb = 16 then
                li := FormatBuf(Buffer^, FieldDef.Size, '%x%.2x:%x%.2x:%x%.2x:%x%.2x:%x%.2x:%x%.2x:%x%.2x:%x%.2x', 55,
                      [inet^.ipaddr[1],inet^.ipaddr[2],inet^.ipaddr[3],inet^.ipaddr[4],inet^.ipaddr[5],inet^.ipaddr[6],inet^.ipaddr[7],inet^.ipaddr[8],inet^.ipaddr[9],inet^.ipaddr[10],inet^.ipaddr[11],inet^.ipaddr[12],inet^.ipaddr[13],inet^.ipaddr[14],inet^.ipaddr[15],inet^.ipaddr[16]])
              else
                li := 0;
            end
            else
            begin
              li := pqgetlength(res,curtuple,x);
              if li > FieldDef.Size then li := FieldDef.Size;
              Move(CurrBuff^, Buffer^, li);
            end;
          end;
          pchar(Buffer + li)^ := #0;
          end;
        ftBlob, ftMemo :
          CreateBlob := True;
        ftDate :
          begin
          dbl := pointer(buffer);
          dbl^ := BEtoN(plongint(CurrBuff)^) + 36526;
          end;
        ftDateTime, ftTime :
          begin
          dbl := pointer(buffer);
          if FIntegerDateTimes then
            dbl^ := BEtoN(pint64(CurrBuff)^) / 1000000
          else
            pint64(dbl)^ := BEtoN(pint64(CurrBuff)^);
          case PQftype(res, x) of
            Oid_Timestamp, Oid_TimestampTZ:
              dbl^ := dbl^ + 3.1558464E+009; // postgres counts seconds elapsed since 1-1-2000
            Oid_Interval:
              dbl^ := dbl^ + BEtoN(plongint(CurrBuff+ 8)^) * SecsPerDay
                           + BEtoN(plongint(CurrBuff+12)^) * SecsPerDay * DAYS_PER_MONTH;
          end;
          dbl^ := dbl^ / SecsPerDay;
          // Now convert the mathematically-correct datetime to the
          // illogical windows/delphi/fpc TDateTime:
          if (dbl^ <= 0) and (frac(dbl^) < 0) then
            dbl^ := trunc(dbl^)-2-frac(dbl^);
          end;
        ftBCD, ftFmtBCD:
          begin
          NumericRecord := pointer(CurrBuff);
          NumericRecord^.Digits := BEtoN(NumericRecord^.Digits);
          NumericRecord^.Weight := BEtoN(NumericRecord^.Weight);
          NumericRecord^.Sign := BEtoN(NumericRecord^.Sign);
          NumericRecord^.Scale := BEtoN(NumericRecord^.Scale);
          inc(pointer(currbuff),sizeof(TNumericRecord));
          if (NumericRecord^.Digits = 0) and (NumericRecord^.Scale = 0) then // = NaN, which is not supported by Currency-type, so we return NULL
            result := false
          else if FieldDef.DataType = ftBCD then
            begin
            cur := 0;
            for i := 0 to NumericRecord^.Digits-1 do
              begin
              cur := cur + beton(pword(CurrBuff)^) * intpower(NBASE, NumericRecord^.weight-i);
              inc(pointer(CurrBuff),2);
              end;
            if NumericRecord^.Sign <> 0 then cur := -cur;
            Move(Cur, Buffer^, sizeof(currency));
            end
          else //ftFmtBCD
            begin
            bcd := 0;
            for i := 0 to NumericRecord^.Digits-1 do
              begin
              BCDAdd(bcd, beton(pword(CurrBuff)^) * intpower(NBASE, NumericRecord^.weight-i), bcd);
              inc(pointer(CurrBuff),2);
              end;
            if NumericRecord^.Sign <> 0 then BCDNegate(bcd);
            Move(bcd, Buffer^, sizeof(bcd));
            end;
          end;
        ftCurrency  :
          begin
          dbl := pointer(buffer);
          dbl^ := BEtoN(PInt64(CurrBuff)^) / 100;
          end;
        ftBoolean:
          pchar(buffer)[0] := CurrBuff[0];
        ftGuid:
          begin
          Move(CurrBuff^, guid, sizeof(guid));
          guid.D1:=BEtoN(guid.D1);
          guid.D2:=BEtoN(guid.D2);
          guid.D3:=BEtoN(guid.D3);
          s:=GUIDToString(guid);
          StrPLCopy(PChar(Buffer), s, FieldDef.Size);
          end
        else
          result := false;
      end;
      end;
    end;
end;

procedure TPQConnection.UpdateIndexDefs(IndexDefs : TIndexDefs;TableName : string);

var qry : TSQLQuery;
    relname : string;

begin
  if not assigned(Transaction) then
    DatabaseError(SErrConnTransactionnSet);

  if (length(TableName)>2) and (TableName[1]='"') and (TableName[length(TableName)]='"') then
    relname := QuotedStr(AnsiDequotedStr(TableName, '"'))
  else
    relname := 'lower(' + QuotedStr(TableName) + ')';  // unquoted names are stored lower case in PostgreSQL which is incompatible with the SQL standard

  qry := tsqlquery.Create(nil);
  qry.transaction := Transaction;
  qry.database := Self;
  with qry do
    begin
    ReadOnly := True;
    sql.clear;

    sql.add('select '+
              'ic.relname as indexname,  '+
              'tc.relname as tablename, '+
              'ia.attname, '+
              'i.indisprimary, '+
              'i.indisunique '+
            'from '+
              'pg_attribute ta, '+
              'pg_attribute ia, '+
              'pg_class tc, '+
              'pg_class ic, '+
              'pg_index i '+
            'where '+
              '(i.indrelid = tc.oid) and '+
              '(ta.attrelid = tc.oid) and '+
              '(ia.attrelid = i.indexrelid) and '+
              '(ic.oid = i.indexrelid) and '+
              '(ta.attnum = i.indkey[ia.attnum-1]) and '+
              '(tc.relname = ' + relname + ') '+
            'order by '+
              'ic.relname;');
    open;
    end;
  while not qry.eof do with IndexDefs.AddIndexDef do
    begin
    Name := trim(qry.fields[0].asstring);
    Fields := trim(qry.Fields[2].asstring);
    If qry.fields[3].asboolean then options := options + [ixPrimary];
    If qry.fields[4].asboolean then options := options + [ixUnique];
    qry.next;
    while (name = qry.fields[0].asstring) and (not qry.eof) do
      begin
      Fields := Fields + ';' + trim(qry.Fields[2].asstring);
      qry.next;
      end;
    end;
  qry.close;
  qry.free;
end;

function TPQConnection.GetSchemaInfoSQL(SchemaType: TSchemaType;
  SchemaObjectName, SchemaPattern: string): string;

var s : string;

begin
  // select * from information_schema.tables with 
  // where table_schema [not] in ('pg_catalog','information_schema') may be better.
  // But the following should work:
  case SchemaType of
    stTables     : s := 'select '+
                          'relfilenode        as recno, '+
                          'current_database() as catalog_name, '+
                          'nspname            as schema_name, '+
                          'relname            as table_name, '+
                          '0                  as table_type '+
                        'from pg_class c '+
                          'left join pg_namespace n on c.relnamespace=n.oid '+
                        'where (relkind=''r'') and not (nspname in (''pg_catalog'',''information_schema''))' +
                        'order by relname';

    stSysTables  : s := 'select '+
                          'relfilenode        as recno, '+
                          'current_database() as catalog_name, '+
                          'nspname            as schema_name, '+
                          'relname            as table_name, '+
                          '0                  as table_type '+
                        'from pg_class c '+
                          'left join pg_namespace n on c.relnamespace=n.oid '+
                        'where (relkind=''r'') and nspname in ((''pg_catalog'',''information_schema'')) ' + // only system tables
                        'order by relname';
    stColumns    : s := 'select '+
                          'a.attnum           as recno, '+
                          'current_database() as catalog_name, '+
                          'nspname            as schema_name, '+
                          'c.relname          as table_name, '+
                          'a.attname          as column_name, '+
                          '0                  as column_position, '+
                          '0                  as column_type, '+
                          '0                  as column_datatype, '+
                          '''''               as column_typename, '+
                          '0                  as column_subtype, '+
                          '0                  as column_precision, '+
                          '0                  as column_scale, '+
                          'a.atttypmod        as column_length, '+
                          'not a.attnotnull   as column_nullable '+
                        'from pg_class c '+
                          'join pg_attribute a on c.oid=a.attrelid '+
                          'left join pg_namespace n on c.relnamespace=n.oid '+
                          // This can lead to problems when case-sensitive tablenames are used.
                        'where (a.attnum>0) and (not a.attisdropped) and (upper(c.relname)=''' + Uppercase(SchemaObjectName) + ''') '+
                        'order by a.attname';
  else
    s := inherited;
  end; {case}
  result := s;
end;

procedure TPQConnection.LoadBlobIntoBuffer(FieldDef: TFieldDef;
  ABlobBuf: PBufBlobField; cursor: TSQLCursor; ATransaction: TSQLTransaction);
var
  x             : integer;
  li            : Longint;
begin
  with cursor as TPQCursor do
    begin
    x := FieldBinding[FieldDef.FieldNo-1];
    li := pqgetlength(res,curtuple,x);
    ReAllocMem(ABlobBuf^.BlobBuffer^.Buffer,li);
    Move(pqgetvalue(res,CurTuple,x)^, ABlobBuf^.BlobBuffer^.Buffer^, li);
    ABlobBuf^.BlobBuffer^.Size := li;
    end;
end;

function TPQConnection.RowsAffected(cursor: TSQLCursor): TRowsCount;
begin
  if assigned(cursor) and assigned((cursor as TPQCursor).res) then
    Result := StrToIntDef(PQcmdTuples((cursor as TPQCursor).res),-1)
  else
    Result := -1;
end;

function TPQConnection.GetConnectionInfo(InfoType: TConnInfoType): string;
begin
  Result:='';
  try
    {$IFDEF LinkDynamically}
    InitialisePostgres3;
    {$ENDIF}
    case InfoType of
      citServerType:
        Result:=TPQConnectionDef.TypeName;
      citServerVersion,
      citServerVersionString:
        if Connected then
          Result:=format('%6.6d', [PQserverVersion(GetHandle)]);
      citClientName:
        Result:=TPQConnectionDef.LoadedLibraryName;
    else
      Result:=inherited GetConnectionInfo(InfoType);
    end;
  finally
    {$IFDEF LinkDynamically}
    ReleasePostgres3;
    {$ENDIF}
  end;
end;


{ TPQConnectionDef }

class function TPQConnectionDef.TypeName: String;
begin
  Result:='PostgreSQL';
end;

class function TPQConnectionDef.ConnectionClass: TSQLConnectionClass;
begin
  Result:=TPQConnection;
end;

class function TPQConnectionDef.Description: String;
begin
  Result:='Connect to a PostgreSQL database directly via the client library';
end;

class function TPQConnectionDef.DefaultLibraryName: String;
begin
  {$IfDef LinkDynamically}
  Result:=pqlib;
  {$else}
  Result:='';
  {$endif}
end;

class function TPQConnectionDef.LoadFunction: TLibraryLoadFunction;
begin
  {$IfDef LinkDynamically}
  Result:=@InitialisePostgres3;
  {$else}
  Result:=Nil;
  {$endif}
end;

class function TPQConnectionDef.UnLoadFunction: TLibraryUnLoadFunction;
begin
  {$IfDef LinkDynamically}
  Result:=@ReleasePostgres3;
  {$else}
  Result:=Nil;
  {$endif}
end;

class function TPQConnectionDef.LoadedLibraryName: string;
begin
  {$IfDef LinkDynamically}
  Result:=Postgres3LoadedLibrary;
  {$else}
  Result:='';
  {$endif}
end;

initialization
  RegisterConnection(TPQConnectionDef);
finalization
  UnRegisterConnection(TPQConnectionDef);
end.