diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index cce753a68fff5..95acc05ba3e8e 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/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) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala index 7188c2b8b2e8e..8008af095b123 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala +++ b/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") } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala index 2989c0975143f..64dc6f3c065ae 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala +++ b/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) } } } 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 index ca44e8b710b13..4cc518972727d 100644 --- 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 @@ -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 = diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala index b20faf52300ea..d78cc74959d4c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala @@ -135,6 +135,10 @@ abstract class JdbcDialect extends Serializable with Logging { *
  • * `scale`: The length of fractional part [[java.sql.ResultSetMetaData#getScale]] *
  • + *
  • + * `numberDefaultScale`: Per-source override of the fallback scale for Oracle bare + * NUMBER columns, from the `oracle.numberDefaultScale` JDBC option. + *
  • * * @return An option the actual DataType (subclasses of [[org.apache.spark.sql.types.DataType]]) * or None if the default type mapping should be used. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala index 46080dd57b1dd..bf7a733ce4949 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala +++ b/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/ diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala index 0213d18922b7b..f8f8a4f31d0f0 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala +++ b/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",