[SPARK-44885][SQL] NullPointerException is thrown when column with RO…

…WID type contains NULL values

### What changes were proposed in this pull request?
If a `rowid` column is `null`, do not call `toString` on it.

### Why are the changes needed?
A column with the `rowid` type may contain NULL values.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Test cases.

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes apache#42576 from tindzk/fix/rowid-null.

Authored-by: Tim Nieradzik <tim@sparse.tech>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/OracleIntegrationSuite.scala

@@ -49,7 +49,7 @@ import org.apache.spark.tags.DockerTest
  * 4. Start docker: sudo service docker start
  *    - Optionally, docker pull $ORACLE_DOCKER_IMAGE_NAME
  * 5. Run Spark integration tests for Oracle with: ./build/sbt -Pdocker-integration-tests
- *    "testOnly org.apache.spark.sql.jdbc.OracleIntegrationSuite"
+ *    "docker-integration-tests/testOnly org.apache.spark.sql.jdbc.OracleIntegrationSuite"
  *
  * A sequence of commands to build the Oracle XE database container image:
  *  $ git clone https://github.com/oracle/docker-images.git
@@ -521,4 +521,19 @@ class OracleIntegrationSuite extends DockerJDBCIntegrationSuite with SharedSpark
     assert(types(0).equals("class java.lang.String"))
     assert(!rows(0).getString(0).isEmpty)
   }
+
+  test("SPARK-44885: query row with ROWID type containing NULL value") {
+    val rows = spark.read.format("jdbc")
+      .option("url", jdbcUrl)
+      // Rename column to `row_id` to prevent the following SQL error:
+      //   ORA-01446: cannot select ROWID from view with DISTINCT, GROUP BY, etc.
+      // See also https://stackoverflow.com/a/42632686/13300239
+      .option("query", "SELECT rowid as row_id from datetime where d = {d '1991-11-09'}\n" +
+        "union all\n" +
+        "select null from dual")
+      .load()
+      .collect()
+    assert(rows(0).getString(0).nonEmpty)
+    assert(rows(1).getString(0) == null)
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -453,7 +453,12 @@ object JdbcUtils extends Logging with SQLConfHelper {
 
     case StringType if metadata.contains("rowid") =>
       (rs: ResultSet, row: InternalRow, pos: Int) =>
-        row.update(pos, UTF8String.fromString(rs.getRowId(pos + 1).toString))
+        val rawRowId = rs.getRowId(pos + 1)
+        if (rawRowId == null) {
+          row.update(pos, null)
+        } else {
+          row.update(pos, UTF8String.fromString(rawRowId.toString))
+        }
 
     case StringType =>
       (rs: ResultSet, row: InternalRow, pos: Int) =>