Fix JDBC+Oservation test

G-Research · Apr 20, 2024 · e3ade6e · e3ade6e
1 parent 91f79a6
commit e3ade6e
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
@@ -1045,9 +1045,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)
     }
   }
 
@@ -1085,9 +1084,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)
     }
   }