Fix JDBC+Oservation test

apache · Jan 23, 2024 · 10fd22b · 10fd22b
1 parent 60e41ca
commit 10fd22b
Showing 1 changed file with 5 additions and 6 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
@@ -968,9 +968,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
       case Some(n) if n < df.rdd.getNumPartitions => df.coalesce(n)
       case _ => df
     }
-    repartitionedDF.rdd.foreachPartition { iterator =>
-      savePartition(None, iterator, rddSchema, insertStmt, batchSize, dialect,
-        isolationLevel, options)
+    repartitionedDF.foreachPartition { iterator: Iterator[Row] => savePartition(
+      None, iterator, rddSchema, insertStmt, batchSize, dialect, isolationLevel, options)
     }
   }
 
@@ -1008,9 +1007,9 @@ object JdbcUtils extends Logging with SQLConfHelper {
       case Some(n) if n < df.rdd.getNumPartitions => df.coalesce(n)
       case _ => df
     }
-    repartitionedDF.rdd.foreachPartition { iterator => upsertPartition(
-      table, iterator, rddSchema, tableSchema, isCaseSensitive, batchSize,
-      dialectWithMerge, isolationLevel, options)
+    repartitionedDF.foreachPartition { iterator: Iterator[Row] => upsertPartition(
+      table, iterator, rddSchema, tableSchema, isCaseSensitive, batchSize, dialectWithMerge,
+      isolationLevel, options)
     }
   }