Copilot commented on code in PR #495:
URL: https://github.com/apache/pekko-projection/pull/495#discussion_r3298637316
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mysql/MySQLR2dbcOffsetStore.scala:
##########
@@ -63,4 +68,69 @@ private[projection] class MySQLR2dbcOffsetStore(
ON DUPLICATE KEY UPDATE
paused = excluded.paused,
last_updated = excluded.last_updated"""
+
+ /**
+ * MySQL's r2dbc driver validates that all parameters are bound before
`add()` is called
+ * on a batch statement, unlike PostgreSQL's driver. We therefore bind the
first record
+ * before folding over the remaining records with `add()`.
+ */
+ override protected def insertTimestampOffsetInTx(
+ conn: Connection,
+ records: immutable.IndexedSeq[R2dbcOffsetStore.Record]): Future[Long] = {
+ require(records.nonEmpty)
+
+ logger.trace("saving timestamp offset [{}], {}", records.last.timestamp,
records)
+
+ val statement = conn.createStatement(insertTimestampOffsetSql)
+
+ if (records.size == 1) {
+ val boundStatement = bindTimestampOffsetRecord(statement, records.head)
+ R2dbcExecutor.updateOneInTx(boundStatement)
+ } else {
+ // Bind the first record before calling add() for the rest; MySQL
validates all parameters
+ // are bound on the current batch row before accepting add().
+ val boundStatement =
+ records.tail.foldLeft(bindTimestampOffsetRecord(statement,
records.head)) { (stmt, rec) =>
+ stmt.add()
+ bindTimestampOffsetRecord(stmt, rec)
+ }
+ R2dbcExecutor.updateBatchInTx(boundStatement)
+ }
+ }
+
+ /**
+ * MySQL does not support `= ANY (?)` with an array parameter or the `||`
string concatenation
+ * operator. This override builds the DELETE SQL dynamically using
`CONCAT()` and
+ * `NOT IN (?, ?, ...)` with one placeholder per exclusion entry.
+ */
+ override protected def executeDeleteOldTimestampOffsets(
+ minSlice: Int,
+ maxSlice: Int,
+ until: Instant,
+ notInLatestBySlice: Array[String]): Future[Long] = {
+ r2dbcExecutor.updateOne("delete old timestamp offset") { conn =>
+ val stmt = if (notInLatestBySlice.isEmpty) {
+ conn
+ .createStatement(
+ sql"DELETE FROM $timestampOffsetTable WHERE slice BETWEEN ? AND ?
AND projection_name = ? AND timestamp_offset < ?")
+ .bind(0, minSlice)
+ .bind(1, maxSlice)
+ .bind(2, projectionId.name)
+ .bind(3, until)
+ } else {
+ val placeholders = notInLatestBySlice.map(_ => "?").mkString(", ")
+ val s = conn
+ .createStatement(
+ sql"DELETE FROM $timestampOffsetTable WHERE slice BETWEEN ? AND ?
AND projection_name = ? AND timestamp_offset < ? AND NOT CONCAT(persistence_id,
'-', seq_nr) IN ($placeholders)")
Review Comment:
The generated DELETE SQL uses `AND NOT CONCAT(persistence_id, '-', seq_nr)
IN (...)`, which is not valid MySQL syntax. MySQL expects `expr NOT IN (...)`
(or `NOT (expr IN (...))`). This will cause the cleanup query to fail at
runtime when `notInLatestBySlice` is non-empty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
