[SPARK-56738][SQL] Make Oracle bare NUMBER default scale configurable

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -52,7 +52,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
 import org.apache.spark.sql.connector.catalog.PathElement.PathRef
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
-import org.apache.spark.sql.types.{AtomicType, TimestampNTZType, TimestampType}
+import org.apache.spark.sql.types.{AtomicType, DecimalType, TimestampNTZType, TimestampType}
 import org.apache.spark.storage.{StorageLevel, StorageLevelMapper}
 import org.apache.spark.unsafe.array.ByteArrayMethods
 import org.apache.spark.util.{HadoopFSUtils, Utils, VersionUtils}
@@ -6743,6 +6743,18 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val JDBC_ORACLE_NUMBER_DEFAULT_SCALE =
+    buildConf("spark.sql.jdbc.oracle.numberDefaultScale")
+      .doc("Default scale for Oracle NUMBER columns that have no explicit precision/scale. " +
+        "Values with more fractional digits than this scale are silently rounded. " +
+        "Can be overridden per source via the oracle.numberDefaultScale JDBC read option.")
+      .version("5.0.0")
+      .withBindingPolicy(ConfigBindingPolicy.SESSION)
+      .intConf
+      .checkValue(s => s >= 0 && s <= DecimalType.MAX_SCALE,
+        s"The scale must be between 0 and ${DecimalType.MAX_SCALE}, inclusive.")
+      .createWithDefault(10)
+
   val LEGACY_DB2_TIMESTAMP_MAPPING_ENABLED =
     buildConf("spark.sql.legacy.db2.numericMapping.enabled")
       .internal()
@@ -8836,6 +8848,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
   def legacyOracleTimestampMappingEnabled: Boolean =
     getConf(LEGACY_ORACLE_TIMESTAMP_MAPPING_ENABLED)
 
+  def jdbcOracleNumberDefaultScale: Int =
+    getConf(JDBC_ORACLE_NUMBER_DEFAULT_SCALE)
+
   def legacyDB2numericMappingEnabled: Boolean =
     getConf(LEGACY_DB2_TIMESTAMP_MAPPING_ENABLED)
 

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

@@ -29,7 +29,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.TimestampNTZType
+import org.apache.spark.sql.types.{DecimalType, TimestampNTZType}
 import org.apache.spark.util.Utils
 
 /**
@@ -262,6 +262,14 @@ class JDBCOptions(
       s"$value "
     }).getOrElse("")
 
+  val oracleNumberDefaultScale = parameters.get(JDBC_ORACLE_NUMBER_DEFAULT_SCALE).map { v =>
+    val scale = v.toInt
+    require(scale >= 0 && scale <= DecimalType.MAX_SCALE,
+      s"Invalid value `$v` for option `$JDBC_ORACLE_NUMBER_DEFAULT_SCALE`." +
+        s" The scale must be between 0 and ${DecimalType.MAX_SCALE}, inclusive.")
+    scale
+  }
+
   override def hashCode: Int = this.parameters.hashCode()
 
   override def equals(other: Any): Boolean = other match {
@@ -367,4 +375,5 @@ object JDBCOptions {
   val JDBC_PREPARE_QUERY = newOption("prepareQuery")
   val JDBC_PREFER_TIMESTAMP_NTZ = newOption("preferTimestampNTZ")
   val JDBC_HINT_STRING = newOption("hint")
+  val JDBC_ORACLE_NUMBER_DEFAULT_SCALE = newOption("oracle.numberDefaultScale")
 }

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

@@ -106,7 +106,8 @@ object JDBCRDD extends Logging {
       statement.setQueryTimeout(options.queryTimeout)
       Using.resource(statement.executeQuery()) { rs =>
         JdbcUtils.getSchema(conn, rs, dialect, alwaysNullable = true,
-          isTimestampNTZ = options.preferTimestampNTZ)
+          isTimestampNTZ = options.preferTimestampNTZ,
+          oracleNumberDefaultScale = options.oracleNumberDefaultScale)
       }
     }
   }

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

@@ -262,7 +262,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
       try {
         statement.setQueryTimeout(options.queryTimeout)
         Some(getSchema(conn, statement.executeQuery(), dialect,
-          isTimestampNTZ = options.preferTimestampNTZ))
+          isTimestampNTZ = options.preferTimestampNTZ,
+          oracleNumberDefaultScale = options.oracleNumberDefaultScale))
       } catch {
         case _: SQLException => None
       } finally {
@@ -286,7 +287,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
       resultSet: ResultSet,
       dialect: JdbcDialect,
       alwaysNullable: Boolean = false,
-      isTimestampNTZ: Boolean = false): StructType = {
+      isTimestampNTZ: Boolean = false,
+      oracleNumberDefaultScale: Option[Int] = None): StructType = {
     val rsmd = resultSet.getMetaData
     val ncols = rsmd.getColumnCount
     val fields = new Array[StructField](ncols)
@@ -328,6 +330,7 @@ object JdbcUtils extends Logging with SQLConfHelper {
       metadata.putBoolean("isTimestampNTZ", isTimestampNTZ)
       metadata.putLong("scale", fieldScale)
       metadata.putString("jdbcClientType", typeName)
+      oracleNumberDefaultScale.foreach(s => metadata.putLong("numberDefaultScale", s))
       dialect.updateExtraColumnMeta(conn, rsmd, i + 1, metadata)
 
       val columnType =

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

@@ -135,6 +135,10 @@ abstract class JdbcDialect extends Serializable with Logging {
    *             <li>
    *               `scale`: The length of fractional part [[java.sql.ResultSetMetaData#getScale]]
    *             </li>
+   *             <li>
+   *               `numberDefaultScale`: Per-source override of the fallback scale for Oracle bare
+   *               NUMBER columns, from the `oracle.numberDefaultScale` JDBC option.
+   *             </li>
    *            </ul>
    * @return An option the actual DataType (subclasses of [[org.apache.spark.sql.types.DataType]])
    *         or None if the default type mapping should be used.

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

@@ -160,12 +160,12 @@ private case class OracleDialect() extends JdbcDialect with SQLConfHelper with N
           // https://github.com/apache/spark/pull/8780#issuecomment-145598968
           // and
           // https://github.com/apache/spark/pull/8780#issuecomment-144541760
-          case 0 => Option(DecimalType(DecimalType.MAX_PRECISION, 10))
+          case 0 => Some(fallbackNumberDecimalType(md))
           // Handle FLOAT fields in a special way because JDBC ResultSetMetaData converts
           // this to NUMERIC with -127 scale
           // Not sure if there is a more robust way to identify the field as a float (or other
           // numeric types that do not specify a scale.
-          case _ if scale == -127L => Option(DecimalType(DecimalType.MAX_PRECISION, 10))
+          case _ if scale == -127L => Some(fallbackNumberDecimalType(md))
           case _ => None
         }
       case TIMESTAMP_TZ | TIMESTAMP_LTZ =>
@@ -185,6 +185,21 @@ private case class OracleDialect() extends JdbcDialect with SQLConfHelper with N
     }
   }
 
+  // Returns the DecimalType to use for bare Oracle NUMBER columns (no precision/scale).
+  // Reads the per-source oracle.numberDefaultScale option from metadata first, then falls back
+  // to spark.sql.jdbc.oracle.numberDefaultScale. Logs a warning because values with more
+  // fractional digits than the resolved scale will be silently rounded.
+  private def fallbackNumberDecimalType(md: MetadataBuilder): DecimalType = {
+    val meta = if (null != md) Some(md.build()) else None
+    val scale = meta.filter(_.contains("numberDefaultScale"))
+      .map(_.getLong("numberDefaultScale").toInt)
+      .getOrElse(conf.jdbcOracleNumberDefaultScale)
+    logWarning(s"Oracle NUMBER column has no precision/scale; using scale=$scale " +
+      s"(spark.sql.jdbc.oracle.numberDefaultScale / oracle.numberDefaultScale). " +
+      "Increase it if values are being silently rounded.")
+    DecimalType(DecimalType.MAX_PRECISION, scale)
+  }
+
   override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
     // For more details, please see
     // https://docs.oracle.com/cd/E19501-01/819-3659/gcmaz/

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -1506,6 +1506,38 @@ class JDBCSuite extends SharedSparkSession {
       Some(TimestampType))
   }
 
+  test("SPARK-56738: OracleDialect bare NUMBER scale is configurable") {
+    val oracleDialect = JdbcDialects.get("jdbc:oracle")
+    val mdNoScale = new MetadataBuilder().putString("name", "c").putLong("scale", 0)
+    val mdNegScale = new MetadataBuilder().putString("name", "c").putLong("scale", -127)
+
+    // default matches legacy hardcoded behavior
+    assert(oracleDialect.getCatalystType(java.sql.Types.NUMERIC, "numeric", 0, null) ==
+      Some(DecimalType(DecimalType.MAX_PRECISION, 10)))
+    assert(oracleDialect.getCatalystType(java.sql.Types.NUMERIC, "float", 1, mdNegScale) ==
+      Some(DecimalType(DecimalType.MAX_PRECISION, 10)))
+
+    withSQLConf(SQLConf.JDBC_ORACLE_NUMBER_DEFAULT_SCALE.key -> "25") {
+      assert(oracleDialect.getCatalystType(java.sql.Types.NUMERIC, "numeric", 0, null) ==
+        Some(DecimalType(DecimalType.MAX_PRECISION, 25)))
+      assert(oracleDialect.getCatalystType(java.sql.Types.NUMERIC, "float", 1, mdNegScale) ==
+        Some(DecimalType(DecimalType.MAX_PRECISION, 25)))
+      assert(oracleDialect.getCatalystType(java.sql.Types.NUMERIC, "numeric", 0, mdNoScale) ==
+        Some(DecimalType(DecimalType.MAX_PRECISION, 25)))
+
+      // per-source option takes precedence over conf
+      val mdWithOption = new MetadataBuilder().putString("name", "c")
+        .putLong("scale", 0).putLong("numberDefaultScale", 30)
+      assert(oracleDialect.getCatalystType(java.sql.Types.NUMERIC, "numeric", 0, mdWithOption) ==
+        Some(DecimalType(DecimalType.MAX_PRECISION, 30)))
+    }
+
+    val e = intercept[IllegalArgumentException] {
+      withSQLConf(SQLConf.JDBC_ORACLE_NUMBER_DEFAULT_SCALE.key -> "39") {}
+    }
+    assert(e.getMessage.contains("spark.sql.jdbc.oracle.numberDefaultScale"))
+  }
+
   test("SPARK-42469: OracleDialect Limit query test") {
     // JDBC url is a required option but is not used in this test.
     val options = new JDBCOptions(Map("url" -> "jdbc:h2://host:port", "dbtable" -> "test"))
@@ -2841,6 +2873,22 @@ class JDBCSuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-56738: oracle.numberDefaultScale option") {
+    val opts = new JDBCOptions(Map("url" -> url, "dbtable" -> "t",
+      "oracle.numberDefaultScale" -> "25"))
+    assert(opts.oracleNumberDefaultScale == Some(25))
+
+    val defaultOpts = new JDBCOptions(Map("url" -> url, "dbtable" -> "t"))
+    assert(defaultOpts.oracleNumberDefaultScale == None)
+
+    Seq("-1", "39").foreach { v =>
+      val e = intercept[IllegalArgumentException] {
+        new JDBCOptions(Map("url" -> url, "dbtable" -> "t", "oracle.numberDefaultScale" -> v))
+      }.getMessage
+      assert(e.contains(s"Invalid value `$v` for option `oracle.numberDefaultScale`."))
+    }
+  }
+
   test("FAILED_JDBC.CONNECTION") {
     val testUrls = Seq(
       "jdbc:mysql",