Skip to content

[FEA] Support yyyyMMdd under CORRECTED policy [databricks] #15820

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gerashegalov wants to merge 4 commits into
base: main
Choose a base branch
from
+167 −20
Open

[FEA] Support yyyyMMdd under CORRECTED policy [databricks] #15820

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5e0ccde
Support yyyyMMdd under CORRECTED policy
gerashegalov Aug 28, 2026
1e57413
Fix yyyyMMdd legacy integration tests
gerashegalov Aug 29, 2026
18bd197
Respect EXCEPTION datetime format semantics
gerashegalov Sep 2, 2026
d6b7179
Merge remote-tracking branch 'origin/main' into codex/pr15820-upmerge…
gerashegalov Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 127 additions & 4 deletions integration_tests/src/main/python/date_time_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -546,7 +546,10 @@ def test_unsupported_fallback_from_unixtime(data_gen):
@pytest.mark.parametrize('operator', ["to_unix_timestamp", "unix_timestamp", "to_timestamp", "to_date"], ids=idfn)
def test_string_to_timestamp_functions_ansi_invalid(invalid, fmt, parser_policy, operator):
sql = "{operator}(a, '{fmt}')".format(fmt=fmt, operator=operator)
parser_policy_dic = {"spark.sql.legacy.timeParserPolicy": "{}".format(parser_policy)}
parser_policy_dic = {
"spark.sql.legacy.timeParserPolicy": "{}".format(parser_policy),
"spark.rapids.sql.hasExtendedYearValues": False,
}

def fun(spark):
df = spark.createDataFrame([(invalid,)], "a string")
Expand All @@ -560,11 +563,15 @@ def fun(spark):
def test_string_to_timestamp_functions_ansi_valid(parser_policy):
expr_format = "{operator}(date_format(a, '{fmt}'), '{fmt}')"
formats = ['yyyy-MM-dd', 'yyyy/MM/dd', 'yyyy-MM', 'yyyy/MM', 'dd/MM/yyyy', 'yyyy-MM-dd HH:mm:ss',
'MM-dd', 'MM/dd', 'dd-MM', 'dd/MM', 'MM/yyyy', 'MM-yyyy', 'MM/dd/yyyy', 'MM-dd-yyyy']
'MM-dd', 'MM/dd', 'dd-MM', 'dd/MM', 'MM/yyyy', 'MM-yyyy', 'MM/dd/yyyy',
'MM-dd-yyyy']
operators = ["to_unix_timestamp", "unix_timestamp", "to_timestamp", "to_date"]
format_operator_pairs = [(fmt, operator) for fmt in formats for operator in operators]
expr_list = [expr_format.format(operator=operator, fmt=fmt) for (fmt, operator) in format_operator_pairs]
parser_policy_dic = {"spark.sql.legacy.timeParserPolicy": "{}".format(parser_policy)}
parser_policy_dic = {
"spark.sql.legacy.timeParserPolicy": "{}".format(parser_policy),
"spark.rapids.sql.hasExtendedYearValues": False,
}

def fun(spark):
df = spark.createDataFrame([(datetime(1970, 8, 12, tzinfo=timezone.utc),)], "a timestamp")
Expand All @@ -573,6 +580,62 @@ def fun(spark):
assert_gpu_and_cpu_are_equal_collect(fun, conf=copy_and_update(parser_policy_dic, ansi_enabled_conf))


@disable_ansi_mode
def test_string_to_timestamp_functions_corrected_yyyyMMdd():
data = [
("20260625",), # valid
("20240229",), # valid leap day
("99991231",), # valid upper four-digit year boundary
("20260625x",), # trailing character
("2024101",), # too few digits
("202606250",), # too many digits
("20260230",), # invalid day
("20261301",), # invalid month
("20260001",), # zero month
("2026a625",), # non-digit
(" 20260625",), # leading whitespace
("20260625 ",), # trailing whitespace
]
operators = ["to_unix_timestamp", "unix_timestamp", "to_timestamp", "to_date"]
assert_gpu_and_cpu_are_equal_collect(
lambda spark: spark.createDataFrame(data, "a string").selectExpr(*[
"{}(a, 'yyyyMMdd')".format(operator) for operator in operators
]),
{'spark.sql.legacy.timeParserPolicy': 'CORRECTED',
'spark.rapids.sql.hasExtendedYearValues': False,
'spark.rapids.sql.incompatibleDateFormats.enabled': False})


@pytest.mark.parametrize('operator',
["to_unix_timestamp", "unix_timestamp", "to_timestamp", "to_date"],
ids=idfn)
def test_string_to_timestamp_functions_corrected_yyyyMMdd_ansi_valid(operator):
assert_gpu_and_cpu_are_equal_collect(
lambda spark: spark.createDataFrame([("20260625",)], "a string")
.selectExpr("{}(a, 'yyyyMMdd')".format(operator)),
{'spark.sql.ansi.enabled': True,
'spark.sql.legacy.timeParserPolicy': 'CORRECTED',
'spark.rapids.sql.hasExtendedYearValues': False,
'spark.rapids.sql.incompatibleDateFormats.enabled': False})


@pytest.mark.parametrize('operator',
["to_unix_timestamp", "unix_timestamp", "to_timestamp", "to_date"],
ids=idfn)
def test_string_to_timestamp_functions_corrected_yyyyMMdd_ansi_invalid(operator):
assert_gpu_and_cpu_error(
lambda spark: spark.createDataFrame([("20260230",)], "a string")
.selectExpr("{}(a, 'yyyyMMdd')".format(operator))
.collect(),
conf={
'spark.sql.ansi.enabled': True,
'spark.sql.legacy.timeParserPolicy': 'CORRECTED',
'spark.rapids.sql.hasExtendedYearValues': False,
'spark.rapids.sql.incompatibleDateFormats.enabled': False,
},
error_message="Exception")


exception_policy_operators = [
"to_unix_timestamp", "unix_timestamp", "to_timestamp", "to_date"]
if not is_before_spark_350():
Expand All @@ -592,6 +655,7 @@ def fun(spark):
conf={
'spark.sql.ansi.enabled': ansi_enabled,
'spark.sql.legacy.timeParserPolicy': 'EXCEPTION',
'spark.rapids.sql.hasExtendedYearValues': False,
'spark.rapids.sql.incompatibleDateFormats.enabled': False,
},
error_message="different result")
Expand Down Expand Up @@ -739,6 +803,7 @@ def test_formats_for_legacy_mode(input_format, output_format):
from tab
'''.format(input_format=input_format, output_format=output_format),
{'spark.sql.legacy.timeParserPolicy': 'LEGACY',
'spark.rapids.sql.hasExtendedYearValues': False,
'spark.rapids.sql.incompatibleDateFormats.enabled': True})


Expand Down Expand Up @@ -887,6 +952,64 @@ def test_to_timestamp_unset_policy_corrected_default():
lambda spark: spark.createDataFrame(data, "a string")
.select(f.to_timestamp(f.col("a"), "yyyy-MM-dd HH:mm:ss")))


@disable_ansi_mode # ANSI mode is tested separately.
@tz_sensitive_test
@pytest.mark.skipif(is_before_spark_400(),
reason="Spark 4.0+ defaults timeParserPolicy to CORRECTED")
def test_to_timestamp_yyyyMMdd_unset_policy_corrected_default():
data = [("20260625",), ("20260625x",), ("2024101",), ("20260230",)]
assert_gpu_and_cpu_are_equal_collect(
lambda spark: spark.createDataFrame(data, "a string")
.select(f.to_timestamp(f.col("a"), "yyyyMMdd")),
{'spark.rapids.sql.hasExtendedYearValues': False})


@disable_ansi_mode
@allow_non_gpu('ProjectExec', 'GetTimestamp')
@pytest.mark.parametrize('parser_policy', ['CORRECTED', 'EXCEPTION'], ids=idfn)
def test_to_timestamp_yyyyMMdd_extended_year_fallback(parser_policy):
data = [("+123450101",), ("-00010101",)]
assert_gpu_fallback_collect(
lambda spark: spark.createDataFrame(data, "a string")
.selectExpr("cast(to_timestamp(a, 'yyyyMMdd') as string)"),
'GetTimestamp',
{'spark.sql.legacy.timeParserPolicy': parser_policy,
'spark.sql.session.timeZone': 'UTC',
'spark.rapids.sql.hasExtendedYearValues': True,
'spark.rapids.sql.expression.cpuBridge.enabled': False})


@disable_ansi_mode
@allow_non_gpu('ProjectExec', 'GetTimestamp')
def test_to_timestamp_yyyyMMdd_exception_policy_fallback():
conf = {
'spark.sql.legacy.timeParserPolicy': 'EXCEPTION',
'spark.rapids.sql.hasExtendedYearValues': False,
'spark.rapids.sql.expression.cpuBridge.enabled': False,
}
assert_gpu_fallback_collect(
lambda spark: spark.createDataFrame([("20240101",)], "a string")
.selectExpr("to_timestamp(a, 'yyyyMMdd')"),
'GetTimestamp',
conf)


@disable_ansi_mode
@allow_non_gpu('ProjectExec', 'GetTimestamp')
def test_to_timestamp_yyyyMMdd_exception_policy_disagreement():
assert_gpu_and_cpu_error(
lambda spark: spark.createDataFrame([("2024101",)], "a string")
.selectExpr("to_timestamp(a, 'yyyyMMdd')")
.collect(),
conf={
'spark.sql.legacy.timeParserPolicy': 'EXCEPTION',
'spark.rapids.sql.hasExtendedYearValues': False,
'spark.rapids.sql.expression.cpuBridge.enabled': False,
},
error_message="different result")


@tz_sensitive_test
@pytest.mark.parametrize("ansi_enabled", [True, False], ids=['ANSI_ON', 'ANSI_OFF'])
def test_to_date(ansi_enabled):
Expand Down Expand Up @@ -927,7 +1050,7 @@ def test_to_date_ansi_exception():
conf=ansi_enabled_conf)

supported_date_formats = ['yyyy-MM-dd', 'yyyy-MM', 'yyyy/MM/dd', 'yyyy/MM', 'dd/MM/yyyy',
'MM-dd', 'MM/dd', 'dd-MM', 'dd/MM']
'MM-dd', 'MM/dd', 'dd-MM', 'dd/MM', 'yyyyMMdd']
@pytest.mark.parametrize('date_format', supported_date_formats, ids=idfn)
@pytest.mark.parametrize('data_gen', [date_gen], ids=idfn)
@allow_non_gpu('DateFormatClass', 'Cast')
Expand Down
17 changes: 13 additions & 4 deletions sql-plugin/src/main/scala/com/nvidia/spark/rapids/DateUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@ import com.nvidia.spark.rapids.shims.DateTimeUtilsShims

import org.apache.spark.sql.catalyst.util.DateTimeUtils.localDateToDays
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.rapids.{GpuToTimestamp, LegacyTimeParserPolicy}
import org.apache.spark.sql.rapids.{ExceptionTimeParserPolicy, GpuToTimestamp,
LegacyTimeParserPolicy}

/**
* Class for helper functions for Date
Expand Down Expand Up @@ -228,8 +229,16 @@ object DateUtils {
} else {
GpuToTimestamp.LEGACY_COMPATIBLE_FORMATS
}
val timeParserPolicy = GpuOverrides.getTimeParserPolicy
val nonLegacyCompatibleFormats = if (parseString &&
timeParserPolicy == ExceptionTimeParserPolicy) {
GpuToTimestamp.EXCEPTION_COMPATIBLE_FORMATS
} else {
// Formatting does not have parser-policy disagreement, so use the CORRECTED set.
GpuToTimestamp.CORRECTED_COMPATIBLE_FORMATS
}
var strfFormat: String = null
if (GpuOverrides.getTimeParserPolicy == LegacyTimeParserPolicy) {
if (timeParserPolicy == LegacyTimeParserPolicy) {
try {
// try and convert the format to cuDF format - this will throw an exception if
// the format contains unsupported characters or words
Expand Down Expand Up @@ -261,9 +270,9 @@ object DateUtils {
// the format contains unsupported characters or words
strfFormat = toStrf(formatToConvert, parseString)
// format parsed ok, so it is either compatible (tested/certified) or incompatible
if (!GpuToTimestamp.CORRECTED_COMPATIBLE_FORMATS.contains(formatToConvert) &&
if (!nonLegacyCompatibleFormats.contains(formatToConvert) &&
!meta.conf.incompatDateFormats) {
meta.willNotWorkOnGpu(s"CORRECTED format '$sparkFormat' on the GPU is not guaranteed " +
meta.willNotWorkOnGpu(s"Format '$sparkFormat' on the GPU is not guaranteed " +
s"to produce the same results as Spark on CPU. Set " +
s"${RapidsConf.INCOMPATIBLE_DATE_FORMATS.key}=true to force onto GPU.")
}
Expand Down
39 changes: 27 additions & 12 deletions sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ import com.nvidia.spark.rapids.RapidsPluginImplicits._
import com.nvidia.spark.rapids.jni.{Arithmetic, CastException, CastStrings, DateTimeUtils,
GpuTimeZoneDB}
import com.nvidia.spark.rapids.shims.{NullIntolerantShim, ShimBinaryExpression, ShimExpression,
TruncTimestampShims}
TruncTimestampShims, YearParseUtil}

import org.apache.spark.sql.catalyst.expressions.{BinaryExpression, ExpectsInputTypes, Expression, FromUnixTime, FromUTCTimestamp, ImplicitCastInputTypes, MonthsBetween, TimeZoneAwareExpression, ToUTCTimestamp, TruncDate, TruncTimestamp}
import org.apache.spark.sql.catalyst.util.DateTimeConstants
Expand Down Expand Up @@ -417,6 +417,10 @@ abstract class UnixTimeExprMeta[A <: BinaryExpression with TimeZoneAwareExpressi
sparkFormat,
expr.left.dataType == DataTypes.StringType,
allowLegacyFormattingOnlyFormats = allowLegacyFormattingOnlyFormats)
// The fused parser only accepts an unsigned four-digit year for this packed format.
if (expr.left.dataType == DataTypes.StringType && sparkFormat == "yyyyMMdd") {
YearParseUtil.tagParseStringAsDate(conf, this)
}
case None =>
willNotWorkOnGpu("format has to be a string literal")
}
Expand Down Expand Up @@ -599,9 +603,9 @@ object ExceptionTimeParserPolicy extends TimeParserPolicy
object CorrectedTimeParserPolicy extends TimeParserPolicy

object GpuToTimestamp {
// We are compatible with Spark for these formats when the timeParserPolicy is CORRECTED
// or EXCEPTION. It is possible that other formats may be supported but these are the only
// ones that we have tests for.
// We are compatible with Spark for these formats when the timeParserPolicy is CORRECTED.
// It is possible that other formats may be supported but these are the only ones that we
// have tests for.
val CORRECTED_COMPATIBLE_FORMATS = Set(
"yyyy-MM-dd",
"yyyy/MM/dd",
Expand All @@ -617,9 +621,14 @@ object GpuToTimestamp {
"MM-yyyy",
"MM/dd/yyyy",
"MM-dd-yyyy",
"yyyyMMdd",

@rishic3 rishic3 Sep 1, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This list claims to be for both CORRECTED and EXCEPTION, and then there is a separate list for LEGACY, but I think that is misleading. EXCEPTION is supposed to:

  1. try the CORRECTED parser (returning on success)
  2. if it fails try the LEGACY parser
  3. if the LEGACY parser succeeds throw SparkUpgradeException

Per step 2 that means being supported under EXCEPTION also requires LEGACY to produce valid success/failure semantics. E.g. AI came up with the combination of 2024101 to yyyyMMdd under EXCEPTION, which should fail (Spark LEGACY accepts it) but we succeed (the JNI throws on it in LEGACY). I think we need three compatibility lists: CORRECTED, EXCEPTION, and LEGACY, where atm yyyyMMdd is not EXCEPTION compatible.

@gerashegalov gerashegalov Sep 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. Fixed in 18bd197 by separating EXCEPTION_COMPATIBLE_FORMATS from CORRECTED_COMPATIBLE_FORMATS and selecting the policy-specific set during both tagging and execution. yyyyMMdd now falls back to CPU under EXCEPTION, so Spark preserves the CORRECTED/LEGACY disagreement behavior. I added coverage that asserts GetTimestamp fallback for a normal yyyyMMdd value and the expected error for 2024101. The focused Spark 3.5.7 GPU run passed.

@rishic3 rishic3 Sep 1, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

On a separate note this list is also used by date_format, not just parsing. I don't know if we've verified that the reverse direction date_format(timestamp, 'yyyyMMdd') matches Spark.

@gerashegalov gerashegalov Sep 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Verified and covered in 18bd197. yyyyMMdd is now included in the direct date_format parity matrix for both Date and Timestamp inputs, including the runtime-fallback and timezone-rule cases. The focused Spark 3.5.7 GPU run selected 25 yyyyMMdd cases: 24 passed and the Spark-4-only case skipped, with no failures or errors.

"MMyyyy"
)

// EXCEPTION first tries CORRECTED parsing and then probes LEGACY parsing on failure. Formats
// in this set must therefore match Spark under both parsers, including success/failure behavior.
val EXCEPTION_COMPATIBLE_FORMATS = CORRECTED_COMPATIBLE_FORMATS - "yyyyMMdd"

// We are compatible with Spark for these formats when the timeParserPolicy is LEGACY. It
// is possible that other formats may be supported but these are the only ones that we have
// tests for.
Expand Down Expand Up @@ -666,12 +675,13 @@ object GpuToTimestamp {
}

// True iff the fused JNI parser handles this (sparkFormat, policy) combination.
// Today the JNI accepts every entry in CORRECTED_COMPATIBLE_FORMATS / LEGACY_COMPATIBLE_FORMATS.
private def isSimpleSparkFormat(sparkFormat: String, isLegacy: Boolean): Boolean = {
if (isLegacy) {
LEGACY_COMPATIBLE_FORMATS.contains(sparkFormat)
} else {
CORRECTED_COMPATIBLE_FORMATS.contains(sparkFormat)
private def isSimpleSparkFormat(
sparkFormat: String,
timeParserPolicy: TimeParserPolicy): Boolean = {
timeParserPolicy match {
case LegacyTimeParserPolicy => LEGACY_COMPATIBLE_FORMATS.contains(sparkFormat)
case ExceptionTimeParserPolicy => EXCEPTION_COMPATIBLE_FORMATS.contains(sparkFormat)
case CorrectedTimeParserPolicy => CORRECTED_COMPATIBLE_FORMATS.contains(sparkFormat)
}
}

Expand Down Expand Up @@ -717,7 +727,12 @@ object GpuToTimestamp {
exceptionPolicy: Boolean): ColumnVector = {

// `tsVector` will be closed in replaceSpecialDates
val tsVector = if (isSimpleSparkFormat(sparkFormat, isLegacy = false)) {
val timeParserPolicy = if (exceptionPolicy) {
ExceptionTimeParserPolicy
} else {
CorrectedTimeParserPolicy
}
val tsVector = if (isSimpleSparkFormat(sparkFormat, timeParserPolicy)) {
// Fused kernel skips the regex+length+cuDF-asTimestamp chain.
val parsed = try {
val parserPolicy = if (exceptionPolicy) {
Expand Down Expand Up @@ -777,7 +792,7 @@ object GpuToTimestamp {
def parseStringAsTimestampWithLegacyParserPolicy(
lhs: GpuColumnVector,
sparkFormat: String): ColumnVector = {
if (!isSimpleSparkFormat(sparkFormat, isLegacy = true)) {
if (!isSimpleSparkFormat(sparkFormat, LegacyTimeParserPolicy)) {
throw new IllegalStateException(s"Unsupported format $sparkFormat")
}
CastStrings.parseTimestampWithFormat(lhs.getBase, sparkFormat, true)
Expand Down
Loading