LOAD DATA LOCAL INFILE with named pipes

[zeltron73]

Hello,

I would like to execute a command like this :

Code: Select all

LOAD DATA LOCAL INFILE "//localhost/pipe/pipename" INTO TABLE test FIELDS TERMINATED BY ";" ENCLOSED BY '"' LINES TERMINATED BY "\r\n";

Unfortunately, this does not work with MyDAC, because the code will ask the file size and the operating system will return 0 for a pipe... So the file is immediately closed and the server will return an error code.

This command works like a charm with the "mysql.exe" console program.

Could you modify the code so that MyDAC components could execute "LOAD DATA LOCAL INFILE" with pipes ?

Info: in the file MySqlSession.pas we can find the method TMySqlSession.ReadQueryResult
and the following code:

Code: Select all

c := f.Size;  // will return 0 for named pipes

while c > 0 do begin
  r := Min(c, Length(buf));
  f.ReadBuffer(buf[0], r);
  net.WriteBytes(buf, 0, r);
  net.Send;
  Dec(c, r);
end;

I think this could be replaced by :

Code: Select all

c := f.ReadBuffer(buf[0], Length(buf));
while c > 0 do begin
  net.WriteBytes(buf, 0, c);
  net.Send;
  c := f.ReadBuffer(buf[0], Length(buf));
end;

Regards.

[zeltron73]

Sorry,

In my previous post, the last piece of code should be replaced by:

Code: Select all

c := f.Read(buf[0], Length(buf));  // Read and not ReadBuffer !!!
while c > 0 do begin
  net.WriteBytes(buf, 0, c);
  net.Send;
  c := f.Read(buf[0], Length(buf));
end;

Regard.

[ViktorV]

Thank you for the information and your help on issue investigation. The issue investigation is in progress. We will inform you when we have any results.

[zeltron73]

Here is a little test to reproduce the problem...

Code: Select all

CREATE TABLE `table1` (
  `Field1` int(10) unsigned NOT NULL,
  `Field2` varchar(45) NOT NULL DEFAULT '""',
  PRIMARY KEY (`Field1`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

Code: Select all

procedure TForm1.Button1Click(Sender: TObject);
const
  BUF_SIZE = $10000;
  LOAD_DATA_QUERY = 'LOAD DATA LOCAL INFILE "%0:s" IGNORE INTO TABLE `%1:s`.`%2:s` ' +
    'FIELDS TERMINATED BY ";" OPTIONALLY ENCLOSED BY "\"" LINES TERMINATED BY "\r\n";';
var
  pipeName: string;
  outputStream: TFileStream;
  mysqlThread: TThread;
  pipeConnectionEvent: TSimpleEvent;
  overlapped: TOverlapped;
  line: AnsiString;
begin
  outputStream := nil;
  mysqlThread := nil;
  FillChar(overlapped, SizeOf(overlapped), 0);
  pipeConnectionEvent := nil;
  try
    pipeName := '\\.\pipe\mysqlTest';
    outputStream := TFileStream.Create(CreateNamedPipe(PChar(pipeName),
      PIPE_ACCESS_OUTBOUND or FILE_FLAG_OVERLAPPED, PIPE_TYPE_BYTE, 1, BUF_SIZE, 0, 100, nil));

    pipeName := StringReplace(pipeName, '\', '/', [rfReplaceAll]);
    pipeName := StringReplace(pipeName, '/./', '/localhost/', [rfIgnoreCase]);
    mysqlThread := TThread.CreateAnonymousThread(
      procedure
      var
        connection: TMyConnection;
        query: TMyQuery;
      begin
        try
          connection := nil;
          query := nil;
          try
            connection := TMyConnection.Create(nil);
            connection.Server := 'localhost';
            connection.Port := 3306;
            connection.Username := 'user';
            connection.Password := 'password';
            connection.Database := 'test';

            query := TMyQuery.Create(nil);
            query.Connection := connection;
            query.SQL.Text := string.Format(LOAD_DATA_QUERY, [pipeName, connection.Database, 'table1']);
            query.ExecSQL;
          finally
            query.Free;
            connection.Free;
          end;
        except
          on E: Exception do OutputDebugString(PChar(E.Message));
        end;
      end);

    pipeConnectionEvent := TSimpleEvent.Create;
    overlapped.hEvent := pipeConnectionEvent.Handle;
    if ConnectNamedPipe(outputStream.Handle, @overlapped) or
       (GetLastError = ERROR_IO_PENDING) or (GetLastError = ERROR_PIPE_CONNECTED) then
    begin
      mysqlThread.FreeOnTerminate := False;
      mysqlThread.Start;
      if (pipeConnectionEvent.WaitFor(5000) <> wrSignaled) then Exit;

      line := '"1";"String1"'#13#10;
      outputStream.Write(Pointer(line)^, Length(line));
      TThread.Sleep(5000);
      line := '"2";"String2"'#13#10;
      outputStream.Write(Pointer(line)^, Length(line));
      line := '"3";"String3"'#13#10;
      outputStream.Write(Pointer(line)^, Length(line));
      line := '"4";"String4"'#13#10;
      outputStream.Write(Pointer(line)^, Length(line));
      line := '"5";"String5"'#13#10;
      outputStream.Write(Pointer(line)^, Length(line));
    end;
  finally
    pipeConnectionEvent.Free;
    outputStream.Free;  // close the pipe
    mysqlThread.Free;   // wait for the thread
  end;
end;

When you execute the test, the first line only is inserted because when the sql command is executed only that line was written... The others are ignored because MyDAC has still closed the pipe !

I hope that it will help you.

Regards.

[ViktorV]

Thank you for being interested in our products and your help on issue investigation.
We will investigate the behavior of MyDAC according to your description and inform you about the results.