배치 삽입 및 LAST_INSERT_슬리크와 마리아의 아이디DB

배치 삽입 및 LAST_INSERT_슬리크와 마리아의 아이디DB

MariaDB 데이터베이스에 데이터를 삽입하려고 합니다.두 개의 테이블이 있는데 첫 번째 테이블에 (배치 삽입을 사용하여) 행을 삽입하고 새로 삽입된 행의 ID를 사용하여 두 번째 테이블에 두 번째 배치 삽입을 수행해야 합니다.

저는 스칼라에서 알파카 슬리크를 이용해서 그렇게 하고 있습니다.이 질문의 목적을 위해, 전화를 걸겠습니다.tests메인 테이블과dependent두 번째 것

현재 제 알고리즘은 다음과 같습니다.

1. 행을 삽입합니다.tests
2. 다음을 사용하여 배치에서 첫 번째 행의 ID 가져오기SELECT LAST_INSERT_ID();
3. 첫 번째 행의 ID와 배치의 행 수를 알고 다른 ID를 수작업으로 계산하여 두 번째 표에 삽입할 때 사용합니다.

이것은 한 번에 하나의 연결에서만 잘 작동합니다.하지만 동시에 쓰기를 여러 번 시도하여 시나리오를 시뮬레이션하려고 합니다.그러기 위해 스칼라 병렬 컬렉션과 아카 스트림을 사용하고 있습니다.Source다음과 같이:

// three sources of 10 random Strings each
val sources = Seq.fill(3)(Source(Seq.fill(10)(Random.alphanumeric.take(3).mkString))).zipWithIndex
val parallelSources: ParSeq[(Source[String, NotUsed], Int)] = sources.par

parallelSources.map { case (source, i) =>
  source
    .grouped(ChunkSize) // performs batch inserts of a given size
    .via(insert(i))
    .zipWithIndex
    .runWith(Sink.foreach { case (_, chunkIndex) => println(s"Chunk $chunkIndex of source $i  done") })
}

각 항목에 색인을 추가합니다.Source제가 DB에 쓰는 데이터에 접두사를 사용하기 위해서입니다.

다음은 코드입니다.insert Flow나는 지금까지 다음과 같이 썼습니다.

def insert(srcIndex: Int): Flow[Seq[String], Unit, NotUsed] = {
  implicit val insertSession: SlickSession = slickSession
  system.registerOnTermination(() => insertSession.close())

  Flow[Seq[String]]
    .via(Slick.flowWithPassThrough { chunk =>
      (for {
        // insert data into `tests`
        _ <- InsTests ++= chunk.map(v => TestProj(s"source$srcIndex-$v"))
        // fetch last insert ID and connection ID
        queryResult <- sql"SELECT CONNECTION_ID(), LAST_INSERT_ID();".as[(Long, Long)].headOption
        _ <- queryResult match {
          case Some((connId, firstIdInChunk)) =>
            println(s"Source $srcIndex, last insert ID $firstIdInChunk, connection $connId")
            // compute IDs by hand and write to `dependent`
            val depValues = Seq.fill(ChunkSize)(s"source$srcIndex-${Random.alphanumeric.take(6).mkString}")
            val depRows =
              (firstIdInChunk to (firstIdInChunk + ChunkSize))
                .zip(depValues)
                .map { case (index, value) => DependentProj(index, value) }

            InsDependent ++= depRows
          case None => DBIO.failed(new Exception("..."))
        }
      } yield ()).transactionally
  })
}

어디에InsTests그리고.InsDependentSlick's입니다.TableQuery물건들.slickSession서로 다른 각 삽입에 대해 새 세션을 생성하며 다음과 같이 정의됩니다.

private def slickSession = {
  val db = Database.forURL(
    url = "jdbc:mariadb://localhost:3306/test",
    user = "root",
    password = "password",
    executor = AsyncExecutor(
      name = "executor",
      minThreads = 20,
      maxThreads = 20,
      queueSize = 1000,
      maxConnections = 20
    )
  )
  val profile = slick.jdbc.MySQLProfile
  SlickSession.forDbAndProfile(db, profile)
}

문제는 알고리즘의 두 번째 단계에서 반환된 마지막 삽입 ID가 겹친다는 것입니다.이 앱을 실행할 때마다 다음과 같은 내용이 인쇄됩니다.

Source 2, last insert ID 6, connection 66
Source 1, last insert ID 5, connection 68
Source 0, last insert ID 7, connection 67
Chunk 0 of source 0  done
Chunk 0 of source 2  done
Chunk 0 of source 1  done
Source 2, last insert ID 40, connection 70
Source 0, last insert ID 26, connection 69
Source 1, last insert ID 27, connection 71
Chunk 1 of source 2  done
Chunk 1 of source 1  done
Chunk 1 of source 0  done

연결이 각각 다른 것처럼 보이는 경우Source하지만 ID가 겹칩니다(소스 0 참조).7소스 1 보기5소스 2 보기2). ID는 에서 시작하는 것이 맞습니다.5테이블을 만든 직후에 더미 행 4개를 추가하기 때문입니다(이 질문의 코드에는 표시되지 않음).분명히, 여러 행이 보입니다.dependent마찬가지로tests.id그런 일이 있어서는 안 됩니다.

마지막 삽입 ID는 단일 연결을 의미하는 것으로 알고 있습니다.전체 흐름이 트랜잭션(Slick's를 통해)으로 포장된다는 점을 고려할 때 서로 다른 세 개의 연결이 중복되는 ID를 보는 것이 어떻게 가능합니까?transactionally)?

이는 에서 발생합니다.innodb_autoinc_lock_mode=1제가 지금까지 본 바로는, 그것은 그렇지 않습니다.innodb_autoinc_lock_mode=0그것은 말이 됩니다, InnoDB는 잠글 것이기 때문입니다.tests전체 배치 삽입이 종료될 때까지.

Georg의 답변 후 업데이트:프로젝트의 다른 제약 조건에 대해서는 MariaDB 10.4와 호환되는 솔루션을 원합니다. 제가 이해하기로는, 이 솔루션은INSERT...RETURNING의 Slick의 Slick's.++=에 대한 운영자의 returning여기서 보고된 것처럼 상당히 안 좋습니다.MariaDB 10.4와 10.5 모두에서 테스트를 했는데, 쿼리 로그에 따르면 Slick은 싱글을 실행합니다.INSERT INTO일괄적인 것이 아닌 진술.저의 경우, 여러 행을 스트리밍 방식으로 작성할 예정이기 때문에 이는 그다지 받아들일 수 없습니다.

나는 또한 자동 증가 값에 대한 가정을 하는 것을 이해한다.1이는 이상적이지 않습니다. 프로덕션 설정을 제어할 수 있으며 다중 마스터 복제가 없습니다.

LAST_INSERT_ID()를 기준으로 다음 값을 생성할 수 없습니다.

1. 동시에 롤백된 두 번째 트랜잭션이 있을 수 있으므로 auto_incremented ID에 공백이 있을 수 있습니다.

2. LAST_INSERT_를 증분하여 행 수에 걸쳐 반복ID 값은 세션 변수 @@auto_increment_increment(특히 1이 아닌 다중 마스터 복제)의 값에 따라 달라지기 때문에 작동하지 않습니다.

대신 RETURNING을 사용하여 삽입된 행의 ID를 가져와야 합니다.

MariaDB [test]> create table t1 (a int not null auto_increment primary key);
Query OK, 0 rows affected (0,022 sec)

MariaDB [test]> insert into t1 (a) values (1),(3),(NULL), (NULL) returning a;
+---+
| a |
+---+
| 1 |
| 3 |
| 4 |
| 5 |
+---+
4 rows in set (0,006 sec)

언급URL : https://stackoverflow.com/questions/73652590/batch-inserts-and-last-insert-id-with-slick-and-mariadb