diff --git a/integration_tests/src/main/python/date_time_test.py b/integration_tests/src/main/python/date_time_test.py index 7ddca6f222b..88172515a37 100644 --- a/integration_tests/src/main/python/date_time_test.py +++ b/integration_tests/src/main/python/date_time_test.py @@ -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") @@ -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") @@ -573,18 +580,94 @@ 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(): exception_policy_operators.append("try_to_timestamp") +exception_policy_disagreement_cases = [ + ("2024-1-1", "yyyy-MM-dd"), + ("2024/1/1", "yyyy/MM/dd"), + ("2024-1", "yyyy-MM"), + ("2024/1", "yyyy/MM"), + ("1/1/2024", "dd/MM/yyyy"), + ("2024-1-1 1:2:3", "yyyy-MM-dd HH:mm:ss"), + ("1-1", "MM-dd"), + ("1/1", "MM/dd"), + ("1-1", "dd-MM"), + ("1/1", "dd/MM"), + ("1/2024", "MM/yyyy"), + ("1-2024", "MM-yyyy"), + ("1/1/2024", "MM/dd/yyyy"), + ("1-1-2024", "MM-dd-yyyy"), +] + + @pytest.mark.parametrize('ansi_enabled', [True, False], ids=['ANSI_ON', 'ANSI_OFF']) @pytest.mark.parametrize('operator', exception_policy_operators, ids=idfn) -def test_string_to_timestamp_functions_exception_policy_disagreement(ansi_enabled, operator): +@pytest.mark.parametrize('value,fmt', exception_policy_disagreement_cases, ids=idfn) +def test_string_to_timestamp_functions_exception_policy_disagreement( + ansi_enabled, operator, value, fmt): def fun(spark): - return spark.createDataFrame([("2024-05-06xxx",)], "a string") \ - .selectExpr("{}(a, 'yyyy-MM-dd')".format(operator)) \ + return spark.createDataFrame([(value,)], "a string") \ + .selectExpr("{}(a, '{}')".format(operator, fmt)) \ .collect() assert_gpu_and_cpu_error( @@ -592,6 +675,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") @@ -739,6 +823,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}) @@ -887,6 +972,76 @@ 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') +def test_to_timestamp_yyyyMMdd_corrected_extended_year_fallback(): + 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': 'CORRECTED', + '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") + + +@disable_ansi_mode +def test_to_timestamp_yyyyMMdd_legacy_incompatible_extended_year_opt_in(): + assert_gpu_and_cpu_are_equal_collect( + lambda spark: spark.createDataFrame([("20240101",)], "a string") + .selectExpr("to_timestamp(a, 'yyyyMMdd')"), + conf={ + 'spark.sql.legacy.timeParserPolicy': 'LEGACY', + 'spark.rapids.sql.hasExtendedYearValues': True, + 'spark.rapids.sql.incompatibleDateFormats.enabled': True, + 'spark.rapids.sql.expression.cpuBridge.enabled': False, + }) + + @tz_sensitive_test @pytest.mark.parametrize("ansi_enabled", [True, False], ids=['ANSI_ON', 'ANSI_OFF']) def test_to_date(ansi_enabled): @@ -918,7 +1073,39 @@ def test_to_timestamp_ansi_on_invalid_value(invalid): @pytest.mark.parametrize('data_gen', [StringGen('0[1-9][0-9]{4}')], ids=idfn) def test_to_date_format_MMyyyy(data_gen): assert_gpu_and_cpu_are_equal_collect( - lambda spark: unary_op_df(spark, data_gen).select(f.to_date(f.col("a"), "MMyyyy"))) + lambda spark: unary_op_df(spark, data_gen).select(f.to_date(f.col("a"), "MMyyyy")), + {'spark.sql.legacy.timeParserPolicy': 'CORRECTED', + 'spark.rapids.sql.hasExtendedYearValues': False}) + + +@disable_ansi_mode +@allow_non_gpu('ProjectExec', 'GetTimestamp') +def test_to_date_format_MMyyyy_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([("012024",)], "a string") + .select(f.to_date(f.col("a"), "MMyyyy")), + 'GetTimestamp', + conf) + + +@disable_ansi_mode +@allow_non_gpu('ProjectExec', 'GetTimestamp') +def test_to_date_format_MMyyyy_exception_policy_disagreement(): + assert_gpu_and_cpu_error( + lambda spark: spark.createDataFrame([("12024",)], "a string") + .select(f.to_date(f.col("a"), "MMyyyy")) + .collect(), + conf={ + 'spark.sql.legacy.timeParserPolicy': 'EXCEPTION', + 'spark.rapids.sql.hasExtendedYearValues': False, + 'spark.rapids.sql.expression.cpuBridge.enabled': False, + }, + error_message="different result") def test_to_date_ansi_exception(): assert_gpu_and_cpu_error( @@ -927,7 +1114,45 @@ 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'] +year_date_formats = [ + 'yyyy-MM-dd', 'yyyy/MM/dd', 'yyyy-MM', 'yyyy/MM', 'dd/MM/yyyy', + 'yyyy-MM-dd HH:mm:ss', 'MM/yyyy', 'MM-yyyy', 'MM/dd/yyyy', 'MM-dd-yyyy', + 'yyyyMMdd', 'MMyyyy'] +non_year_date_formats = ['MM-dd', 'MM/dd', 'dd-MM', 'dd/MM'] + + +@disable_ansi_mode +@allow_non_gpu('ProjectExec', 'DateFormatClass') +@pytest.mark.parametrize('date_format', year_date_formats, ids=idfn) +def test_date_format_extended_year_fallback(date_format): + def format_extended_years(spark): + extended_year_micros = \ + "IF(id = 0, 327403411200000000L + id, -62198755200000000L + id)" + return spark.range(2).selectExpr( + "date_format(timestamp_micros({}), '{}')".format( + extended_year_micros, date_format)) + + assert_gpu_fallback_collect( + format_extended_years, + 'DateFormatClass', + {'spark.sql.session.timeZone': 'UTC', + 'spark.rapids.sql.hasExtendedYearValues': True, + 'spark.rapids.sql.incompatibleDateFormats.enabled': False, + 'spark.rapids.sql.expression.cpuBridge.enabled': False}) + + +@disable_ansi_mode +@pytest.mark.parametrize('date_format', non_year_date_formats, ids=idfn) +def test_date_format_without_year_extended_year_enabled(date_format): + assert_gpu_and_cpu_are_equal_collect( + lambda spark: spark.range(1).selectExpr( + "date_format(timestamp_micros(327403411200000000L + id), '{}')".format( + date_format)), + {'spark.sql.session.timeZone': 'UTC', + 'spark.rapids.sql.hasExtendedYearValues': True}) + + @pytest.mark.parametrize('date_format', supported_date_formats, ids=idfn) @pytest.mark.parametrize('data_gen', [date_gen], ids=idfn) @allow_non_gpu('DateFormatClass', 'Cast') @@ -935,14 +1160,18 @@ def test_date_format_for_date_runtime_fallback(data_gen, date_format): # We will do a CPU fallback during runtime for timezones with transitions during # years > 2200 as described in https://github.com/NVIDIA/spark-rapids/issues/6840 assert_gpu_and_cpu_are_equal_collect( - lambda spark : unary_op_df(spark, data_gen).selectExpr("date_format(a, '{}')".format(date_format))) + lambda spark : unary_op_df(spark, data_gen).selectExpr( + "date_format(a, '{}')".format(date_format)), + {'spark.rapids.sql.hasExtendedYearValues': False}) @pytest.mark.parametrize('date_format', supported_date_formats, ids=idfn) @pytest.mark.parametrize('data_gen', [tz_rules_date_gen], ids=idfn) @allow_non_gpu('ProjectExec') def test_date_format_for_date_tz_rules(data_gen, date_format): assert_gpu_and_cpu_are_equal_collect( - lambda spark : unary_op_df(spark, data_gen).selectExpr("date_format(a, '{}')".format(date_format))) + lambda spark : unary_op_df(spark, data_gen).selectExpr( + "date_format(a, '{}')".format(date_format)), + {'spark.rapids.sql.hasExtendedYearValues': False}) @pytest.mark.parametrize('date_format', supported_date_formats, ids=idfn) # use 9999-12-30 instead of 9999-12-31 to avoid the issue: https://github.com/NVIDIA/spark-rapids/issues/10083 @@ -951,21 +1180,27 @@ def test_date_format_for_time_runtime_fallback(data_gen, date_format): # We will do a CPU fallback during runtime for timezones with transitions during # years > 2200 as described in https://github.com/NVIDIA/spark-rapids/issues/6840 assert_gpu_and_cpu_are_equal_collect( - lambda spark : unary_op_df(spark, data_gen).selectExpr("date_format(a, '{}')".format(date_format))) + lambda spark : unary_op_df(spark, data_gen).selectExpr( + "date_format(a, '{}')".format(date_format)), + {'spark.rapids.sql.hasExtendedYearValues': False}) @pytest.mark.parametrize('date_format', supported_date_formats, ids=idfn) # use 9999-12-30 instead of 9999-12-31 to avoid the issue: https://github.com/NVIDIA/spark-rapids/issues/10083 @pytest.mark.parametrize('data_gen', [TimestampGen(end=datetime(2200, 12, 30, 23, 59, 59, 999999, tzinfo=timezone.utc))], ids=idfn) def test_date_format_for_time_tz_rules(data_gen, date_format): assert_gpu_and_cpu_are_equal_collect( - lambda spark : unary_op_df(spark, data_gen).selectExpr("date_format(a, '{}')".format(date_format))) + lambda spark : unary_op_df(spark, data_gen).selectExpr( + "date_format(a, '{}')".format(date_format)), + {'spark.rapids.sql.hasExtendedYearValues': False}) @pytest.mark.parametrize('date_format', supported_date_formats + ['yyyyMMdd'], ids=idfn) # from 0001-02-01 to 9999-12-30 to avoid 'year 0 is out of range' @pytest.mark.parametrize('data_gen', [LongGen(min_val=int(datetime(1, 2, 1).timestamp()), max_val=int(datetime(2200, 12, 30).timestamp()))], ids=idfn) def test_from_unixtime(data_gen, date_format): assert_gpu_and_cpu_are_equal_collect( - lambda spark : unary_op_df(spark, data_gen, length=5).selectExpr("from_unixtime(a, '{}')".format(date_format))) + lambda spark : unary_op_df(spark, data_gen, length=5).selectExpr( + "from_unixtime(a, '{}')".format(date_format)), + {'spark.rapids.sql.hasExtendedYearValues': False}) @pytest.mark.parametrize('date_format', supported_date_formats + ['yyyyMMdd'], ids=idfn) # from 0001-02-01 to 9999-12-30 to avoid 'year 0 is out of range' @@ -974,7 +1209,22 @@ def test_from_unixtime_runtime_fallback(data_gen, date_format): # We will do a CPU fallback during runtime for timezones with transitions during # years > 2200 as described in https://github.com/NVIDIA/spark-rapids/issues/6840 assert_gpu_and_cpu_are_equal_collect( - lambda spark : unary_op_df(spark, data_gen, length=5).selectExpr("from_unixtime(a, '{}')".format(date_format))) + lambda spark : unary_op_df(spark, data_gen, length=5).selectExpr( + "from_unixtime(a, '{}')".format(date_format)), + {'spark.rapids.sql.hasExtendedYearValues': False}) + + +@disable_ansi_mode +@allow_non_gpu('ProjectExec', 'FromUnixTime') +def test_from_unixtime_extended_year_fallback(): + assert_gpu_fallback_collect( + lambda spark: spark.range(1).selectExpr( + "from_unixtime(327403411200L + id, 'yyyyMMdd')"), + 'FromUnixTime', + {'spark.sql.session.timeZone': 'UTC', + 'spark.rapids.sql.hasExtendedYearValues': True, + 'spark.rapids.sql.incompatibleDateFormats.enabled': False, + 'spark.rapids.sql.expression.cpuBridge.enabled': False}) unsupported_date_formats = ['F'] @pytest.mark.parametrize('date_format', unsupported_date_formats, ids=idfn) diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/DateUtils.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/DateUtils.scala index 4c7bafa28b8..b838fdee404 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/DateUtils.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/DateUtils.scala @@ -26,12 +26,17 @@ 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 */ object DateUtils { + sealed trait FormatDirection + case object Parsing extends FormatDirection + case object Formatting extends FormatDirection + val unsupportedCharacter = Set( 'k', 'K','z', 'V', 'c', 'F', 'W', 'Q', 'q', 'G', 'A', 'n', 'N', 'O', 'X', 'p', '\'', '[', ']', '#', '{', '}', 'Z', 'w', 'e', 'E', 'x', 'Z', 'Y') @@ -220,6 +225,7 @@ object DateUtils { meta: RapidsMeta[_, _, _], sparkFormat: String, parseString: Boolean, + formatDirection: FormatDirection = Parsing, inputFormat: Option[String] = None, allowLegacyFormattingOnlyFormats: Boolean = false): String = { val formatToConvert = inputFormat.getOrElse(sparkFormat) @@ -228,8 +234,15 @@ object DateUtils { } else { GpuToTimestamp.LEGACY_COMPATIBLE_FORMATS } + val timeParserPolicy = GpuOverrides.getTimeParserPolicy + val nonLegacyCompatibleFormats = formatDirection match { + case Formatting => GpuToTimestamp.FORMATTING_COMPATIBLE_FORMATS + case Parsing if parseString && timeParserPolicy == ExceptionTimeParserPolicy => + GpuToTimestamp.EXCEPTION_COMPATIBLE_FORMATS + case Parsing => 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 @@ -261,9 +274,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.") } @@ -272,6 +285,12 @@ object DateUtils { meta.willNotWorkOnGpu(s"Failed to convert ${e.reason} ${e.getMessage}") } } + if (formatDirection == Formatting && Option(strfFormat).exists(_.contains("%Y")) && + meta.conf.hasExtendedYearValues && !meta.conf.incompatDateFormats) { + meta.willNotWorkOnGpu("Formatting the full range of supported years is not supported. " + + "If your years are limited to 4 positive digits set " + + s"${RapidsConf.HAS_EXTENDED_YEAR_VALUES} to false.") + } strfFormat } } diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuMathAndDateExpressionRuleMetas.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuMathAndDateExpressionRuleMetas.scala index a7060cc834e..ca7b0b9d444 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuMathAndDateExpressionRuleMetas.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuMathAndDateExpressionRuleMetas.scala @@ -731,6 +731,7 @@ case class DateFormatClassRuleMeta( r: DataFromReplacementRule) extends UnixTimeExprMeta[DateFormatClass](a, conf, p, r) { override def isTimeZoneSupported = true + override protected def formatDirection: DateUtils.FormatDirection = DateUtils.Formatting override protected def allowLegacyFormattingOnlyFormats: Boolean = true override def convertToGpu(lhs: Expression, rhs: Expression): GpuExpression = GpuDateFormatClass(lhs, rhs, strfFormat, a.timeZoneId) diff --git a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala index 45ab4fe7881..c3845cc40a3 100644 --- a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala +++ b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala @@ -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 @@ -406,6 +406,7 @@ abstract class UnixTimeExprMeta[A <: BinaryExpression with TimeZoneAwareExpressi var strfFormat: String = _ protected def allowLegacyFormattingOnlyFormats: Boolean = false + protected def formatDirection: DateUtils.FormatDirection = DateUtils.Parsing override def tagExprForGpu(): Unit = { // Date and Timestamp work too @@ -416,7 +417,13 @@ abstract class UnixTimeExprMeta[A <: BinaryExpression with TimeZoneAwareExpressi strfFormat = DateUtils.tagAndGetCudfFormat(this, sparkFormat, expr.left.dataType == DataTypes.StringType, + formatDirection, allowLegacyFormattingOnlyFormats = allowLegacyFormattingOnlyFormats) + // The fused parser only accepts an unsigned four-digit year for this packed format. + if (expr.left.dataType == DataTypes.StringType && sparkFormat == "yyyyMMdd" && + GpuOverrides.getTimeParserPolicy == CorrectedTimeParserPolicy) { + YearParseUtil.tagParseStringAsDate(conf, this) + } case None => willNotWorkOnGpu("format has to be a string literal") } @@ -599,9 +606,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", @@ -617,6 +624,48 @@ object GpuToTimestamp { "MM-yyyy", "MM/dd/yyyy", "MM-dd-yyyy", + "yyyyMMdd", + "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. + // TODO(#15977): Re-add MMyyyy after the fused parser preserves parser-policy disagreements. + val EXCEPTION_COMPATIBLE_FORMATS = Set( + "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" + ) + + // Formatting compatibility is independent of parser-policy compatibility. Keep this explicit + // so certifying a parsing format cannot accidentally certify the reverse direction. + val FORMATTING_COMPATIBLE_FORMATS = Set( + "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", + "yyyyMMdd", "MMyyyy" ) @@ -666,12 +715,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) } } @@ -717,7 +767,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) { @@ -777,7 +832,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) @@ -966,6 +1021,7 @@ class FromUnixTimeMeta(a: FromUnixTime, } strfFormat = DateUtils.tagAndGetCudfFormat(this, sparkFormat, a.left.dataType == DataTypes.StringType, + DateUtils.Formatting, inputFormat, allowLegacyFormattingOnlyFormats = allowLegacyFormattingOnlyFormats) case None =>