Оптимизация при работе с БД

[Arinyshka]

Суть проблемы в следующем. Delphi 7.0, SQL Server 2005.
В один из моментов работы программа считывает данные из нетипизированного файла. Считывает "строку", заполняет поля qruni:TAdoQuery и запихивает данные в БД. Запихнув очередную "строку", запоминает последнюю считанную позицию в файле. Файл (логи работы) может пополняться, чтобы не возиться с уже обработанным еще раз - следующее считывание начнется с сохраненной позиции.
Файл может содержать тысячи таких строк. вставка в БД происходит после каждой считанной "строки". Медленно!!! очень медленно. Обработка нескольких десятков файлов размером в мегабайты занимает часы...
Как можно оптимизировать процесс? понятно, что обращаться к БД лучше бы пореже. Как разумнее хранить временные данные? В чем их накопить перед вставкой в таблицы БД?

[Sla]

подробней

[Arinyshka]

Код:

//чтение данных из Log-файла

procedure TFormMainLogReader.ReadAllData(FileName: TFileName);

  //функция чтения данных переменной длины из двоичного файла
  function ReadStringToLogFile(const AFile: Cardinal; ASize: Integer): String;
    //Ограничение на длину поля, необходимо, если log-файл "битый".
    //В этом случае Size может принимать неправильные (большие) значения,
    //что приводит к ошибке Memory of out.
    const MaxLengthField = 255;

    var sTmp: String;
        Size: integer;
        BytesRead: Cardinal;
        Res: boolean;
        pc: PChar;
  begin
    Size := ASize;
    Result := '';
    Res := ReadFile(AFile, Size, SizeOf(Size), BytesRead, nil);
    Res := (Res and (BytesRead > 0));
    if Res then begin
      Size := Min(Size, MaxLengthField);
      SetLength(sTmp, Size);
      pc := PChar(sTmp);
      //Res := ReadFile(AFile, PChar(sTmp)^, Size, BytesRead, nil);
      Res := ReadFile(AFile, pc[0], Size, BytesRead, nil);
      Res := (Res and (BytesRead > 0));
      sTmp := pc;
      if Res then Result := sTmp;
    end;
  end;

const BeginRecord = #7#77;
var lrLog: TLogRecord;
    hFile: THANDLE;
    BytesRead: Cardinal;
    Rez: Boolean;
    //First: Boolean;
    iCurPos, iSizeFile: integer;
    Buff, UserCrypt: String;
begin
  Screen.Cursor := crSQLWait;
  Buff := (BeginRecord + Chr(C_STREAM_VERSION));
  Rez := True;
  BytesRead := 1;
  hFile := CreateFile(PChar(FileName), GENERIC_READ, 0, nil, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL or  FILE_FLAG_SEQUENTIAL_SCAN, 0);
  try
    if not qrUni.Connection.InTransaction then qrUni.Connection.BeginTrans;
    try
      iSizeFile := GetFileSize(hFile, nil);
      if iSizeFile = INVALID_SET_FILE_POINTER then Raise Exception.Create('Ошибка определения размера файла ' +FileName);
      qrTmp.Close;
      afSetCommandText(qrTmp, 'log_GetPosLog', [FileName], False);
      qrTmp.Open;
      iCurPos := qrTmp.FieldByName('Position').AsInteger;
      if iCurPos = null then
      begin
        iCurPos := SetFilePointer(hFile, 0, nil, FILE_BEGIN);
        if iCurPos = INVALID_SET_FILE_POINTER then Raise Exception.Create('Ошибка установки позиции указателя файла  ' +FileName);
      end
      else
      begin
        iCurPos := SetFilePointer(hFile, iCurPos, nil, FILE_BEGIN);
        if iCurPos = INVALID_SET_FILE_POINTER then Raise Exception.Create('Ошибка установки позиции указателя файла  ' +FileName);
      end;
      //First := True;
      while (Rez) and (BytesRead > 0) do
      begin
        //**********************************************************************
        //Начало записи или нет
        //**********************************************************************
        Rez := ReadFile(hFile, Buff[1], Length(Buff), BytesRead, nil);
        if not (Rez and (BytesRead > 0)) then Break;
        if Buff = (BeginRecord + Chr(C_STREAM_VERSION)) then
        begin
          lrLog.StreamVersion := 0;
          lrLog.PresentParams := 0;
          lrLog.DateTime := 0;
          lrLog.DateDoc := 0;
          lrLog.sIdRec := 0;
          lrLog.sDatabase := '';
          lrLog.sUser := '';
          lrLog.sTable := '';
          lrLog.sProc_Id := '';
          lrLog.sProc_Name := '';
          lrLog.sType_Event := '';
          lrLog.sType_Result := '';
          lrLog.sNote := '';
          lrLog.sNumDoc := '';

          //**********************************************************************
          //Читаем версию потока => C_STREAM_VERSION
          //**********************************************************************
         { Rez := ReadFile(hFile, lrLog.StreamVersion, SizeOf(byte), BytesRead, nil);
          if not (Rez and (BytesRead > 0)) then Break;
          if lrLog.StreamVersion = C_STREAM_VERSION then  }
          //**********************************************************************
           {raise Exception.Create('Неизвестная версия ' + Inttostr(lrLog.StreamVersion) +
            ' записи в LOG файле ' + FileName + ' ' + lrLog.sProc_Id + ' ' +floattostr(lrLog.DateTime) +
            #10#13 + 'Смещение: 0x' + IntToHex(SetFilePointer(hFile, 0, nil, FILE_CURRENT	),2)); }

          //**********************************************************************
          //Читаем количество параметров
          //**********************************************************************
          Rez := ReadFile(hFile, lrLog.PresentParams, SizeOf(lrLog.PresentParams), BytesRead, nil);
          if not (Rez and (BytesRead > 0)) then Break;

          //**********************************************************************
          //Читаем дату и время записи
          //**********************************************************************
          Rez := ReadFile(hFile, lrLog.DateTime, SizeOf(lrLog.DateTime), BytesRead, nil);
          if not (Rez and (BytesRead > 0)) then Break;
          (*if First then begin
            //ShowMes('Удаление старой информации за ' + DateToStr(lrLog.DateTime));
            afSetCommandText(qrTmp, 'log_delInfo',[lrLog.DateTime, lrLog.DateTime],False);
            qrTmp.ExecSQL;
            //ShowMes('Информация за ' + DateToStr(lrLog.DateTime) + ' удалена');
          end;*)

          //**********************************************************************
          //Читаем остальные данные
          //**********************************************************************
          if WordBool(lrLog.PresentParams and C_DATABASE) then
            lrLog.sDatabase := ReadStringToLogFile(hFile, 50);
          if WordBool(lrLog.PresentParams and C_USER) then
            lrLog.sUser := ReadStringToLogFile(hFile, 50);
          if WordBool(lrLog.PresentParams and C_TABLE) then
            lrLog.sTable := ReadStringToLogFile(hFile, 50);
          if WordBool(lrLog.PresentParams and C_PROC_ID) then
            lrLog.sProc_Id := ReadStringToLogFile(hFile, 36);
          if WordBool(lrLog.PresentParams and C_PROC_NAME) then
            lrLog.sProc_Name := ReadStringToLogFile(hFile, 100);
          if WordBool(lrLog.PresentParams and C_TYPE_EVENT) then
            lrLog.sType_Event := ReadStringToLogFile(hFile, 50);
          if WordBool(lrLog.PresentParams and C_TYPE_RESULT) then
            lrLog.sType_Result := ReadStringToLogFile(hFile, 50);
          if WordBool(lrLog.PresentParams and C_NOTE) then
            lrLog.sNote := ReadStringToLogFile(hFile, 250);
          if WordBool(lrLog.PresentParams and C_IDREC) then begin
            Rez := ReadFile(hFile, lrLog.sIdRec, SizeOf(lrLog.sIdRec), BytesRead, nil);
            if not (Rez and (BytesRead > 0)) then Break;
          end;
          if WordBool(lrLog.PresentParams and C_NUMDOC) then
            lrLog.sNumDoc := ReadStringToLogFile(hFile, 50);
          if WordBool(lrLog.PresentParams and C_DATEDOC) then begin
            Rez := ReadFile(hFile, lrLog.DateDoc, SizeOf(lrLog.DateDoc), BytesRead, nil);
            if not (Rez and (BytesRead > 0)) then Break;
          end;
 //требуется расшифровать параметры    lrLog.sDatabase,   lrLog.sUser
    //   showmessage (lrLog.sDatabase);
       lrLog.sDatabase := ClientManager.CryptString(lrLog.sDatabase) ;
       UserCrypt := ClientManager.CryptString(lrLog.sUser) ;
          qrTmp.Close;
          afSetCommandText(qrTmp, 'log_insInfo',[lrLog.DateTime,
                                                qrServersId_DataSource.Value,
                                                lrLog.sDatabase,
                                                UserCrypt,
                                                lrLog.sUser,
                                                lrLog.sTable,
                                                lrLog.sProc_Id,
                                                lrLog.sProc_Name,
                                                lrLog.sType_Event,
                                                lrLog.sType_Result,
                                                lrLog.sIdRec,
                                                lrLog.sNumDoc,
                                                lrLog.DateDoc,
                                                lrLog.sNote,
                                                ''],False);
          qrTmp.ExecSQL;
          //First := False;
          iCountRec := iCountRec + 1;
          sbStatus.Panels[3].Text := FloatToStrF(iCountRec,ffNumber,12,0);
        end;
        Application.ProcessMessages;
      end;
      //находим текущую позицию указателя
      iCurPos := SetFilePointer(hFile, 0, nil, FILE_CURRENT);
      if iCurPos = INVALID_SET_FILE_POINTER then Raise Exception.Create('Ошибка установки позиции указателя файла ' +FileName);
      qrTmp.Close;
      afSetCommandText(qrTmp, 'log_SavePosLog',[FileName, iCurPos],False);
      qrTmp.ExecSQL;
      if qrUni.Connection.InTransaction then qrUni.Connection.CommitTrans;
    except
      on E: Exception do
      begin
        if qrUni.Connection.InTransaction then qrUni.Connection.RollbackTrans;
        MessageDlg('Ошибка при обработке Log файла ' + FileName + #10#13 + 'Оригинал сообщения: ' +
          E.Message + #10#13 + 'Транзакция отменена, изменения в базу не внесены.', mtError, [mbOk], E.HelpContext);
      end;
    end;
  finally
    qrTmp.Close;
    CloseHandle(hFile);
    Screen.Cursor := crDefault;
  end;
end;

[Arinyshka]

уффф. Первое, что мне посоветовали - использовать потоки для чтения из файла. Теперь сижу, ковыряюсь в попытках понять. Утверждается, что  Filestream ReadStr совсем медленный. Что буфферизированное чтение будет многократно быстрее. Что можно использовать memorystream.
А можно пример - как организовать буферизированное чтение?  Попробую ускорить сам процесс чтения из файла. А потом уже думать, как возиться с записью данных в БД...

[Алексей++]

Медленно!!! очень медленно. Обработка нескольких десятков файлов размером в мегабайты занимает часы

из чего следует - неправильная работа с чем то, по любому Не могет так медленно...
И дело ТОЧНО не в чтении файла, несчастные мегабайты прочитаются быстро.

Нужно в отладчике позамерять время выполнения участков кода и определить, где тормоза.

Можно ещё поотключать отдельные блоки и посмотреть, как увеличивается скорость работы в разных случаях. Например, оставить только работу с файлом, без участия базы

[Sla]

Arinyshka, вопросик

ты не запуталась в RES и REZ?

[Arinyshka]

Ой, я запуталась уже везде, где только можно Процедурка (опять таки) ни разу не моя, она работает... но медленно.
При отключении записи в БД процесс ускорился раз в 15 То есть, если просто читать из файла - скорость работы вполне сносна. Значит, плохо именно то, что для каждой строки вызывается хранимая...
Хранимая богата подзапросами - возможно, это еще снижает скорость? Куда можно размещать записи так, чтобы вставлять их пореже...Например, пихать их в массив - и потом каждую сотню записей переносить в базу... Может быть, при запуске приложения загружать во временные массивы те таблицы, по которым идут подзапросы - и делать это на уровне дельфы? тогда в БД будет уходить сформированная строка, необходимость в подзапросах исчезнет...

Код:

CREATE PROCEDURE log_insInfo
  @DateTime datetime,
  @Id_DS smallint,
  @DataBase nvarchar(50),
  @UserCrypt nvarchar(50),
  @User nvarchar(50),
  @Table nvarchar(50),
  @Id_Proc nvarchar(36),
  @Proc nvarchar(100),
  @Event nvarchar(50),
  @Result nvarchar(50),
  @Id_Rec int,
  @NumDoc nvarchar(50),
  @DateDoc datetime,
  @Note nvarchar(250),
  @Comment nvarchar(250)
AS
  DECLARE
    @ID_DB int,
    @ID_User int,
    @Id_Table int,
    @Id_TypeEvent int,
    @Id_TypeResult int,
    @smallDateDoc smalldatetime

  IF CAST(@DateDoc AS float) <= 0 SET @DateDoc = NULL

  SET @DataBase = RTRIM(LTRIM(@DataBase))
  SET @User = RTRIM(LTRIM(@User))
  SET @Table = RTRIM(LTRIM(@Table))
  SET @Event = RTRIM(LTRIM(@Event))
  SET @Result = RTRIM(LTRIM(@Result))

  SET @id_DB = (SELECT TOP 1 DB.id_TB0003 id_db
              FROM Admin.dbo.TB0003 DB
             WHERE UPPER(DB.Alias_01) = UPPER(@DataBase))
             
             
  SET @id_User = (SELECT TOP 1 Usr.id_TB0026 id_user
                FROM Admin.dbo.TB0026 Usr
                WHERE UPPER(Usr.Alias_02) = UPPER(@UserCrypt))

  IF @id_User IS NULL
    SET @Comment = 'В таблице < Users > не найден пользователь < ' + @User + ' >'
               
  SET @id_Table = (SELECT TOP 1 Tbl.id_TB0019 Tbl
                 FROM Admin.dbo.TB0019 Tbl
                 WHERE UPPER(Tbl.Alias_01) = UPPER(@Table))

  IF @id_Table IS NULL
  BEGIN
    INSERT INTO Admin.dbo.TB0019(Alias_01) VALUES(@Table)
    SET @id_Table = SCOPE_IDENTITY()
  END
                
  SET @id_TypeEvent = (SELECT TOP 1 TLE.id_TB0017 Id_TypeEvent
                      FROM Admin.dbo.TB0017 TLE
                      WHERE UPPER(TLE.Alias_01) = UPPER(@Event))
                      
  IF @id_TypeEvent IS NULL
  BEGIN
    INSERT INTO Admin.dbo.TB0017(Alias_01) VALUES(@Event)
    SET @id_TypeEvent = SCOPE_IDENTITY()
  END
                     
  SET @id_TypeResult = (SELECT TOP 1 TLR.id_TB0018 Id_TypeResult
                      FROM Admin.dbo.TB0018 TLR
                      WHERE UPPER(TLR.Alias_01) = UPPER(@Result))
                      
  IF @id_TypeResult IS NULL
  BEGIN
    INSERT INTO Admin.dbo.TB0018(Alias_01) VALUES(@Result)
    SET @id_TypeResult = SCOPE_IDENTITY()
  END

--  BEGIN TRAN
  INSERT INTO Admin.dbo.TB0008(Alias_01,
                               id_TB0004,
                               id_TB0003,
                               id_TB0026,
                               id_TB0019,
                               Alias_02,
                               Alias_03,
                               id_TB0017,
                               id_TB0018,
                               Alias_04,
                               Alias_05,
                               Alias_06,
                               Alias_07,
                               Alias_08)
  VALUES(@DateTime,
         @Id_DS,
         @id_DB,
         @id_User,
         @id_Table,
         @Id_Proc,
         @Proc,
         @id_TypeEvent,
         @id_TypeResult,
         @Id_Rec,
         @NumDoc,
         @DateDoc,
         @Note,
         @Comment)

[Arinyshka]

в общем-то, создаю теперь временную таблицу. Вписываю в нее построчно все эти данные. И остается вопрос - а как ее оптимально скинуть в БД? Пока есть только одна идея - формировать SQL-запрос из строк типа

Код:

INSERT INTO dbo.T1 (column_2, column_4) 
    VALUES ('Explicit value', 'Explicit value');

Формировать его циклом, INSERT  повторить требуемое количество раз. А вызывать exec один раз. Вопросов 1001 в реализации всего этого... начиная с того, как же я параметры передам - 1 строка это 15 параметров для  INSERT.
Но хоть скорость-то я повышу в результате адских мук? Не бросайте меня, пожалуйста... совсем я умучалась с этим Логридером

[Алексей++]

Arinyshka, ты пробовала сделать, что я написал ? Попробуй хотя бы. Убери строку, где производится запись в базу, пусть всё отработает "вхолостую" - с какой скоростью всё выполнится ?

[Sla]

Алексей1153++, читай по слогам
При отключении записи в БД процесс ускорился раз в 15

[Arinyshka]

Arinyshka, ты пробовала сделать, что я написал ? Попробуй хотя бы. Убери строку, где производится запись в базу, пусть всё отработает "вхолостую" - с какой скоростью всё выполнится ?

Конечно, пробовала. Скорость возросла многократно! Поэтому и вывод такой - нужно все логи собирать во временную таблицу и инсертить ее в БД как можно реже. Не для каждой записи каждого файла, как сейчас... а, например. один раз в конце обработки каждого файла.
Вот мучаюсь мыслью - как инсертнуть таблицу целиком...

[Sla]

Arinyshka, я все же думаю, что это связано с lock'ами таблиц, как при чтении, так и при записи.
Я тебе уже давал пример как делать нелокирумое чтение.

[Sla]

кроме того

Код:

SET @DataBase = RTRIM(LTRIM(@DataBase))
  SET @User = RTRIM(LTRIM(@User))
  SET @Table = RTRIM(LTRIM(@Table))
  SET @Event = RTRIM(LTRIM(@Event))
  SET @Result = RTRIM(LTRIM(@Result))

Не плохой стиль, но например в твоем случае, видимо эту часть, желательно делать на клиенте.

[Sla]

или же

Код:

SET @DataBase = UPPER(RTRIM(LTRIM(@DataBase)))
  SET @User = UPPER(RTRIM(LTRIM(@User)))
  SET @Table = UPPER(RTRIM(LTRIM(@Table)))
  SET @Event = UPPER(RTRIM(LTRIM(@Event)))
  SET @Result = UPPER(RTRIM(LTRIM(@Result)))

т.е. я не совсем понимаю, необходимость делать upper, потом сравнивать, потом инсертить оригинальный, затем опять upper чтобы сравнить.

[Arinyshka]

Arinyshka, я все же думаю, что это связано с lock'ами таблиц, как при чтении, так и при записи.
Я тебе уже давал пример как делать нелокирумое чтение.

эээ... есть вопрос...

NOLOCK
Блокировки не используются. Этот параметр  не применяется к инструкциям INSERT, UPDATE, и DELETE.

Чего-то я не понимаю... на Инсерт его использовать нельзя вроде... Или можно?
На чтение поставила сразу же. Форматирования поправлю, это даже мне несложно

[Sla]

Блокировки не используются. Этот параметр  не применяется к инструкциям INSERT, UPDATE, и DELETE.

Естественно, на этот момент таблица лочится

[Алексей++]

ой, сорри, невнимательно прочитал про 15 раз ))

[Алексей++]

Arinyshka, какие нибудь триггеры посажены на insert, update ?

[Arinyshka]

Нет, триггеров никаких не посажено.

[Sla]

как вариант инсертить во временную таблицу

а потом из временной переносить в постоянную
при этом очищая временную

INSERT INTO Admin.dbo.TB0008
    SELECT *
        FROM Temp1;
DELETE TEMP1;

[Arinyshka]

как вариант инсертить во временную таблицу

а потом из временной переносить в постоянную
при этом очищая временную

INSERT INTO Admin.dbo.TB0008
    SELECT *
        FROM Temp1;
DELETE TEMP1;

Да, вариант понятен. А можно все-таки скинуть за раз, за один exec таблицу? Как скидывать ее по строчкам циклом из дельфы, мне понятно. Но это не ускорит процесс, так ведь?
Формировать тело Sql-запроса внутри query - бежать циклом по таблице, формировать на каждую строку Insert... выполнять одним махом во временную таблицу. Так можно? Потом разбирать ее со всеми моими подзапросами вида

Код:

SET @id_Table = (SELECT TOP 1 Tbl.id_TB0019 Tbl
                 FROM Admin.dbo.TB0019 Tbl
                 WHERE UPPER(Tbl.Alias_01) = UPPER(@Table))

А зачем очищать временную построчно? Как вообще правильнее - если я скину в нее млн записей и разберу за один раз. или по 1000, или по 1 разбирать по порциям, очищая?

[PooH]

первое что нашлось:
http://www.askit.ru/custom/sql2005_admin/m10/10_08_bulk_insert_task.htm
а также:
http://blog.sqlauthority.com/2007/06/08/sql-server-insert-multiple-records-using-one-insert-statement-use-of-union-all/

про эффективность нечиго не могу сказать - я с SQL Server-ом не общаюсь... но идея думаю понятна.

[PooH]

во второй ссылке в кооментах есть:
 INSERT INTO YourTable (FirstCol, SecondCol)
VALUES (’First’ , 1) , (’Second’ , 2) , (’Third’ , ‘3′), (’Fourth’ , ‘4′) (’and so on’) ;

а также:
BULK INSERT Table_1 FROM ‘D:\Test\t2.txt’ WITH (ROWTERMINATOR = ‘\n’)

[PooH]

немного поискав нашел еще: http://sql.ru/forum/actualthread.aspx?tid=559728&hl=%e2%f1%f2%e0%e2%e8%f2%fc+%e7%e0%ef%e8%f1%e5%e9
там помимо bcp еще говорится:

Для быстрой вставки большого количества данных уже давно придумана технология BULK INSERT. Есть клиентская поддержка через оледибишный интерфейс.