[SPARK-57778][SQL] Handle Oracle JDBC objects that are not selectable as tables

common/utils/src/main/resources/error/error-conditions.json

@@ -5292,6 +5292,12 @@
     },
     "sqlState" : "42000"
   },
+  "JDBC_OBJECT_NOT_SELECTABLE" : {
+    "message" : [
+      "The JDBC object <objectName> cannot be read as a table or view."
+    ],
+    "sqlState" : "42000"
+  },
   "JOIN_CONDITION_IS_NOT_BOOLEAN_TYPE" : {
     "message" : [
       "The join condition <joinCondition> has the invalid type <conditionType>, expected \"BOOLEAN\"."

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

@@ -20,9 +20,12 @@ package org.apache.spark.sql.jdbc.v2
 import java.sql.Connection
 import java.util.Locale
 
+import scala.util.Using
+
 import org.apache.spark.{SparkConf, SparkRuntimeException}
 import org.apache.spark.sql.{AnalysisException, Row}
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils.CHAR_VARCHAR_TYPE_STRING_METADATA_KEY
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
 import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
 import org.apache.spark.sql.jdbc.OracleDatabaseOnDocker
 import org.apache.spark.sql.types._
@@ -203,6 +206,60 @@ class OracleIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCTes
     }
   }
 
+  // An object that Oracle's table listing surfaces but that cannot be read as a table.
+  // `setup`/`teardown` are the DDL to create/drop it; `objectName` is its name in SYSTEM.
+  case class NonSelectableObjectCase(setup: Seq[String], objectName: String, teardown: Seq[String])
+
+  private val nonSelectableObjectCases = Map(
+    "synonym to a procedure (ORA-04044)" -> NonSelectableObjectCase(
+      setup = Seq(
+        "CREATE PROCEDURE test_proc AS BEGIN NULL; END;",
+        "CREATE SYNONYM proc_synonym FOR test_proc"),
+      objectName = "PROC_SYNONYM",
+      teardown = Seq("DROP SYNONYM proc_synonym", "DROP PROCEDURE test_proc")),
+    "synonym to a broken function (ORA-04044)" -> NonSelectableObjectCase(
+      // Function referencing a missing object -> created INVALID.
+      setup = Seq(
+        """CREATE FUNCTION broken_func RETURN NUMBER AS
+          |  x NUMBER;
+          |BEGIN
+          |  SELECT col INTO x FROM non_existent_table_xyz;
+          |  RETURN x;
+          |END;""".stripMargin,
+        "CREATE SYNONYM func_synonym FOR broken_func"),
+      objectName = "FUNC_SYNONYM",
+      teardown = Seq("DROP SYNONYM func_synonym", "DROP FUNCTION broken_func")),
+    "invalid view (ORA-04063)" -> NonSelectableObjectCase(
+      // FORCE-create a view over a missing table -> view is INVALID; SELECT raises ORA-04063.
+      setup = Seq("CREATE FORCE VIEW invalid_view AS SELECT * FROM non_existent_table_xyz"),
+      objectName = "INVALID_VIEW",
+      teardown = Seq("DROP VIEW invalid_view")))
+
+  namedGridTest("SPARK-57778: non-selectable object is handled gracefully")(
+      nonSelectableObjectCases) { testCase =>
+    Using.resource(getConnection()) { conn =>
+      testCase.setup.foreach(conn.prepareStatement(_).executeUpdate())
+    }
+    try {
+      val tableCatalog =
+        spark.sessionState.catalogManager.catalog(catalogName).asInstanceOf[TableCatalog]
+
+      // tableExists treats a non-selectable object as non-existent instead of throwing.
+      assert(!tableCatalog.tableExists(Identifier.of(Array("SYSTEM"), testCase.objectName)))
+
+      // Reading it surfaces a dedicated, clear error instead of a raw JDBC failure.
+      val e = intercept[AnalysisException] {
+        sql(s"SELECT * FROM $catalogName.SYSTEM.${testCase.objectName}").collect()
+      }
+      assert(e.getCondition == "JDBC_OBJECT_NOT_SELECTABLE")
+      assert(e.getMessageParameters.get("objectName").contains(testCase.objectName))
+    } finally {
+      Using.resource(getConnection()) { conn =>
+        testCase.teardown.foreach(conn.prepareStatement(_).executeUpdate())
+      }
+    }
+  }
+
   override def testDatetime(tbl: String): Unit = {
     val df1 = sql(s"SELECT name FROM $tbl WHERE " +
       "dayofyear(date1) > 100 AND dayofmonth(date1) > 10 ")

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -1703,6 +1703,17 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
     new NoSuchTableException(catalogName +: ident.asMultipartIdentifier)
   }
 
+  def objectNotSelectableError(
+      catalogName: String,
+      ident: Identifier,
+      cause: Throwable): Throwable = {
+    new AnalysisException(
+      errorClass = "JDBC_OBJECT_NOT_SELECTABLE",
+      messageParameters = Map(
+        "objectName" -> toSQLId(catalogName +: ident.asMultipartIdentifier)),
+      cause = Some(cause))
+  }
+
   /**
    * Table or view not found (TABLE_OR_VIEW_NOT_FOUND). The `searchPath` segment uses
    * `nameParts.dropRight(1)` when `nameParts` has more than one part (catalog plus namespace);

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

@@ -79,6 +79,9 @@ object JDBCRDD extends Logging {
       case e: SQLException if ident.isDefined &&
         dialect.isObjectNotFoundException(e) =>
         throw QueryCompilationErrors.noSuchTableError(catalogName.get, ident.get)
+      case e: SQLException if ident.isDefined &&
+        dialect.isNotSelectableObjectException(e) =>
+        throw QueryCompilationErrors.objectNotSelectableError(catalogName.get, ident.get, e)
       case e: SQLException if dialect.isSyntaxErrorBestEffort(e) =>
         throw new SparkException(
           errorClass = "JDBC_EXTERNAL_ENGINE_SYNTAX_ERROR.DURING_OUTPUT_SCHEMA_RESOLUTION",

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

@@ -77,7 +77,9 @@ object JdbcUtils extends Logging with SQLConfHelper {
 
     executionResult match {
       case Success(_) => true
-      case Failure(e: SQLException) if dialect.isObjectNotFoundException(e) => false
+      case Failure(e: SQLException)
+        if dialect.isObjectNotFoundException(e) || dialect.isNotSelectableObjectException(e) =>
+        false
       case Failure(e) => throw e  // Re-throw unexpected exceptions
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala

@@ -808,6 +808,9 @@ abstract class JdbcDialect extends Serializable with Logging {
     Option(e.getSQLState).exists(_.startsWith("42"))
   }
 
+  @Since("4.3.0")
+  def isNotSelectableObjectException(e: SQLException): Boolean = false
+
   /**
    * Gets a dialect exception, classifies it and wraps it by `AnalysisException`.
    * @param e The dialect specific exception.

sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala

@@ -54,6 +54,13 @@ private case class OracleDialect() extends JdbcDialect with SQLConfHelper with N
       e.getMessage.contains("ORA-39165")
   }
 
+  override def isNotSelectableObjectException(e: SQLException): Boolean = {
+    // ORA-04044: object is not a table (e.g. a synonym to a procedure/function/package).
+    e.getMessage.contains("ORA-04044") ||
+      // ORA-04063: object is invalid (e.g. a view over a dropped base table).
+      e.getMessage.contains("ORA-04063")
+  }
+
   class OracleSQLBuilder extends JDBCSQLBuilder {
 
     override def visitExtract(extract: Extract): String = {