This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 6e8c5efcee16 [SPARK-57837][SQL] Support the precision argument in
current_timestamp(p) and localtimestamp(p)
6e8c5efcee16 is described below
commit 6e8c5efcee16077ee97a62e518da923d834c11bb
Author: Stevo Mitric <[email protected]>
AuthorDate: Tue Aug 11 10:36:24 2026 +0000
[SPARK-57837][SQL] Support the precision argument in current_timestamp(p)
and localtimestamp(p)
### What changes were proposed in this pull request?
This PR adds an optional fractional-seconds precision argument `p` to
`current_timestamp(p)` / `now(p)` and `localtimestamp(p)`, mirroring the
existing `current_time(p)`. It is part of the umbrella SPARK-56822 (timestamps
with nanosecond precision).
- `p == 6` (and the no-argument forms) keep the historical microsecond
`TIMESTAMP` / `TIMESTAMP_NTZ` types unchanged.
- `p` in `[7, 9]` returns a nanosecond `TIMESTAMP_LTZ(p)` /
`TIMESTAMP_NTZ(p)`, gated behind `spark.sql.timestampNanosTypes.enabled`.
- Any other precision is rejected with `INVALID_TIMESTAMP_PRECISION`,
matching the `TIMESTAMP(p)` type parser (the range check runs before the
feature-flag check).
The micro leaf expressions `CurrentTimestamp` / `Now` / `LocalTimestamp`
are kept as-is to preserve their rendering and pattern matching; two new
placeholder expressions `CurrentTimestampNanos` / `LocalTimestampNanos` carry
the precision, and `CurrentTimestampExpressionBuilder` /
`LocalTimestampExpressionBuilder` dispatch on the argument count.
`ComputeCurrentTime` constant-folds the new expressions into query-stable
nanosecond literals, just like the microsecond variants.
### Why are the changes needed?
ANSI SQL / Foundation defines an optional `<timestamp precision>` on
`CURRENT_TIMESTAMP` and `LOCALTIMESTAMP` (Feature F555). This lets users obtain
the current time at nanosecond precision when the nanosecond timestamp types
are enabled.
### Does this PR introduce any user-facing change?
Yes. `current_timestamp(p)` / `now(p)` / `localtimestamp(p)` now accept a
precision argument. The no-argument forms are unchanged.
### How was this patch tested?
- Catalyst unit tests: `ComputeCurrentTimeSuite` (folding, query-stability,
precision flooring, time flow) and `DateExpressionsSuite` (the two expression
builders, including the error paths).
- End-to-end tests in `TimestampNanosFunctionsSuiteBase` (ANSI on and off):
result types, precision flooring, query-stable self-equality, out-of-range
precision, and the disabled-flag error.
- New golden cases in `timestamp-ltz-nanos.sql` /
`timestamp-ntz-nanos.sql`; regenerated `sql-expression-schema.md`.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored: Claude Code (Claude Opus 4.8)
Closes #57697 from
stevomitric/stevomitric/spark-57837-current-timestamp-precision.
Authored-by: Stevo Mitric <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit 23f1ad89b3dda1cc91757ff52028e8d815a82c43)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../sql/catalyst/analysis/FunctionRegistry.scala | 6 +-
.../analysis/UnsupportedOperationChecker.scala | 5 +-
.../catalyst/expressions/datetimeExpressions.scala | 237 +++++++++++++++++++--
.../sql/catalyst/optimizer/finishAnalysis.scala | 20 +-
.../expressions/DateExpressionsSuite.scala | 102 ++++++++-
.../optimizer/ComputeCurrentTimeSuite.scala | 84 +++++++-
.../streaming/continuous/ContinuousExecution.scala | 5 +-
.../streaming/runtime/MicroBatchExecution.scala | 12 +-
.../sql-functions/sql-expression-schema.md | 6 +-
.../analyzer-results/timestamp-ltz-nanos.sql.out | 58 +++++
.../analyzer-results/timestamp-ntz-nanos.sql.out | 56 +++++
.../sql-tests/inputs/timestamp-ltz-nanos.sql | 14 ++
.../sql-tests/inputs/timestamp-ntz-nanos.sql | 13 ++
.../sql-tests/results/timestamp-ltz-nanos.sql.out | 72 +++++++
.../sql-tests/results/timestamp-ntz-nanos.sql.out | 64 ++++++
.../sql/TimestampNanosFunctionsSuiteBase.scala | 84 +++++++-
.../sql/expressions/ExpressionInfoSuite.scala | 3 +
.../apache/spark/sql/streaming/StreamSuite.scala | 30 +++
.../sql/streaming/continuous/ContinuousSuite.scala | 18 ++
.../org/apache/spark/sql/hive/HiveInspectors.scala | 1 +
20 files changed, 850 insertions(+), 40 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
index e80544a6e260..e8bc63914716 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
@@ -701,11 +701,11 @@ object FunctionRegistry {
expression[AddMonths]("add_months"),
expression[CurrentDate]("current_date"),
expressionBuilder("curdate", CurDateExpressionBuilder, setAlias = true),
- expression[CurrentTimestamp]("current_timestamp"),
+ expressionBuilder("current_timestamp", CurrentTimestampExpressionBuilder),
expression[CurrentTime]("current_time"),
expression[CurrentTime]("localtime", since = Some("4.3.0")),
expression[CurrentTimeZone]("current_timezone"),
- expression[LocalTimestamp]("localtimestamp"),
+ expressionBuilder("localtimestamp", LocalTimestampExpressionBuilder),
expression[DateDiff]("datediff"),
expression[DateDiff]("date_diff", setAlias = true, Some("3.4.0")),
expression[DateAdd]("date_add"),
@@ -723,7 +723,7 @@ object FunctionRegistry {
expression[Month]("month"),
expression[MonthsBetween]("months_between"),
expression[NextDay]("next_day"),
- expression[Now]("now"),
+ expressionBuilder("now", NowExpressionBuilder),
expression[Quarter]("quarter"),
expressionBuilder("second", SecondExpressionBuilder),
expression[ParseToTimestamp]("to_timestamp"),
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
index eddcee169e37..1e10767b5316 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
@@ -23,7 +23,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{ANALYSIS_ERROR, QUERY_PLAN}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.ExtendedAnalysisException
-import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, CurrentDate, CurrentTimestampLike, Expression,
GroupingSets, LocalTimestamp, MonotonicallyIncreasingID, NamedExpression,
SessionWindow, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, CurrentDate, CurrentTimestampLike, Expression,
GroupingSets, LocalTimestamp, LocalTimestampNanos, MonotonicallyIncreasingID,
NamedExpression, SessionWindow, WindowExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
@@ -637,7 +637,8 @@ object UnsupportedOperationChecker extends Logging {
subPlan.expressions.foreach { e =>
if (e.collectLeaves().exists {
- case (_: CurrentTimestampLike | _: CurrentDate | _: LocalTimestamp)
=> true
+ case (_: CurrentTimestampLike | _: CurrentDate | _: LocalTimestamp |
+ _: LocalTimestampNanos) => true
case _ => false
}) {
throwError(s"Continuous processing does not support current time
operations.")
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala
index 992f58d09c3a..f2ad3107ed6d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.catalyst.util.{DateTimeUtils,
LegacyDateFormats, Tim
import org.apache.spark.sql.catalyst.util.DateTimeConstants._
import org.apache.spark.sql.catalyst.util.DateTimeUtils._
import org.apache.spark.sql.catalyst.util.LegacyDateFormats.SIMPLE_DATE_FORMAT
-import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryExecutionErrors}
+import org.apache.spark.sql.errors.{DataTypeErrors, QueryCompilationErrors,
QueryExecutionErrors}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.types.StringTypeWithCollation
import org.apache.spark.sql.types._
@@ -197,15 +197,141 @@ abstract class CurrentTimestampLike() extends
LeafExpression with CodegenFallbac
}
/**
- * Returns the current timestamp at the start of query evaluation.
+ * Returns the current timestamp at the start of query evaluation. The
no-argument micro form
+ * registered as `current_timestamp` / `now`; see
[[CurrentTimestampExpressionBuilder]] for the
+ * `current_timestamp(precision)` / `now(precision)` variants.
* There is no code generation since this expression should get constant
folded by the optimizer.
*/
-// scalastyle:off line.size.limit
+case class CurrentTimestamp() extends CurrentTimestampLike {
+ override def prettyName: String = "current_timestamp"
+}
+
+case class Now() extends CurrentTimestampLike {
+ override def prettyName: String = "now"
+}
+
+/**
+ * Returns the current timestamp without time zone at the start of query
evaluation. The
+ * no-argument micro form registered as `localtimestamp`; see
+ * [[LocalTimestampExpressionBuilder]] for the `localtimestamp(precision)`
variant.
+ * There is no code generation since this expression should get constant
folded by the optimizer.
+ */
+case class LocalTimestamp(timeZoneId: Option[String] = None) extends
LeafExpression
+ with TimeZoneAwareExpression with CodegenFallback {
+ def this() = this(None)
+ override def foldable: Boolean = true
+ override def nullable: Boolean = false
+ override def dataType: DataType = TimestampNTZType
+ final override def nodePatternsInternal(): Seq[TreePattern] =
Seq(CURRENT_LIKE)
+ override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
+ copy(timeZoneId = Option(timeZoneId))
+ override def eval(input: InternalRow): Any =
localDateTimeToMicros(LocalDateTime.now(zoneId))
+ override def prettyName: String = "localtimestamp"
+}
+
+/**
+ * Returns the current timestamp with local time zone at the start of query
evaluation, as a
+ * nanosecond-precision `TIMESTAMP_LTZ(precision)` (`precision` in `[7, 9]`).
This is the
+ * nanosecond counterpart of [[CurrentTimestamp]] / [[Now]] and is produced by
+ * [[CurrentTimestampExpressionBuilder]] when `current_timestamp(p)` /
`now(p)` is called with a
+ * nanosecond precision. Like the microsecond current-timestamp expressions it
is foldable and
+ * gets constant folded by
[[org.apache.spark.sql.catalyst.optimizer.ComputeCurrentTime]]; there
+ * is no code generation.
+ */
+case class CurrentTimestampNanos(precision: Int) extends CurrentTimestampLike {
+ override def dataType: DataType = TimestampLTZNanosType(precision)
+ override def eval(input: InternalRow): Any =
+ instantToTimestampNanos(java.time.Instant.now(), precision)
+ override def prettyName: String = "current_timestamp"
+}
+
+/**
+ * Returns the current timestamp without time zone at the start of query
evaluation, as a
+ * nanosecond-precision `TIMESTAMP_NTZ(precision)` (`precision` in `[7, 9]`).
This is the
+ * nanosecond counterpart of [[LocalTimestamp]] and is produced by
+ * [[LocalTimestampExpressionBuilder]] when `localtimestamp(p)` is called with
a nanosecond
+ * precision. Like [[LocalTimestamp]] it is time-zone aware (the session time
zone determines the
+ * wall-clock value), foldable, and gets constant folded by
+ * [[org.apache.spark.sql.catalyst.optimizer.ComputeCurrentTime]]; there is no
code generation.
+ */
+case class LocalTimestampNanos(precision: Int, timeZoneId: Option[String] =
None)
+ extends LeafExpression with TimeZoneAwareExpression with CodegenFallback {
+ def this(precision: Int) = this(precision, None)
+ override def foldable: Boolean = true
+ override def nullable: Boolean = false
+ override def dataType: DataType = TimestampNTZNanosType(precision)
+ final override def nodePatternsInternal(): Seq[TreePattern] =
Seq(CURRENT_LIKE)
+ override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
+ copy(timeZoneId = Option(timeZoneId))
+ override def eval(input: InternalRow): Any =
+ localDateTimeToTimestampNanos(LocalDateTime.now(zoneId), precision)
+ override def prettyName: String = "localtimestamp"
+}
+
+/**
+ * Shared precision handling for the `current_timestamp(p)` / `now(p)` /
`localtimestamp(p)`
+ * expression builders.
+ */
+private[expressions] object CurrentTimestampPrecision {
+ /**
+ * Validates the foldable integer precision argument `p` of a
current-timestamp function and
+ * returns it. `p == 6` selects the historical microsecond type; `p` in `[7,
9]` selects the
+ * nanosecond type (which requires `spark.sql.timestampNanosTypes.enabled`).
Any other value is
+ * rejected with `INVALID_TIMESTAMP_PRECISION`. The `typeName`
(`TIMESTAMP_LTZ` / `TIMESTAMP_NTZ`)
+ * is used in the precision / feature-flag error messages, matching the
`TIMESTAMP(p)` type
+ * parser.
+ */
+ def validate(funcName: String, precision: Expression, typeName: String): Int
= {
+ if (!precision.foldable) {
+ throw QueryCompilationErrors.nonFoldableArgumentError(
+ funcName, "precision", precision.dataType)
+ }
+ if (!precision.dataType.isInstanceOf[IntegralType]) {
+ throw QueryCompilationErrors.unexpectedInputDataTypeError(
+ funcName, 1, IntegerType, precision)
+ }
+ val value = precision.eval()
+ if (value == null) {
+ throw QueryCompilationErrors.unexpectedNullError("precision", precision)
+ }
+ val p = value.asInstanceOf[Number].intValue()
+ // Reject out-of-range precisions before the feature-flag check so the
error is always
+ // INVALID_TIMESTAMP_PRECISION, not FEATURE_NOT_ENABLED (mirrors the
TIMESTAMP(p) type parser).
+ if (p != 6 &&
+ (p < TimestampLTZNanosType.MIN_PRECISION || p >
TimestampLTZNanosType.MAX_PRECISION)) {
+ throw DataTypeErrors.invalidTimestampPrecisionError(p.toString, typeName)
+ }
+ if (p != 6) {
+ DataTypeErrors.checkTimestampNanosTypesEnabled()
+ }
+ p
+ }
+}
+
+/**
+ * Builds `current_timestamp` / `now`. The no-argument form keeps the
historical microsecond
+ * `TIMESTAMP` type ([[CurrentTimestamp]] / [[Now]]); the single-argument form
accepts a foldable
+ * integer precision `p`, returning the microsecond `TIMESTAMP` for `p == 6`
and a nanosecond
+ * `TIMESTAMP_LTZ(p)` for `p` in `[7, 9]` (gated behind
`spark.sql.timestampNanosTypes.enabled`).
+ * The precision handling mirrors the `TIMESTAMP_LTZ(p)` type parser and
`current_time(p)`.
+ *
+ * `now` is registered through the separate [[NowExpressionBuilder]] so it can
document its own
+ * examples (`now` has no braceless form, unlike the `current_timestamp`
keyword), but delegates
+ * its build logic here.
+ */
+// scalastyle:off line.size.limit line.contains.tab
@ExpressionDescription(
usage = """
_FUNC_() - Returns the current timestamp at the start of query evaluation.
All calls of current_timestamp within the same query return the same value.
_FUNC_ - Returns the current timestamp at the start of query evaluation.
+
+ _FUNC_(precision) - Returns the current timestamp at the start of query
evaluation, with the given fractional-seconds precision. A precision in [7, 9]
returns a nanosecond-precision TIMESTAMP_LTZ(precision) and requires
spark.sql.timestampNanosTypes.enabled to be true; precision 6 returns the
standard microsecond TIMESTAMP.
+ """,
+ arguments = """
+ Arguments:
+ * precision - An optional integer literal. Either 6 (the microsecond
TIMESTAMP default) or a
+ value in [7, 9] selecting a nanosecond-precision
TIMESTAMP_LTZ(precision).
""",
examples = """
Examples:
@@ -213,59 +339,116 @@ abstract class CurrentTimestampLike() extends
LeafExpression with CodegenFallbac
2020-04-25 15:49:11.914
> SELECT _FUNC_;
2020-04-25 15:49:11.914
+ > SET spark.sql.timestampNanosTypes.enabled=true;
+ spark.sql.timestampNanosTypes.enabled true
+ > SELECT _FUNC_(9);
+ 2020-04-25 15:49:11.914120463
""",
note = """
The syntax without braces has been supported since 2.0.1.
""",
group = "datetime_funcs",
since = "1.5.0")
-// scalastyle:on line.size.limit
-case class CurrentTimestamp() extends CurrentTimestampLike {
- override def prettyName: String = "current_timestamp"
+// scalastyle:on line.size.limit line.contains.tab
+object CurrentTimestampExpressionBuilder extends ExpressionBuilder {
+ override def build(funcName: String, expressions: Seq[Expression]):
Expression = {
+ expressions.length match {
+ case 0 =>
+ // Preserve the exact historical expression so `now` still maps to
`Now` and rendering /
+ // pattern matching are unchanged.
+ if (funcName.equalsIgnoreCase("now")) Now() else CurrentTimestamp()
+ case 1 =>
+ val p = CurrentTimestampPrecision.validate(funcName, expressions.head,
"TIMESTAMP_LTZ")
+ if (p == 6) {
+ if (funcName.equalsIgnoreCase("now")) Now() else CurrentTimestamp()
+ } else {
+ CurrentTimestampNanos(p)
+ }
+ case n =>
+ throw QueryCompilationErrors.wrongNumArgsError(funcName, Seq(0, 1), n)
+ }
+ }
}
+/**
+ * Builds `now` / `now(precision)`. Shares
[[CurrentTimestampExpressionBuilder]]'s build logic but
+ * carries its own `@ExpressionDescription`: `now` is a regular function with
no braceless form,
+ * so its examples must not include the `SELECT now` case that the
`current_timestamp` keyword
+ * documents.
+ */
+// scalastyle:off line.size.limit line.contains.tab
@ExpressionDescription(
- usage = "_FUNC_() - Returns the current timestamp at the start of query
evaluation.",
+ usage = """
+ _FUNC_() - Returns the current timestamp at the start of query evaluation.
All calls of now within the same query return the same value.
+
+ _FUNC_(precision) - Returns the current timestamp at the start of query
evaluation, with the given fractional-seconds precision. A precision in [7, 9]
returns a nanosecond-precision TIMESTAMP_LTZ(precision) and requires
spark.sql.timestampNanosTypes.enabled to be true; precision 6 returns the
standard microsecond TIMESTAMP.
+ """,
+ arguments = """
+ Arguments:
+ * precision - An optional integer literal. Either 6 (the microsecond
TIMESTAMP default) or a
+ value in [7, 9] selecting a nanosecond-precision
TIMESTAMP_LTZ(precision).
+ """,
examples = """
Examples:
> SELECT _FUNC_();
2020-04-25 15:49:11.914
+ > SET spark.sql.timestampNanosTypes.enabled=true;
+ spark.sql.timestampNanosTypes.enabled true
+ > SELECT _FUNC_(9);
+ 2020-04-25 15:49:11.914120463
""",
group = "datetime_funcs",
since = "1.6.0")
-case class Now() extends CurrentTimestampLike {
- override def prettyName: String = "now"
+// scalastyle:on line.size.limit line.contains.tab
+object NowExpressionBuilder extends ExpressionBuilder {
+ override def build(funcName: String, expressions: Seq[Expression]):
Expression =
+ CurrentTimestampExpressionBuilder.build(funcName, expressions)
}
/**
- * Returns the current timestamp without time zone at the start of query
evaluation.
- * There is no code generation since this expression should get constant
folded by the optimizer.
+ * Builds `localtimestamp`. The no-argument form keeps the historical
microsecond `TIMESTAMP_NTZ`
+ * type ([[LocalTimestamp]]); the single-argument form accepts a foldable
integer precision `p`,
+ * returning the microsecond `TIMESTAMP_NTZ` for `p == 6` and a nanosecond
`TIMESTAMP_NTZ(p)` for
+ * `p` in `[7, 9]` (gated behind `spark.sql.timestampNanosTypes.enabled`). The
precision handling
+ * mirrors the `TIMESTAMP_NTZ(p)` type parser and `current_time(p)`.
*/
-// scalastyle:off line.size.limit
+// scalastyle:off line.size.limit line.contains.tab
@ExpressionDescription(
usage = """
_FUNC_() - Returns the current timestamp without time zone at the start of
query evaluation. All calls of localtimestamp within the same query return the
same value.
_FUNC_ - Returns the current local date-time at the session time zone at
the start of query evaluation.
+
+ _FUNC_(precision) - Returns the current local date-time with the given
fractional-seconds precision. A precision in [7, 9] returns a
nanosecond-precision TIMESTAMP_NTZ(precision) and requires
spark.sql.timestampNanosTypes.enabled to be true; precision 6 returns the
standard microsecond TIMESTAMP_NTZ.
+ """,
+ arguments = """
+ Arguments:
+ * precision - An optional integer literal. Either 6 (the microsecond
TIMESTAMP_NTZ default)
+ or a value in [7, 9] selecting a nanosecond-precision
TIMESTAMP_NTZ(precision).
""",
examples = """
Examples:
> SELECT _FUNC_();
2020-04-25 15:49:11.914
+ > SET spark.sql.timestampNanosTypes.enabled=true;
+ spark.sql.timestampNanosTypes.enabled true
+ > SELECT _FUNC_(9);
+ 2020-04-25 15:49:11.914120463
""",
group = "datetime_funcs",
since = "3.4.0")
-case class LocalTimestamp(timeZoneId: Option[String] = None) extends
LeafExpression
- with TimeZoneAwareExpression with CodegenFallback {
- def this() = this(None)
- override def foldable: Boolean = true
- override def nullable: Boolean = false
- override def dataType: DataType = TimestampNTZType
- final override def nodePatternsInternal(): Seq[TreePattern] =
Seq(CURRENT_LIKE)
- override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
- copy(timeZoneId = Option(timeZoneId))
- override def eval(input: InternalRow): Any =
localDateTimeToMicros(LocalDateTime.now(zoneId))
- override def prettyName: String = "localtimestamp"
+// scalastyle:on line.size.limit line.contains.tab
+object LocalTimestampExpressionBuilder extends ExpressionBuilder {
+ override def build(funcName: String, expressions: Seq[Expression]):
Expression = {
+ expressions.length match {
+ case 0 => LocalTimestamp()
+ case 1 =>
+ val p = CurrentTimestampPrecision.validate(funcName, expressions.head,
"TIMESTAMP_NTZ")
+ if (p == 6) LocalTimestamp() else LocalTimestampNanos(p)
+ case n =>
+ throw QueryCompilationErrors.wrongNumArgsError(funcName, Seq(0, 1), n)
+ }
+ }
}
/**
@@ -306,6 +489,14 @@ case class CurrentBatchTimestamp(
case _: TimestampType => Literal(timestampUs, TimestampType)
case _: TimestampNTZType =>
Literal(convertTz(timestampUs, ZoneOffset.UTC, zoneId),
TimestampNTZType)
+ // The batch timestamp is millisecond resolution, so nanos-within-micro
is always 0.
+ // TIMESTAMP_LTZ is instant-based (zone-independent); TIMESTAMP_NTZ
takes the wall clock
+ // in the session zone, mirroring the micro TimestampNTZType branch
above.
+ case ltz: TimestampLTZNanosType =>
+ Literal(TimestampNanosVal.fromParts(timestampUs, 0), ltz)
+ case ntz: TimestampNTZNanosType =>
+ Literal(
+ TimestampNanosVal.fromParts(convertTz(timestampUs, ZoneOffset.UTC,
zoneId), 0), ntz)
case _: DateType => Literal(microsToDays(timestampUs, zoneId), DateType)
}
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala
index c8c00a3fa13a..f1a9ffd54e9a 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.catalyst.trees.TreePatternBits
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.catalyst.util.DateTimeUtils.{convertSpecialDate,
convertSpecialTimestamp, convertSpecialTimestampNTZ, instantToMicros,
localDateTimeToMicros}
-import
org.apache.spark.sql.catalyst.util.SparkDateTimeUtils.{instantToNanosOfDay,
truncateTimeToPrecision}
+import
org.apache.spark.sql.catalyst.util.SparkDateTimeUtils.{instantToNanosOfDay,
instantToTimestampNanos, localDateTimeToTimestampNanos, truncateTimeToPrecision}
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLExpr
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.types._
@@ -119,6 +119,11 @@ object ComputeCurrentTime extends Rule[LogicalPlan] {
val timezone = Literal.create(conf.sessionLocalTimeZone, StringType)
val currentDates = collection.mutable.HashMap.empty[ZoneId, Literal]
val localTimestamps = collection.mutable.HashMap.empty[ZoneId, Literal]
+ // Nanosecond current-timestamp literals depend on the requested precision
(sub-precision
+ // digits are floored), so they are cached separately from the microsecond
ones. LTZ is keyed
+ // by precision only; NTZ (localtimestamp) is keyed by (zone, precision)
like its micro sibling.
+ val currentTimestampNanos = collection.mutable.HashMap.empty[Int, Literal]
+ val localTimestampNanos = collection.mutable.HashMap.empty[(ZoneId, Int),
Literal]
// CAST_TO_TIMESTAMP is a dedicated tree-pattern bit set on Cast nodes
whose target type is
// any timestamp type (NTZ or LTZ family). This lets the rule reach both
TIME -> TIMESTAMP_NTZ
@@ -186,12 +191,25 @@ object ComputeCurrentTime extends Rule[LogicalPlan] {
currentTimeType.precision)
Literal.create(truncatedTime, TimeType(currentTimeType.precision))
case CurrentTimestamp() | Now() => currentTime
+ case ct: CurrentTimestampNanos =>
+ currentTimestampNanos.getOrElseUpdate(ct.precision, {
+ Literal.create(
+ instantToTimestampNanos(instant, ct.precision),
+ TimestampLTZNanosType(ct.precision))
+ })
case CurrentTimeZone() => timezone
case localTimestamp: LocalTimestamp =>
localTimestamps.getOrElseUpdate(localTimestamp.zoneId, {
val asDateTime = LocalDateTime.ofInstant(instant,
localTimestamp.zoneId)
Literal.create(localDateTimeToMicros(asDateTime),
TimestampNTZType)
})
+ case lt: LocalTimestampNanos =>
+ localTimestampNanos.getOrElseUpdate((lt.zoneId, lt.precision), {
+ val asDateTime = LocalDateTime.ofInstant(instant, lt.zoneId)
+ Literal.create(
+ localDateTimeToTimestampNanos(asDateTime, lt.precision),
+ TimestampNTZNanosType(lt.precision))
+ })
}
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DateExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DateExpressionsSuite.scala
index 5feecc86bdfc..071e1949ad96 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DateExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DateExpressionsSuite.scala
@@ -28,7 +28,7 @@ import scala.language.postfixOps
import scala.reflect.ClassTag
import scala.util.Random
-import org.apache.spark.{SparkArithmeticException, SparkDateTimeException,
SparkFunSuite, SparkIllegalArgumentException, SparkRuntimeException,
SparkUpgradeException}
+import org.apache.spark.{SPARK_DOC_ROOT, SparkArithmeticException,
SparkDateTimeException, SparkException, SparkFunSuite,
SparkIllegalArgumentException, SparkRuntimeException, SparkUpgradeException}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
@@ -3302,4 +3302,104 @@ class DateExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
val r5 = expr5.checkInputDataTypes().asInstanceOf[DataTypeMismatch]
assert(r5.errorSubClass == "UNEXPECTED_INPUT_TYPE")
}
+
+ test("SPARK-57837: CurrentTimestampExpressionBuilder") {
+ // No argument keeps the historical micro TIMESTAMP expressions.
+ assert(CurrentTimestampExpressionBuilder.build("current_timestamp",
Seq.empty) ===
+ CurrentTimestamp())
+ assert(CurrentTimestampExpressionBuilder.build("now", Seq.empty) === Now())
+
+ // Precision 6 stays on the micro type, per function name.
+ assert(CurrentTimestampExpressionBuilder.build("current_timestamp",
Seq(Literal(6))) ===
+ CurrentTimestamp())
+ assert(CurrentTimestampExpressionBuilder.build("now", Seq(Literal(6))) ===
Now())
+
+ // Precisions 7-9 build the nanosecond TIMESTAMP_LTZ variant, including a
foldable arg.
+ Seq(7, 8, 9).foreach { p =>
+ val built = CurrentTimestampExpressionBuilder.build("current_timestamp",
Seq(Literal(p)))
+ assert(built === CurrentTimestampNanos(p))
+ assert(built.dataType === TimestampLTZNanosType(p))
+ }
+ assert(CurrentTimestampExpressionBuilder.build("current_timestamp",
Seq(Add(Literal(4),
+ Literal(5)))) === CurrentTimestampNanos(9))
+
+ // Out-of-range precision (other than 6) is rejected with
INVALID_TIMESTAMP_PRECISION.
+ Seq(0, 3, 5, 10).foreach { p =>
+ checkError(
+ exception = intercept[SparkException] {
+ CurrentTimestampExpressionBuilder.build("current_timestamp",
Seq(Literal(p)))
+ },
+ condition = "INVALID_TIMESTAMP_PRECISION",
+ parameters = Map("precision" -> p.toString, "type" -> "TIMESTAMP_LTZ"))
+ }
+
+ // Non-foldable precision.
+ checkError(
+ exception = intercept[AnalysisException] {
+ CurrentTimestampExpressionBuilder.build(
+ "current_timestamp", Seq(AttributeReference("a", IntegerType)()))
+ },
+ condition = "NON_FOLDABLE_ARGUMENT",
+ parameters = Map(
+ "funcName" -> "`current_timestamp`",
+ "paramName" -> "`precision`",
+ "paramType" -> "\"INT\""))
+
+ // Non-integral precision.
+ checkError(
+ exception = intercept[AnalysisException] {
+ CurrentTimestampExpressionBuilder.build("current_timestamp",
Seq(Literal("9")))
+ },
+ condition = "UNEXPECTED_INPUT_TYPE",
+ parameters = Map(
+ "paramIndex" -> "first",
+ "functionName" -> "`current_timestamp`",
+ "requiredType" -> "\"INT\"",
+ "inputSql" -> "\"9\"",
+ "inputType" -> "\"STRING\""))
+
+ // Too many arguments.
+ checkError(
+ exception = intercept[AnalysisException] {
+ CurrentTimestampExpressionBuilder.build("current_timestamp",
Seq(Literal(9), Literal(9)))
+ },
+ condition = "WRONG_NUM_ARGS.WITHOUT_SUGGESTION",
+ parameters = Map(
+ "functionName" -> "`current_timestamp`",
+ "expectedNum" -> "[0, 1]",
+ "actualNum" -> "2",
+ "docroot" -> SPARK_DOC_ROOT))
+ }
+
+ test("SPARK-57837: LocalTimestampExpressionBuilder") {
+ assert(LocalTimestampExpressionBuilder.build("localtimestamp", Seq.empty)
=== LocalTimestamp())
+ assert(LocalTimestampExpressionBuilder.build("localtimestamp",
Seq(Literal(6))) ===
+ LocalTimestamp())
+
+ Seq(7, 8, 9).foreach { p =>
+ val built = LocalTimestampExpressionBuilder.build("localtimestamp",
Seq(Literal(p)))
+ assert(built === LocalTimestampNanos(p))
+ assert(built.dataType === TimestampNTZNanosType(p))
+ }
+
+ Seq(0, 3, 5, 10).foreach { p =>
+ checkError(
+ exception = intercept[SparkException] {
+ LocalTimestampExpressionBuilder.build("localtimestamp",
Seq(Literal(p)))
+ },
+ condition = "INVALID_TIMESTAMP_PRECISION",
+ parameters = Map("precision" -> p.toString, "type" -> "TIMESTAMP_NTZ"))
+ }
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ LocalTimestampExpressionBuilder.build("localtimestamp",
Seq(Literal(9), Literal(9)))
+ },
+ condition = "WRONG_NUM_ARGS.WITHOUT_SUGGESTION",
+ parameters = Map(
+ "functionName" -> "`localtimestamp`",
+ "expectedNum" -> "[0, 1]",
+ "actualNum" -> "2",
+ "docroot" -> SPARK_DOC_ROOT))
+ }
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ComputeCurrentTimeSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ComputeCurrentTimeSuite.scala
index be24f9c9f01f..dc8cfd74c785 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ComputeCurrentTimeSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ComputeCurrentTimeSuite.scala
@@ -18,13 +18,13 @@
package org.apache.spark.sql.catalyst.optimizer
import java.lang.Thread.sleep
-import java.time.{LocalDateTime, ZoneId}
+import java.time.{Instant, LocalDateTime, ZoneId}
import scala.concurrent.duration._
import scala.jdk.CollectionConverters.MapHasAsScala
import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{Add, Alias, Cast,
CurrentDate, CurrentTime, CurrentTimestamp, CurrentTimeZone, Expression,
InSubquery, ListQuery, Literal, LocalTimestamp, Now}
+import org.apache.spark.sql.catalyst.expressions.{Add, Alias, Cast,
CurrentDate, CurrentTime, CurrentTimestamp, CurrentTimestampNanos,
CurrentTimeZone, Expression, InSubquery, ListQuery, Literal, LocalTimestamp,
LocalTimestampNanos, Now}
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation,
LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.trees.TreePattern
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DateType, IntegerType, StringType,
TimestampLTZNanosType, TimestampNTZNanosType, TimestampNTZType, TimestampType,
TimeType}
-import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.unsafe.types.{TimestampNanosVal, UTF8String}
class ComputeCurrentTimeSuite extends PlanTest {
object Optimize extends RuleExecutor[LogicalPlan] {
@@ -197,6 +197,70 @@ class ComputeCurrentTimeSuite extends PlanTest {
assert(lits(0) == lits(1))
}
+ test("SPARK-57837: analyzer should replace current_timestamp(p) with nanos
literals") {
+ Seq(7, 8, 9).foreach { p =>
+ val in = Project(
+ Seq(Alias(CurrentTimestampNanos(p), "a")(),
Alias(CurrentTimestampNanos(p), "b")()),
+ LocalRelation())
+
+ val min = DateTimeUtils.instantToMicros(Instant.now())
+ val plan = Optimize.execute(in.analyze).asInstanceOf[Project]
+ val max = DateTimeUtils.instantToMicros(Instant.now())
+
+ // The literals carry the nanosecond TIMESTAMP_LTZ(p) type.
+ val typedLits = literalsWithType(plan, TimestampLTZNanosType(p))
+ assert(typedLits.size == 2, s"precision $p should yield two
TIMESTAMP_LTZ($p) literals")
+ val vals = typedLits.map(_.asInstanceOf[TimestampNanosVal])
+ // All calls in one query return the same value.
+ assert(vals(0) == vals(1))
+ assert(vals(0).epochMicros >= min && vals(0).epochMicros <= max)
+ // Sub-precision digits are floored: at p == 7 the last two
nanos-within-micro digits are 0,
+ // at p == 8 the last digit is 0, at p == 9 all three are kept.
+ val step = math.pow(10, 9 - p).toInt % 1000
+ if (step != 0) {
+ assert(vals(0).nanosWithinMicro % step == 0,
+ s"nanosWithinMicro ${vals(0).nanosWithinMicro} should be floored to
precision $p")
+ }
+ }
+ }
+
+ test("SPARK-57837: analyzer should replace localtimestamp(p) with nanos
literals") {
+ Seq(7, 8, 9).foreach { p =>
+ val in = Project(
+ Seq(Alias(LocalTimestampNanos(p), "a")(),
Alias(LocalTimestampNanos(p), "b")()),
+ LocalRelation())
+
+ val plan = Optimize.execute(in.analyze).asInstanceOf[Project]
+
+ val typedLits = literalsWithType(plan, TimestampNTZNanosType(p))
+ assert(typedLits.size == 2, s"precision $p should yield two
TIMESTAMP_NTZ($p) literals")
+ val vals = typedLits.map(_.asInstanceOf[TimestampNanosVal])
+ assert(vals(0) == vals(1))
+ val step = math.pow(10, 9 - p).toInt % 1000
+ if (step != 0) {
+ assert(vals(0).nanosWithinMicro % step == 0,
+ s"nanosWithinMicro ${vals(0).nanosWithinMicro} should be floored to
precision $p")
+ }
+ }
+ }
+
+ test("SPARK-57837: nanos current-timestamp respects time flow across
analyses") {
+ val in = Project(Alias(CurrentTimestampNanos(9), "t1")() :: Nil,
LocalRelation())
+
+ val planT1 = Optimize.execute(in.analyze).asInstanceOf[Project]
+ sleep(5)
+ val planT2 = Optimize.execute(in.analyze).asInstanceOf[Project]
+
+ val t1 = literalsWithType(planT1, TimestampLTZNanosType(9))
+ .head.asInstanceOf[TimestampNanosVal]
+ val t2 = literalsWithType(planT2, TimestampLTZNanosType(9))
+ .head.asInstanceOf[TimestampNanosVal]
+
+ // A later analysis observes a strictly newer instant (each re-analysis
re-reads the clock).
+ assert(t2.epochMicros > t1.epochMicros,
+ s"Expected a newer time in the second analysis, but got t1=$t1, t2=$t2")
+ }
+
test("analyzer should use equal timestamps across subqueries") {
val timestampInSubQuery = Project(Seq(Alias(LocalTimestamp(),
"timestamp1")()), LocalRelation())
val listSubQuery = ListQuery(timestampInSubQuery)
@@ -344,6 +408,20 @@ class ComputeCurrentTimeSuite extends PlanTest {
literals
}
+ private def literalsWithType(
+ plan: LogicalPlan,
+ dataType: org.apache.spark.sql.types.DataType)
+ : scala.collection.mutable.ArrayBuffer[Any] = {
+ val buf = new scala.collection.mutable.ArrayBuffer[Any]
+ plan.transformWithSubqueries { case subQuery =>
+ subQuery.transformAllExpressions { case lit: Literal if lit.dataType ==
dataType =>
+ buf += lit.value
+ lit
+ }
+ }
+ buf
+ }
+
test("SPARK-57748: TIME->TIMESTAMP cast is rewritten even with no
CURRENT_LIKE node") {
val timeLit = Literal(0L, TimeType(6))
Seq(TimestampNTZType, TimestampType).foreach { target =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
index 4c7a8437a46f..aae0b7c7121c 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
@@ -26,7 +26,7 @@ import scala.collection.mutable.{Map => MutableMap}
import org.apache.spark.SparkEnv
import org.apache.spark.internal.LogKeys._
-import org.apache.spark.sql.catalyst.expressions.{CurrentDate,
CurrentTimestampLike, LocalTimestamp}
+import org.apache.spark.sql.catalyst.expressions.{CurrentDate,
CurrentTimestampLike, LocalTimestamp, LocalTimestampNanos}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.streaming.{StreamingRelationV2,
WriteToStream}
import org.apache.spark.sql.catalyst.trees.TreePattern.CURRENT_LIKE
@@ -234,7 +234,8 @@ class ContinuousExecution(
}
withNewSources.transformAllExpressionsWithPruning(_.containsPattern(CURRENT_LIKE))
{
- case (_: CurrentTimestampLike | _: CurrentDate | _: LocalTimestamp) =>
+ case (_: CurrentTimestampLike | _: CurrentDate | _: LocalTimestamp |
+ _: LocalTimestampNanos) =>
throw new IllegalStateException("CurrentTimestamp, Now, CurrentDate
and LocalTimestamp" +
" not yet supported for continuous processing")
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
index 39fb0038ad4a..8621271768f4 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
@@ -31,7 +31,7 @@ import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
import org.apache.spark.sql.catalyst.analysis.{ResolveDeduplicate,
V2TableReference}
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
CurrentBatchTimestamp, CurrentDate, CurrentTimestamp,
FileSourceMetadataAttribute, LocalTimestamp}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
CurrentBatchTimestamp, CurrentDate, CurrentTimestamp, CurrentTimestampNanos,
FileSourceMetadataAttribute, LocalTimestamp, LocalTimestampNanos}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Deduplicate,
DeduplicateWithinWatermark, Distinct, FlatMapGroupsInPandasWithState,
FlatMapGroupsWithState, GlobalLimit, Join, LeafNode, LocalRelation,
LogicalPlan, Project, StreamSourceAwareLogicalPlan, TransformWithState,
TransformWithStateInPySpark}
import org.apache.spark.sql.catalyst.streaming.{StreamingRelationV2,
Unassigned, WriteToStream}
import org.apache.spark.sql.catalyst.trees.TreePattern.CURRENT_LIKE
@@ -1215,9 +1215,19 @@ class MicroBatchExecution(
// dummy string to prevent UnresolvedException and to prevent to be
used in the future.
CurrentBatchTimestamp(execCtx.offsetSeqMetadata.batchTimestampMs,
ct.dataType, Some("Dummy TimeZoneId"))
+ case ct: CurrentTimestampNanos =>
+ // Like CurrentTimestamp, the nanosecond current_timestamp(p) is not
+ // TimeZoneAwareExpression, so supply the dummy time zone. The batch
timestamp is
+ // millisecond resolution, so the folded TIMESTAMP_LTZ(p) literal has
zero
+ // nanos-within-micro.
+ CurrentBatchTimestamp(execCtx.offsetSeqMetadata.batchTimestampMs,
+ ct.dataType, Some("Dummy TimeZoneId"))
case lt: LocalTimestamp =>
CurrentBatchTimestamp(execCtx.offsetSeqMetadata.batchTimestampMs,
lt.dataType, lt.timeZoneId)
+ case lt: LocalTimestampNanos =>
+ CurrentBatchTimestamp(execCtx.offsetSeqMetadata.batchTimestampMs,
+ lt.dataType, lt.timeZoneId)
case cd: CurrentDate =>
CurrentBatchTimestamp(execCtx.offsetSeqMetadata.batchTimestampMs,
cd.dataType, cd.timeZoneId)
diff --git a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
index 10f68f708b1b..815408f225e7 100644
--- a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
+++ b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
@@ -113,7 +113,7 @@
| org.apache.spark.sql.catalyst.expressions.CurrentTime | current_time |
SELECT current_time() | struct<current_time(6):time(6)> |
| org.apache.spark.sql.catalyst.expressions.CurrentTime | localtime | SELECT
localtime() | struct<current_time(6):time(6)> |
| org.apache.spark.sql.catalyst.expressions.CurrentTimeZone | current_timezone
| SELECT current_timezone() | struct<current_timezone():string> |
-| org.apache.spark.sql.catalyst.expressions.CurrentTimestamp |
current_timestamp | SELECT current_timestamp() |
struct<current_timestamp():timestamp> |
+| org.apache.spark.sql.catalyst.expressions.CurrentTimestampExpressionBuilder
| current_timestamp | SELECT current_timestamp() |
struct<current_timestamp():timestamp> |
| org.apache.spark.sql.catalyst.expressions.CurrentUser | current_user |
SELECT current_user() | struct<current_user():string> |
| org.apache.spark.sql.catalyst.expressions.CurrentUser | session_user |
SELECT session_user() | struct<session_user():string> |
| org.apache.spark.sql.catalyst.expressions.CurrentUser | user | SELECT user()
| struct<user():string> |
@@ -218,7 +218,7 @@
| org.apache.spark.sql.catalyst.expressions.LessThanOrEqual | <= | SELECT 2 <=
2 | struct<(2 <= 2):boolean> |
| org.apache.spark.sql.catalyst.expressions.Levenshtein | levenshtein | SELECT
levenshtein('kitten', 'sitting') | struct<levenshtein(kitten, sitting):int> |
| org.apache.spark.sql.catalyst.expressions.Like | like | SELECT like('Spark',
'_park') | struct<Spark LIKE _park:boolean> |
-| org.apache.spark.sql.catalyst.expressions.LocalTimestamp | localtimestamp |
SELECT localtimestamp() | struct<localtimestamp():timestamp_ntz> |
+| org.apache.spark.sql.catalyst.expressions.LocalTimestampExpressionBuilder |
localtimestamp | SELECT localtimestamp() |
struct<localtimestamp():timestamp_ntz> |
| org.apache.spark.sql.catalyst.expressions.Log | ln | SELECT ln(1) |
struct<ln(1):double> |
| org.apache.spark.sql.catalyst.expressions.Log10 | log10 | SELECT log10(10) |
struct<LOG10(10):double> |
| org.apache.spark.sql.catalyst.expressions.Log1p | log1p | SELECT log1p(0) |
struct<LOG1P(0):double> |
@@ -262,7 +262,7 @@
| org.apache.spark.sql.catalyst.expressions.NextDay | next_day | SELECT
next_day('2015-01-14', 'TU') | struct<next_day(2015-01-14, TU):date> |
| org.apache.spark.sql.catalyst.expressions.Not | ! | SELECT ! true |
struct<(NOT true):boolean> |
| org.apache.spark.sql.catalyst.expressions.Not | not | SELECT not true |
struct<(NOT true):boolean> |
-| org.apache.spark.sql.catalyst.expressions.Now | now | SELECT now() |
struct<now():timestamp> |
+| org.apache.spark.sql.catalyst.expressions.NowExpressionBuilder | now |
SELECT now() | struct<now():timestamp> |
| org.apache.spark.sql.catalyst.expressions.NthValue | nth_value | SELECT a,
b, nth_value(b, 2) OVER (PARTITION BY a ORDER BY b) FROM VALUES ('A1', 2),
('A1', 1), ('A2', 3), ('A1', 1) tab(a, b) | struct<a:string,b:int,nth_value(b,
2) OVER (PARTITION BY a ORDER BY b ASC NULLS FIRST RANGE BETWEEN UNBOUNDED
PRECEDING AND CURRENT ROW):int> |
| org.apache.spark.sql.catalyst.expressions.NullIf | nullif | SELECT nullif(2,
2) | struct<nullif(2, 2):int> |
| org.apache.spark.sql.catalyst.expressions.NullIfZero | nullifzero | SELECT
nullifzero(0) | struct<nullifzero(0):int> |
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
index 7664ff3b273d..69b136fc0caa 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out
@@ -1246,6 +1246,64 @@ Project [date_trunc(NANOSECOND, 2020-01-01
04:34:56.123456789, Some(America/Los_
+- OneRowRelation
+-- !query
+SELECT typeof(current_timestamp(9)), typeof(current_timestamp(8)),
typeof(current_timestamp(7))
+-- !query analysis
+[Analyzer test output redacted due to nondeterminism]
+
+
+-- !query
+SELECT typeof(now(9)), typeof(now(6))
+-- !query analysis
+[Analyzer test output redacted due to nondeterminism]
+
+
+-- !query
+SELECT typeof(current_timestamp()), typeof(current_timestamp(6))
+-- !query analysis
+[Analyzer test output redacted due to nondeterminism]
+
+
+-- !query
+SELECT typeof(current_timestamp(7 + 2))
+-- !query analysis
+[Analyzer test output redacted due to nondeterminism]
+
+
+-- !query
+SELECT current_timestamp(9) = current_timestamp(9), now(9) =
current_timestamp(9)
+-- !query analysis
+[Analyzer test output redacted due to nondeterminism]
+
+
+-- !query
+SELECT current_timestamp(3)
+-- !query analysis
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "3",
+ "type" : "TIMESTAMP_LTZ"
+ }
+}
+
+
+-- !query
+SELECT current_timestamp(10)
+-- !query analysis
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "10",
+ "type" : "TIMESTAMP_LTZ"
+ }
+}
+
+
-- !query
SELECT c FROM (SELECT TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC' AS c
UNION ALL SELECT TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC')
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
index a610d8a3570a..bfe32a1f2b9c 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out
@@ -1073,6 +1073,62 @@ Project [date_trunc(NANOSECOND, 2020-01-01
12:34:56.123456789, Some(America/Los_
+- OneRowRelation
+-- !query
+SELECT typeof(localtimestamp(9)), typeof(localtimestamp(8)),
typeof(localtimestamp(7))
+-- !query analysis
+Project [typeof(localtimestamp(9, Some(America/Los_Angeles))) AS
typeof(localtimestamp())#x, typeof(localtimestamp(8,
Some(America/Los_Angeles))) AS typeof(localtimestamp())#x,
typeof(localtimestamp(7, Some(America/Los_Angeles))) AS
typeof(localtimestamp())#x]
++- OneRowRelation
+
+
+-- !query
+SELECT typeof(localtimestamp()), typeof(localtimestamp(6))
+-- !query analysis
+Project [typeof(localtimestamp(Some(America/Los_Angeles))) AS
typeof(localtimestamp())#x, typeof(localtimestamp(Some(America/Los_Angeles)))
AS typeof(localtimestamp())#x]
++- OneRowRelation
+
+
+-- !query
+SELECT typeof(localtimestamp(8 + 1))
+-- !query analysis
+Project [typeof(localtimestamp(9, Some(America/Los_Angeles))) AS
typeof(localtimestamp())#x]
++- OneRowRelation
+
+
+-- !query
+SELECT localtimestamp(9) = localtimestamp(9)
+-- !query analysis
+Project [(localtimestamp(9, Some(America/Los_Angeles)) = localtimestamp(9,
Some(America/Los_Angeles))) AS (localtimestamp() = localtimestamp())#x]
++- OneRowRelation
+
+
+-- !query
+SELECT localtimestamp(3)
+-- !query analysis
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "3",
+ "type" : "TIMESTAMP_NTZ"
+ }
+}
+
+
+-- !query
+SELECT localtimestamp(10)
+-- !query analysis
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "10",
+ "type" : "TIMESTAMP_NTZ"
+ }
+}
+
+
-- !query
SELECT c FROM (SELECT TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001' AS c
UNION ALL SELECT TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999')
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
index 93ffae607f10..385b242749e7 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql
@@ -409,6 +409,20 @@ SELECT date_trunc('DAY', '2020-01-01 04:00:00.000000123
UTC' :: timestamp_ltz(7)
-- An unsupported (sub-microsecond) unit yields NULL; the result still carries
the nanos type.
SELECT date_trunc('NANOSECOND', TIMESTAMP_LTZ '2020-01-01 12:34:56.123456789
UTC');
+-- SPARK-57837: current_timestamp(p) / now(p) with a nanosecond precision
return TIMESTAMP_LTZ(p).
+-- The values are non-deterministic, so only the (deterministic) result type
and query-stable
+-- self-equality are checked. Precision 6 keeps the standard microsecond
TIMESTAMP.
+SELECT typeof(current_timestamp(9)), typeof(current_timestamp(8)),
typeof(current_timestamp(7));
+SELECT typeof(now(9)), typeof(now(6));
+SELECT typeof(current_timestamp()), typeof(current_timestamp(6));
+-- A foldable (constant) precision expression is accepted.
+SELECT typeof(current_timestamp(7 + 2));
+-- All references to current_timestamp(p) within a query see the same value.
+SELECT current_timestamp(9) = current_timestamp(9), now(9) =
current_timestamp(9);
+-- Out-of-range precision is rejected.
+SELECT current_timestamp(3);
+SELECT current_timestamp(10);
+
-- SPARK-57841: end-to-end coverage for operators that ride on the resolved
widening (SPARK-57454)
-- and complex-type access over nanosecond values, mirroring
timestamp-ntz-nanos.sql for the LTZ
-- family. Every case turns on the SUB-MICROSECOND remainder or on
cross-precision widening. LTZ is
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
index 3896f9da4d3f..5a458a6b265e 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql
@@ -342,6 +342,19 @@ SELECT date_trunc('DAY', '2020-06-21 23:30:00.000000123'
:: timestamp_ntz(7));
-- An unsupported (sub-microsecond) unit yields NULL; the result still carries
the nanos type.
SELECT date_trunc('NANOSECOND', TIMESTAMP_NTZ '2020-01-01 12:34:56.123456789');
+-- SPARK-57837: localtimestamp(p) with a nanosecond precision returns
TIMESTAMP_NTZ(p). The values
+-- are non-deterministic, so only the (deterministic) result type and
query-stable self-equality
+-- are checked. Precision 6 keeps the standard microsecond TIMESTAMP_NTZ.
+SELECT typeof(localtimestamp(9)), typeof(localtimestamp(8)),
typeof(localtimestamp(7));
+SELECT typeof(localtimestamp()), typeof(localtimestamp(6));
+-- A foldable (constant) precision expression is accepted.
+SELECT typeof(localtimestamp(8 + 1));
+-- All references to localtimestamp(p) within a query see the same value.
+SELECT localtimestamp(9) = localtimestamp(9);
+-- Out-of-range precision is rejected.
+SELECT localtimestamp(3);
+SELECT localtimestamp(10);
+
-- SPARK-57841: end-to-end coverage for operators that ride on the resolved
widening (SPARK-57454)
-- and complex-type access over nanosecond values. Every case turns on the
SUB-MICROSECOND remainder
-- (.000000001 vs .000000999 share a microsecond; only the full nanos value
tells them apart) or on
diff --git
a/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
index f2bb95ff833e..3ae6f96251cc 100644
--- a/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out
@@ -1367,6 +1367,78 @@ struct<date_trunc(NANOSECOND, TIMESTAMP_LTZ '2020-01-01
04:34:56.123456789'):tim
NULL
+-- !query
+SELECT typeof(current_timestamp(9)), typeof(current_timestamp(8)),
typeof(current_timestamp(7))
+-- !query schema
+struct<typeof(current_timestamp()):string,typeof(current_timestamp()):string,typeof(current_timestamp()):string>
+-- !query output
+timestamp_ltz(9) timestamp_ltz(8) timestamp_ltz(7)
+
+
+-- !query
+SELECT typeof(now(9)), typeof(now(6))
+-- !query schema
+struct<typeof(current_timestamp()):string,typeof(now()):string>
+-- !query output
+timestamp_ltz(9) timestamp
+
+
+-- !query
+SELECT typeof(current_timestamp()), typeof(current_timestamp(6))
+-- !query schema
+struct<typeof(current_timestamp()):string,typeof(current_timestamp()):string>
+-- !query output
+timestamp timestamp
+
+
+-- !query
+SELECT typeof(current_timestamp(7 + 2))
+-- !query schema
+struct<typeof(current_timestamp()):string>
+-- !query output
+timestamp_ltz(9)
+
+
+-- !query
+SELECT current_timestamp(9) = current_timestamp(9), now(9) =
current_timestamp(9)
+-- !query schema
+struct<(current_timestamp() =
current_timestamp()):boolean,(current_timestamp() =
current_timestamp()):boolean>
+-- !query output
+true true
+
+
+-- !query
+SELECT current_timestamp(3)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "3",
+ "type" : "TIMESTAMP_LTZ"
+ }
+}
+
+
+-- !query
+SELECT current_timestamp(10)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "10",
+ "type" : "TIMESTAMP_LTZ"
+ }
+}
+
+
-- !query
SELECT c FROM (SELECT TIMESTAMP_LTZ '2020-01-01 00:00:00.000000001 UTC' AS c
UNION ALL SELECT TIMESTAMP_LTZ '2020-01-01 00:00:00.000000999 UTC')
diff --git
a/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
b/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
index dd55757374bd..dd16231ba635 100644
--- a/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out
@@ -1156,6 +1156,70 @@ struct<date_trunc(NANOSECOND, TIMESTAMP_NTZ '2020-01-01
12:34:56.123456789'):tim
NULL
+-- !query
+SELECT typeof(localtimestamp(9)), typeof(localtimestamp(8)),
typeof(localtimestamp(7))
+-- !query schema
+struct<typeof(localtimestamp()):string,typeof(localtimestamp()):string,typeof(localtimestamp()):string>
+-- !query output
+timestamp_ntz(9) timestamp_ntz(8) timestamp_ntz(7)
+
+
+-- !query
+SELECT typeof(localtimestamp()), typeof(localtimestamp(6))
+-- !query schema
+struct<typeof(localtimestamp()):string,typeof(localtimestamp()):string>
+-- !query output
+timestamp_ntz timestamp_ntz
+
+
+-- !query
+SELECT typeof(localtimestamp(8 + 1))
+-- !query schema
+struct<typeof(localtimestamp()):string>
+-- !query output
+timestamp_ntz(9)
+
+
+-- !query
+SELECT localtimestamp(9) = localtimestamp(9)
+-- !query schema
+struct<(localtimestamp() = localtimestamp()):boolean>
+-- !query output
+true
+
+
+-- !query
+SELECT localtimestamp(3)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "3",
+ "type" : "TIMESTAMP_NTZ"
+ }
+}
+
+
+-- !query
+SELECT localtimestamp(10)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkException
+{
+ "errorClass" : "INVALID_TIMESTAMP_PRECISION",
+ "sqlState" : "22023",
+ "messageParameters" : {
+ "precision" : "10",
+ "type" : "TIMESTAMP_NTZ"
+ }
+}
+
+
-- !query
SELECT c FROM (SELECT TIMESTAMP_NTZ '2020-01-01 00:00:00.000000001' AS c
UNION ALL SELECT TIMESTAMP_NTZ '2020-01-01 00:00:00.000000999')
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosFunctionsSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosFunctionsSuiteBase.scala
index adfeb902cc03..3f15cb2ef3cd 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosFunctionsSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosFunctionsSuiteBase.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql
import java.time.{Instant, LocalDateTime}
-import org.apache.spark.{SparkConf, SparkRuntimeException,
SparkUnsupportedOperationException}
+import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException,
SparkUnsupportedOperationException}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -824,6 +824,88 @@ abstract class TimestampNanosFunctionsSuiteBase extends
SharedSparkSession {
}
}
}
+
+ test("SPARK-57837: current_timestamp(p) / now(p) return TIMESTAMP_LTZ(p)") {
+ val df = spark.range(1)
+ Seq("current_timestamp", "now").foreach { fn =>
+ Seq(7, 8, 9).foreach { p =>
+ val schema = df.selectExpr(s"$fn($p) AS c").schema
+ assert(schema.head.dataType === TimestampLTZNanosType(p),
+ s"$fn($p) should be TIMESTAMP_LTZ($p)")
+ }
+ // Precision 6 and the no-arg form keep the micro TIMESTAMP type.
+ assert(df.selectExpr(s"$fn(6) AS c").schema.head.dataType ===
TimestampType)
+ assert(df.selectExpr(s"$fn() AS c").schema.head.dataType ===
TimestampType)
+ }
+ }
+
+ test("SPARK-57837: localtimestamp(p) returns TIMESTAMP_NTZ(p)") {
+ val df = spark.range(1)
+ Seq(7, 8, 9).foreach { p =>
+ val schema = df.selectExpr(s"localtimestamp($p) AS c").schema
+ assert(schema.head.dataType === TimestampNTZNanosType(p),
+ s"localtimestamp($p) should be TIMESTAMP_NTZ($p)")
+ }
+ assert(df.selectExpr("localtimestamp(6) AS c").schema.head.dataType ===
TimestampNTZType)
+ assert(df.selectExpr("localtimestamp() AS c").schema.head.dataType ===
TimestampNTZType)
+ }
+
+ test("SPARK-57837: nanos current-timestamp values are query-stable and
floored to precision") {
+ val df = spark.range(1)
+ // All references within a query see the same value.
+ checkAnswer(
+ df.selectExpr(
+ "current_timestamp(9) = current_timestamp(9)",
+ "now(9) = current_timestamp(9)",
+ "localtimestamp(9) = localtimestamp(9)"),
+ Row(true, true, true))
+
+ // Sub-precision digits are floored: the extracted DECIMAL(11, 9) second
has zeros below `p`.
+ Seq(7, 8, 9).foreach { p =>
+ val secExpr = s"extract(SECOND FROM current_timestamp($p))"
+ val secondDecimal = df.selectExpr(secExpr).head().getDecimal(0)
+ val scaled = secondDecimal.movePointRight(9).longValueExact()
+ val step = math.pow(10, 9 - p).toLong
+ assert(scaled % step == 0,
+ s"extract(SECOND) of current_timestamp($p) = $secondDecimal not
floored to precision $p")
+ }
+ }
+
+ test("SPARK-57837: current_timestamp(p) / localtimestamp(p) reject
out-of-range precision") {
+ val df = spark.range(1)
+ Seq(
+ "current_timestamp(3)" -> "TIMESTAMP_LTZ",
+ "now(3)" -> "TIMESTAMP_LTZ",
+ "localtimestamp(10)" -> "TIMESTAMP_NTZ").foreach { case (expr, typeName)
=>
+ val precision = expr.replaceAll("\\D", "")
+ checkError(
+ exception = intercept[SparkException] {
+ df.selectExpr(expr).collect()
+ },
+ condition = "INVALID_TIMESTAMP_PRECISION",
+ parameters = Map("precision" -> precision, "type" -> typeName))
+ }
+ }
+
+ test("SPARK-57837: current_timestamp(p) / localtimestamp(p) require the
nanos preview flag") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "false") {
+ val df = spark.range(1)
+ Seq("current_timestamp(9)", "now(8)", "localtimestamp(7)").foreach {
expr =>
+ checkError(
+ exception = intercept[SparkException] {
+ df.selectExpr(expr).collect()
+ },
+ condition = "FEATURE_NOT_ENABLED",
+ parameters = Map(
+ "featureName" -> "Nanosecond-precision timestamp types",
+ "configKey" -> "spark.sql.timestampNanosTypes.enabled",
+ "configValue" -> "true"))
+ }
+ // With the flag off, precision 6 and the no-arg forms still work (micro
types).
+ assert(df.selectExpr("current_timestamp(6) AS c").schema.head.dataType
=== TimestampType)
+ assert(df.selectExpr("localtimestamp() AS c").schema.head.dataType ===
TimestampNTZType)
+ }
+ }
}
// Runs the nanosecond timestamp function tests with ANSI mode enabled
explicitly.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/expressions/ExpressionInfoSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/expressions/ExpressionInfoSuite.scala
index e61299161fdf..dab117fa6713 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/expressions/ExpressionInfoSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/expressions/ExpressionInfoSuite.scala
@@ -248,9 +248,12 @@ class ExpressionInfoSuite extends SharedSparkSession {
"org.apache.spark.sql.catalyst.expressions.CurrentDate",
"org.apache.spark.sql.catalyst.expressions.CurDateExpressionBuilder",
"org.apache.spark.sql.catalyst.expressions.CurrentTimestamp",
+
"org.apache.spark.sql.catalyst.expressions.CurrentTimestampExpressionBuilder",
"org.apache.spark.sql.catalyst.expressions.CurrentTimeZone",
"org.apache.spark.sql.catalyst.expressions.Now",
+ "org.apache.spark.sql.catalyst.expressions.NowExpressionBuilder",
"org.apache.spark.sql.catalyst.expressions.LocalTimestamp",
+
"org.apache.spark.sql.catalyst.expressions.LocalTimestampExpressionBuilder",
"org.apache.spark.sql.catalyst.expressions.CurrentTime",
// Random output without a seed
"org.apache.spark.sql.catalyst.expressions.Rand",
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
index b4fd41a6b550..1bad6cf32877 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
@@ -1236,6 +1236,36 @@ class StreamSuite extends StreamTest {
testCurrentTimestampOnStreamingQuery()
}
+ test("SPARK-57837: nanosecond current_timestamp(p)/localtimestamp(p) use the
batch timestamp") {
+ // In micro-batch streaming, current_timestamp()/localtimestamp() are
rewritten to the
+ // replay-stable batch timestamp (CurrentBatchTimestamp), not folded to a
wall clock at
+ // planning time. The nanosecond variants must follow the same rewrite:
the batch timestamp is
+ // millisecond resolution, so the collected value must be
millisecond-aligned (a wall-clock
+ // fold from Instant.now() would carry sub-millisecond digits and fail
this check).
+ val input = MemoryStream[Int]
+ val df = input.toDS().selectExpr("current_timestamp(9) AS ltz",
"localtimestamp(9) AS ntz")
+
+ def assertBatchAligned(rows: Seq[Row]): Unit = {
+ assert(rows.size === 1)
+ val row = rows.head
+ // LTZ collects as java.time.Instant, NTZ as java.time.LocalDateTime.
+ val ltzNanos = row.getAs[java.time.Instant]("ltz").getNano
+ val ntzNanos = row.getAs[java.time.LocalDateTime]("ntz").getNano
+ assert(ltzNanos % 1000000 === 0,
+ s"current_timestamp(9) sub-second $ltzNanos is not
millisecond-aligned")
+ assert(ntzNanos % 1000000 === 0,
+ s"localtimestamp(9) sub-second $ntzNanos is not millisecond-aligned")
+ }
+
+ testStream(df)(
+ AddData(input, 1),
+ CheckLastBatch { rows: Seq[Row] => assertBatchAligned(rows) },
+ Execute { _ => Thread.sleep(1000) },
+ AddData(input, 2),
+ CheckLastBatch { rows: Seq[Row] => assertBatchAligned(rows) }
+ )
+ }
+
private def testCurrentTimestampOnStreamingQuery(): Unit = {
val input = MemoryStream[Int]
val df = input.toDS().withColumn("cur_timestamp", lit(current_timestamp()))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
index c70f21ae144b..4de3867fd61e 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
@@ -185,6 +185,24 @@ class ContinuousSuite extends ContinuousSuiteBase {
"Continuous processing does not support current time operations."))
}
+ test("SPARK-57837: nanosecond current-timestamp functions are rejected") {
+ // current_timestamp(p) -> CurrentTimestampNanos and localtimestamp(p) ->
LocalTimestampNanos
+ // must be rejected in continuous processing just like their microsecond
forms. The nanos
+ // types are enabled by default under Utils.isTesting.
+ Seq("current_timestamp(9)", "now(9)", "localtimestamp(9)").foreach { fn =>
+ val input = ContinuousMemoryStream[Int]
+ val df = input.toDF().selectExpr(fn)
+
+ val except = intercept[AnalysisException] {
+ testStream(df)(StartStream())
+ }
+
+ assert(except.message.contains(
+ "Continuous processing does not support current time operations."),
+ s"$fn should be rejected for continuous processing")
+ }
+ }
+
test("subquery alias") {
withTempView("memory") {
val input = ContinuousMemoryStream[Int]
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
index 662e55dac543..76d9a08f603d 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
@@ -1071,6 +1071,7 @@ private[hive] trait HiveInspectors {
case _: CurrentTime => false
case _: CurrentTimestampLike => false
case _: LocalTimestamp => false
+ case _: LocalTimestampNanos => false
case _ => e.children.forall(canEarlyEval)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
