PostgreSQL COPY FROM STDIN
lazarus / 16-Databases

----------------------------------------------------------------------
SymbolicFrank · 2023-04-11 08:41

How do I do that? (The target computer is only accessible through a
socket.)

Something like this?
Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i
Connection.ExecuteDirect('copy <tablename> from stdin  with delimiter
E''\t'' csv header;');for s in CSVFile dobegin 
Connection.ExecuteDirect(s);end;
I am using a TSpecialCSVExporter to generate the file, can I redirect
the file on Execute? It inherits from TStreamExporter.

----------------------------------------------------------------------
marcov · 2023-04-11 09:22

I think that is functionality from the psql commandline tool, not the
client library.

----------------------------------------------------------------------
SymbolicFrank · 2023-04-11 09:28

It's an SQL command. There's also a command line '/copy' command, but
that's not what I'm using.

Edit: this is what I'm doing now:

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i csv :=
TSpecialCSVExporter.Create(MyConnection, ThisSchema, ThisTable);try 
csv.Dataset:= MyDataset;  csv.FileName:= WinDir + ThisTable + '.csv'; 
csv.FormatSettings.FieldDelimiter := #9;  csv.FormatSettings.HeaderRow
:= True;  csv.FormatSettings.QuoteChar := '"'; 
csv.FormatSettings.DateFormat := 'yyyy-mm-dd'; 
csv.FormatSettings.DateTimeFormat := 'yyyy-mm-dd"T"hh:mm:ss'; 
csv.FormatSettings.TimeFormat := 'hh:mm:ss';  csv.Execute;finally 
csv.Free;end;MyConnection.ExecuteDirect('copy ' + ThisTable + ' from
''' + LinuxDir +   ThisTable + '.csv' + ''' with delimiter E''\t'' csv
header;');
But that requires a shared directory (WinDir and LinuxDir,
respectively).

----------------------------------------------------------------------
marcov · 2023-04-11 10:02

Quote from: SymbolicFrank on April 11, 2023, 11:28:23 amIt's an SQL
command. There's also a command line '/copy' command, but that's not
what I'm using.

And if you search for stdin in that link, it says "copy from the
client application".

----------------------------------------------------------------------
SymbolicFrank · 2023-04-11 10:30

Quote from: marcov on April 11, 2023, 12:02:25 pmQuote from:
SymbolicFrank on April 11, 2023, 11:28:23 amIt's an SQL command.
There's also a command line '/copy' command, but that's not what I'm
using.

And if you search for stdin in that link, it says "copy from the
client application".

Yes. Above that, it says: "When STDIN or STDOUT is specified, data is
transmitted via the connection between the client and the server."

My example is how it works with Go, across the DB socket to a
different server.

----------------------------------------------------------------------
Zvoni · 2023-04-11 10:49

What i don't understand: if you can establish a connection from the
client (Your App?) to the PG-Server, why do you want ot read input-
data from Stdin??
In that case: what's that got to do with "The target computer is only
accessible through a socket"?

----------------------------------------------------------------------
SymbolicFrank · 2023-04-11 10:53

It's an AWS PostgreSQL (variant) server, it has no associated storage
you can access.

----------------------------------------------------------------------
SymbolicFrank · 2023-04-11 11:19

Ah, I found something:

Code: Text  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i note that
COPY .... FROM STDIN  doesn't actually read from stdin, it requires
the data to be passed through to it with a special API
(PQputCopyData() in libpq, or similar in other APIs).
This is defined in unit postgres3dyn. Let's see if I can get that to
work. The hardest part is probably the accessibility.

----------------------------------------------------------------------
SymbolicFrank · 2023-04-11 12:29

Ok, I think this should be it (prepare for some ugly casting):

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i uses
postgres3; var  export: TStringList;  MyConnection: TSQLConnector; 
buffer: PChar;  MyTable, WinDir, s: string;  res: Integer;begin  //
init and such  export := TStringList.Create; 
export.LoadFromFile(WinDir + MyTable + '.csv'); 
MyConnection.ExecuteDirect('copy ' + MyTable + ' from stdin with
(delimiter E''\t'', format csv, header);');  for s in export do 
begin    buffer := PChar(s);    res := 0;    while res = 0 do   
begin      res :=
PQputCopyData(@(TPGHandle(MyConnection.Handle).NativeConn), buffer,
Length(buffer));      if res = 0 then Sleep(100);    end;    if res =
-1 then    begin      buffer :=
PQerrorMessage(@(TPGHandle(MyConnection.Handle).NativeConn));     
raise Exception.Create(buffer);    end;  end;  res :=
PQputCopyEnd(@(TPGHandle(MyConnection.Handle).NativeConn), nil);  if
res <= 0 then  begin    buffer :=
PQerrorMessage(@(TPGHandle(MyConnection.Handle).NativeConn));    raise
Exception.Create(buffer);  end;  // free and suchend;
But that doesn't compile (error in program file): myimport.lpr(24,1)
Error: Import library not found for pq

?

Edit: uses postgres3dyn, duh. But it doesn't work. "Execution of query
failed." No other error.

----------------------------------------------------------------------
SymbolicFrank · 2023-04-17 10:21

This works:

Code: Pascal  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i unit
PostgresBulkInsert; {$mode ObjFPC}{$H+}{$COPERATORS
ON} interface uses  Classes, SysUtils, SQLDB, DB, Variants,
PQConnection, postgres3dyn; const  TranslateToUTF8 = False; type 
TSpecialFormatSettings = Class(TCustomExportFormatSettings)  public  
FieldDelimiter: Char;    HeaderRow: Boolean;    QuoteChar: Char; 
Published    Property HandleNullField;    Property IntegerFormat;   
Property DecimalSeparator;    Property CurrencySymbol;    Property
CurrencyDigits;    Property BooleanTrue;    Property BooleanFalse ;  
Property DateFormat;    Property TimeFormat;    Property
DateTimeFormat;  end; var  DefaultFormatSettings : TFormatSettings =
(    CurrencyFormat: 1;    NegCurrFormat: 5;    ThousandSeparator:
',';    DecimalSeparator: '.';    CurrencyDecimals: 2;   
DateSeparator: '-';    TimeSeparator: ':';    ListSeparator: ',';   
CurrencyString: '$';    ShortDateFormat: 'd/m/y';    LongDateFormat:
'dd" "mmmm" "yyyy';    TimeAMString: 'AM';    TimePMString: 'PM';   
ShortTimeFormat: 'hh:nn';    LongTimeFormat: 'hh:nn:ss';   
ShortMonthNames: ('Jan','Feb','Mar','Apr','May','Jun',               
     'Jul','Aug','Sep','Oct','Nov','Dec');    LongMonthNames:
('January','February','March','April','May','June',                  
 'July','August','September','October','November','December');   
ShortDayNames: ('Sun','Mon','Tue','Wed','Thu','Fri','Sat');   
LongDayNames:  ('Sunday','Monday','Tuesday','Wednesday','Thursday','Fr
iday','Saturday');    TwoDigitYearCenturyWindow: 50;  );  
PostgresCopyConnection: TPQConnection = nil; function FormatField(F:
TField; FormatSettings: TSpecialFormatSettings;  NullString: string;
ToUTF8: Boolean = TranslateToUTF8): UTF8string;function
PostgresCopy(ThisConnection: TSpecialConnector; ThisTable: string; 
ThisData: TDataset; CreateTable: Boolean = True):
string; implementation function FormatField(F: TField; FormatSettings:
TSpecialFormatSettings;  NullString: string; ToUTF8: Boolean):
UTF8string;const  StringFieldTypes =
[ftString,ftFixedChar,ftWidestring,ftFixedWideChar];  IntFieldTypes  
 = [ftInteger,ftWord,ftSmallint,ftAutoinc];  FloatFieldTypes  =
[ftFloat, ftCurrency, ftFMTBcd, ftBCD];  OrdFieldTypes    =
IntFieldTypes +[ftBoolean,ftLargeInt];  DateFieldTypes   =
[ftDate,ftTime,ftDateTime,ftTimeStamp];  MemoFieldTypes   =
[ftMemo,ftFmtMemo,ftWideMemo];  BlobFieldTypes   =
[ftBlob,ftDBaseOLE,ftGraphic,ftOraBlob,ftOraClob,ftParadoxOLE];var  FS
: TFormatSettings;  s: string;  q: TSQLQuery;begin  if F.IsNull and
FormatSettings.HandleNullField then Exit(NullString);  If (F.DataType
in IntFieldTypes) then    begin    If
((FormatSettings.IntegerFormat)<>'') and (not F.IsNull) then     
Result:=Format(FormatSettings.IntegerFormat,[F.AsInteger])    else if
FormatSettings.UseDisplayText then      Result:=F.DisplayText    else 
    Result:=F.AsUTF8String;    end  else if (F.DataType=ftBoolean)
then    begin    if (Not F.IsNull) then      If F.AsBoolean then     
 Result:=FormatSettings.BooleanTrue      else       
Result:=FormatSettings.BooleanFalse;    If (Result='') then      if
FormatSettings.UseDisplayText then        Result:=F.DisplayText     
else        Result:=F.AsUTF8String;    end  else if
(F.DataType=ftDate) then    begin    If
(FormatSettings.DateFormat<>'') and (not F.IsNull) then     
Result:=FormatDateTime(FormatSettings.DateFormat,F.AsDateTime)    else
if FormatSettings.UseDisplayText then      Result:=F.DisplayText   
else      Result:=F.AsUTF8String;    end  else if (F.DataType=ftTime)
then    begin    If (FormatSettings.TimeFormat<>'') and (not F.IsNull)
then     
Result:=FormatDateTime(FormatSettings.TimeFormat,F.AsDateTime)    else
if FormatSettings.UseDisplayText then      Result:=F.DisplayText   
else      Result:=F.AsUTF8String;    end  else if (F.DataType in
[ftDateTime,ftTimeStamp]) then    begin    If
(FormatSettings.DateTimeFormat<>'') and (not F.IsNull) then     
Result:=FormatDateTime(FormatSettings.DateTimeFormat,F.AsDateTime)   
else if FormatSettings.UseDisplayText then      Result:=F.DisplayText 
  else      Result:=F.AsUTF8String;    end  else if
(F.DataType=ftCurrency) then    begin    If
(FormatSettings.CurrencySymbol<>'') and (not F.IsNull) then     
begin      FS:=DefaultFormatSettings;     
FS.CurrencyString:=FormatSettings.CurrencySymbol;      Result:=CurrToS
trF(F.AsCurrency,ffCurrency,FormatSettings.CurrencyDigits,FS);     
end    else if FormatSettings.UseDisplayText then     
Result:=F.DisplayText    else      Result:=F.AsUTF8String;    end 
else if (F.DataType in BlobFieldTypes) then    Result := ''  else if
FormatSettings.UseDisplayText then    Result:=F.DisplayText  else   
Result:=F.AsUTF8String;   if ToUTF8 then Result :=
CP1252ToUTF8(Result);  Result := StringReplace(Result, '\', '\\',
[rfReplaceAll]);  Result := StringReplace(Result, #8, '\b',
[rfReplaceAll]);  Result := StringReplace(Result, #9, '\t',
[rfReplaceAll]);  Result := StringReplace(Result, #10, '\n',
[rfReplaceAll]);  Result := StringReplace(Result, #11, '\v',
[rfReplaceAll]);  Result := StringReplace(Result, #12, '\f',
[rfReplaceAll]);  Result := StringReplace(Result, #13, '\r',
[rfReplaceAll]);  Result := StringReplace(Result, #0, '',
[rfReplaceAll]);end; function PostgresCopy(ThisConnection:
TSQLConnector; ThisTable: string;  ThisData: TDataset; CreateTable:
Boolean): string;const  Delimiter = #9;var  MyConnection:
TPQConnection;  MyHandle: TPGHandle;  MyPGConn: TPGConn;  MyResult:
TPGresult;  buffer: PChar;  FormatSettings: TSpecialFormatSettings; 
F: TField;  query, buf: string;  FieldList: TStringList;  res:
Integer;  ExecStatus: TExecStatusType;begin  Result := 'Uninitialized
parameter found!';  if (not Assigned(ThisConnection)) or (not
Assigned(ThisData)) then Exit;  Result := 'Not a PostgreSQL database
connection!';  if ThisConnection.ConnectorType <> 'PostgreSQL' then
Exit;  ThisTable := LowerCase(Trim(ThisTable));  Result := 'Missing
table name!';  if Length(ThisTable) = 0 then Exit;   MyConnection :=
nil;  if Assigned(PostgresCopyConnection) then  begin    MyConnection
:= PostgresCopyConnection;    try      MyConnection.Open;    except  
  PostgresCopyConnection := nil;      FreeAndNil(MyConnection);   
end;  end;   if not Assigned(MyConnection) then  begin   
PostgresCopyConnection := TPQConnection.Create(nil);   
PostgresCopyConnection.HostName := ThisConnection.HostName;   
PostgresCopyConnection.DatabaseName := ThisConnection.DatabaseName;  
PostgresCopyConnection.UserName := ThisConnection.UserName;   
PostgresCopyConnection.Password := ThisConnection.Password;   
PostgresCopyConnection.Transaction :=
TSQLTransaction.Create(MyConnection);    MyConnection :=
PostgresCopyConnection;  end;  try    MyConnection.Close;   
MyConnection.Open;  except    MyConnection.Free;    Result := 'Unable
to open database connection!';    Exit;  end;   if CreateTable then 
begin    if not CreateSQLTable(ThisTable, ThisData, ThisConnection)
then    begin      Result := 'Could not create table ' + ThisTable +
'!';      Exit;    end;  end;   Result := '';  ThisData.First;  if
ThisData.EOF then Exit; // Nothing to do   MyHandle :=
TPGHandle(MyConnection.Handle);  MyPGConn := MyHandle.NativeConn^;  
FieldList := GetSQLFields(ThisTable, ThisConnection); 
FieldList.AlwaysQuote := True;  query := 'copy ' + ThisTable + '(' +
FieldList.DelimitedText +    ') from stdin with (delimiter E''\t'',
format text);';  MyResult := PQExec(@MyPGConn, PChar(query))^; 
ExecStatus := PQresultStatus(@MyResult);  if ExecStatus <>
PGRES_COPY_IN then  begin    FieldList.Free;    Result := 'Unable to
initialise COPY FROM STDIN!';    Exit;  end;   FormatSettings :=
TSpecialFormatSettings.Create(True);  FormatSettings.HandleNullField
:= True;  FormatSettings.FieldDelimiter := Delimiter; 
FormatSettings.HeaderRow := True;  FormatSettings.QuoteChar := '"'; 
FormatSettings.DateFormat := 'yyyy-mm-dd'; 
FormatSettings.DateTimeFormat := 'yyyy-mm-dd"T"hh:mm:ss'; 
FormatSettings.TimeFormat := 'hh:mm:ss';   try    while not
ThisData.EOF do    try      buf := '';      buffer := '';      // For
each record...      for F in ThisData.Fields do      begin        if
Length(buf) > 0 then buf += Delimiter;        buf += FormatField(F,
FormatSettings);      end;      buf += LineEnding;      buffer :=
PChar(buf);       // Send...      res := 0;      while res = 0 do    
begin        res := PQputCopyData(@MyPGConn, buffer, Length(buffer)); 
      if res = 0 then Sleep(100);      end;      if res <> 1 then //
Fatal error      begin        buffer := PQerrorMessage(@MyPGConn);   
   Result := buffer;        Exit;      end;       ThisData.Next;   
except      on e: Exception do      begin        Result := e.Message; 
      Exit;      end;    end;     Result := '';    res :=
PQputCopyEnd(@MyPGConn, nil);    if res <> 1 then    begin      buffer
:= PQerrorMessage(@MyPGConn);      Result := buffer;    end;   
MyConnection.Transaction.Commit;  finally    FieldList.Free;   
FormatSettings.Free;  end;end; end.
This is without the obvious helper functions, like CreateSQLTable and
GetSQLFields (and their helper functions), to keep it to the point.
The FormatField function is copied from TCustomDatasetExporter and
modified.

----------------------------------------------------------------------
SymbolicFrank · 2023-04-18 13:53

It's not stable. If I insert ~1000 tables, I really want to create a
new connection for each one of them, that is only used for the COPY
command. But connections that are closed and freed, are still active
connections until the application exits, as far as PostgreSQL is
concerned. Even if I set max_connections to 5000, I get this error:

Code: Text  [Select][+][-]window.onload = function(){var x1 =
document.getElementById("main_content_section"); if (x1) { var x =
document.getElementsByClassName("geshi");for (var i = 0; i Connection
to database failed (PostgreSQL: connection to server at "localhost"
(::1), port 5432 failed: FATAL:  sorry, too many clients already)
Manually resetting the database is not a workable solution.

Reusing a single connector isn't optimal either, as sometimes the old
COPY is still active when the new one comes in and packets get lost.

So, is there anything I could try, before I start making my own, low-
level database connection?