[GitHub] [incubator-streampark] wolfboys commented on a diff in pull request #2919: [Improvement] avoid errors in batch insert caused by different tables or columns

["wolfboys (via GitHub)" <gi...@apache.org>]

wolfboys commented on code in PR #2919:
URL: https://github.com/apache/incubator-streampark/pull/2919#discussion_r1284964660


##########
streampark-flink/streampark-flink-connector/streampark-flink-connector-base/src/main/scala/org/apache/streampark/flink/connector/failover/SinkRequest.scala:
##########
@@ -34,22 +34,24 @@ case class SinkRequest(records: util.List[String], var attemptCounter: Int = 0)
     Pattern.compile("^(.*)\\s+(values|value)(.*)", Pattern.CASE_INSENSITIVE)
 
   lazy val sqlStatement: String = {
-    val matcher = INSERT_REGEXP.matcher(records.head)
-    if (!matcher.find()) null;
-    else {
-      val prefix = matcher.group(1)
-      val values = records
-        .map(
-          x => {
-            val valueMatcher = INSERT_REGEXP.matcher(x)
-            if (valueMatcher.find()) {
-              valueMatcher.group(3)
-            } else {
-              null
-            }
-          })
-        .mkString(",")
-      s"$prefix VALUES $values"
+    val prefixMap: Map[String, List[String]] = Map[String, List[String]]()
+    records.foreach(
+      x => {
+        val valueMatcher = INSERT_REGEXP.matcher(x)
+        if (valueMatcher.find()) {
+          val prefix = valueMatcher.group(1)
+          prefixMap.get(prefix) match {
+            case Some(value) => value.add(valueMatcher.group(3))
+            case None => prefixMap.put(prefix, List(valueMatcher.group(3)))
+          }
+        } else {
+          println(s"[WARN] ignore record: $x")

Review Comment:
   Thanks for your contribution, recommend using [logWarn("...")](https://github.com/apache/incubator-streampark/blob/dev/streampark-flink/streampark-flink-connector/streampark-flink-connector-base/src/main/scala/org/apache/streampark/flink/connector/failover/FailoverWriter.scala#L111) instead of println



-- 
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: issues-unsubscribe@streampark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org