Use Substrait timestamp_tz for Spark TimestampType to preserve timezone-aware semantics (#11074)

liujiayi771 · web-flow · commit 8399ac767655 · 2025-11-15T15:52:10.000-08:00
Spark’s TimestampType is timezone-aware: it internally stores timestamps in UTC (by converting input values to UTC based on the session time zone or just read UTC timestamp from parquet file) and represents an absolute point in time. This semantics aligns with Substrait’s timestamp_tz type, which also denotes a timezone-aware timestamp that can be unambiguously mapped to a moment on the timeline.

To maintain semantic consistency between Spark and Substrait, this PR maps Spark’s TimestampType to Substrait’s timestamp_tz.
diff --git a/cpp-ch/local-engine/Builder/SerializedPlanBuilder.cpp b/cpp-ch/local-engine/Builder/SerializedPlanBuilder.cpp
@@ -107,7 +107,7 @@ SchemaPtr SerializedSchemaBuilder::build()
         else if (type == "Timestamp")
         {
             auto * t = type_struct->mutable_types()->Add();
-            t->mutable_timestamp()->set_nullability(
+            t->mutable_timestamp_tz()->set_nullability(
                 this->nullability_map[name] ? substrait::Type_Nullability_NULLABILITY_NULLABLE
                                             : substrait::Type_Nullability_NULLABILITY_REQUIRED);
         }
@@ -256,7 +256,7 @@ std::shared_ptr<substrait::Type> SerializedPlanBuilder::buildType(const DB::Data
         const auto * ch_type_datetime64 = checkAndGetDataType<DataTypeDateTime64>(ch_type_without_nullable.get());
         if (ch_type_datetime64->getScale() != 6)
             throw Exception(ErrorCodes::UNKNOWN_TYPE, "Spark doesn't support converting from {}", ch_type->getName());
-        res->mutable_timestamp()->set_nullability(type_nullability);
+        res->mutable_timestamp_tz()->set_nullability(type_nullability);
     }
     else if (which.isDate32())
         res->mutable_date()->set_nullability(type_nullability);
@@ -365,7 +365,7 @@ substrait::Expression * literalTimestamp(int64_t value)
 {
     substrait::Expression * rel = new substrait::Expression();
     auto * literal = rel->mutable_literal();
-    literal->set_timestamp(value);
+    literal->set_timestamp_tz(value);
     return rel;
 }
 
diff --git a/cpp-ch/local-engine/Parser/ExpressionParser.cpp b/cpp-ch/local-engine/Parser/ExpressionParser.cpp
@@ -112,9 +112,9 @@ std::pair<DB::DataTypePtr, DB::Field> LiteralParser::parse(const substrait::Expr
             field = literal.date();
             break;
         }
-        case substrait::Expression_Literal::kTimestamp: {
+        case substrait::Expression_Literal::kTimestampTz: {
             type = std::make_shared<DB::DataTypeDateTime64>(6);
-            field = DecimalField<DB::DateTime64>(literal.timestamp(), 6);
+            field = DecimalField<DB::DateTime64>(literal.timestamp_tz(), 6);
             break;
         }
         case substrait::Expression_Literal::kDecimal: {
diff --git a/cpp-ch/local-engine/Parser/TypeParser.cpp b/cpp-ch/local-engine/Parser/TypeParser.cpp
@@ -153,10 +153,10 @@ DB::DataTypePtr TypeParser::parseType(const substrait::Type & substrait_type, st
         ch_type = std::make_shared<DB::DataTypeFloat64>();
         ch_type = tryWrapNullable(substrait_type.fp64().nullability(), ch_type);
     }
-    else if (substrait_type.has_timestamp())
+    else if (substrait_type.has_timestamp_tz())
     {
         ch_type = std::make_shared<DB::DataTypeDateTime64>(6);
-        ch_type = tryWrapNullable(substrait_type.timestamp().nullability(), ch_type);
+        ch_type = tryWrapNullable(substrait_type.timestamp_tz().nullability(), ch_type);
     }
     else if (substrait_type.has_date())
     {
diff --git a/cpp/velox/substrait/SubstraitParser.cc b/cpp/velox/substrait/SubstraitParser.cc
@@ -76,7 +76,7 @@ TypePtr SubstraitParser::parseType(const ::substrait::Type& substraitType, bool
       return UNKNOWN();
     case ::substrait::Type::KindCase::kDate:
       return DATE();
-    case ::substrait::Type::KindCase::kTimestamp:
+    case ::substrait::Type::KindCase::kTimestampTz:
       return TIMESTAMP();
     case ::substrait::Type::KindCase::kDecimal: {
       auto precision = substraitType.decimal().precision();
@@ -368,7 +368,7 @@ bool SubstraitParser::getLiteralValue(const ::substrait::Expression::Literal& li
 
 template <>
 Timestamp SubstraitParser::getLiteralValue(const ::substrait::Expression::Literal& literal) {
-  return Timestamp::fromMicros(literal.timestamp());
+  return Timestamp::fromMicros(literal.timestamp_tz());
 }
 
 template <>
diff --git a/cpp/velox/substrait/SubstraitToVeloxExpr.cc b/cpp/velox/substrait/SubstraitToVeloxExpr.cc
@@ -131,7 +131,7 @@ TypePtr getScalarType(const ::substrait::Expression::Literal& literal) {
     }
     case ::substrait::Expression_Literal::LiteralTypeCase::kDate:
       return DATE();
-    case ::substrait::Expression_Literal::LiteralTypeCase::kTimestamp:
+    case ::substrait::Expression_Literal::LiteralTypeCase::kTimestampTz:
       return TIMESTAMP();
     case ::substrait::Expression_Literal::LiteralTypeCase::kString:
       return VARCHAR();
diff --git a/cpp/velox/substrait/VeloxToSubstraitExpr.cc b/cpp/velox/substrait/VeloxToSubstraitExpr.cc
@@ -136,7 +136,7 @@ const ::substrait::Expression_Literal& toSubstraitNotNullLiteral(
     case velox::TypeKind::TIMESTAMP: {
       auto vTimeStamp = variantValue.value<TypeKind::TIMESTAMP>();
       auto micros = vTimeStamp.getSeconds() * 1000000 + vTimeStamp.getNanos() / 1000;
-      literalExpr->set_timestamp(micros);
+      literalExpr->set_timestamp_tz(micros);
       break;
     }
     case velox::TypeKind::VARCHAR: {
@@ -250,7 +250,7 @@ const ::substrait::Expression_Literal& toSubstraitNotNullLiteral<TypeKind::TIMES
   ::substrait::Expression_Literal* literalExpr =
       google::protobuf::Arena::CreateMessage<::substrait::Expression_Literal>(&arena);
   auto micros = value.getSeconds() * 1000000 + value.getNanos() / 1000;
-  literalExpr->set_timestamp(micros);
+  literalExpr->set_timestamp_tz(micros);
   literalExpr->set_nullable(false);
   return *literalExpr;
 }
diff --git a/cpp/velox/substrait/VeloxToSubstraitType.cc b/cpp/velox/substrait/VeloxToSubstraitType.cc
@@ -88,9 +88,9 @@ const ::substrait::Type& VeloxToSubstraitTypeConvertor::toSubstraitType(
       break;
     }
     case velox::TypeKind::TIMESTAMP: {
-      auto substraitTimestamp = google::protobuf::Arena::CreateMessage<::substrait::Type_Timestamp>(&arena);
-      substraitTimestamp->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
-      substraitType->set_allocated_timestamp(substraitTimestamp);
+      auto substraitTimestampTZ = google::protobuf::Arena::CreateMessage<::substrait::Type_TimestampTZ>(&arena);
+      substraitTimestampTZ->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+      substraitType->set_allocated_timestamp_tz(substraitTimestampTZ);
       break;
     }
     case velox::TypeKind::ARRAY: {
diff --git a/gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/TimestampLiteralNode.java b/gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/TimestampLiteralNode.java
@@ -32,6 +32,6 @@ public TimestampLiteralNode(Long value, TypeNode typeNode) {
 
   @Override
   protected void updateLiteralBuilder(Builder literalBuilder, Long value) {
-    literalBuilder.setTimestamp(value);
+    literalBuilder.setTimestampTz(value);
   }
 }
diff --git a/gluten-substrait/src/main/java/org/apache/gluten/substrait/type/TimestampTypeNode.java b/gluten-substrait/src/main/java/org/apache/gluten/substrait/type/TimestampTypeNode.java
@@ -26,15 +26,15 @@ public TimestampTypeNode(Boolean nullable) {
 
   @Override
   public Type toProtobuf() {
-    Type.Timestamp.Builder timestampBuilder = Type.Timestamp.newBuilder();
+    Type.TimestampTZ.Builder timestampBuilder = Type.TimestampTZ.newBuilder();
     if (nullable) {
       timestampBuilder.setNullability(Type.Nullability.NULLABILITY_NULLABLE);
     } else {
       timestampBuilder.setNullability(Type.Nullability.NULLABILITY_REQUIRED);
     }
 
     Type.Builder builder = Type.newBuilder();
-    builder.setTimestamp(timestampBuilder.build());
+    builder.setTimestampTz(timestampBuilder.build());
     return builder.build();
   }
 }
diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ConverterUtils.scala b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ConverterUtils.scala
@@ -160,8 +160,8 @@ object ConverterUtils extends Logging {
         (StringType, isNullable(substraitType.getString.getNullability))
       case Type.KindCase.BINARY =>
         (BinaryType, isNullable(substraitType.getBinary.getNullability))
-      case Type.KindCase.TIMESTAMP =>
-        (TimestampType, isNullable(substraitType.getTimestamp.getNullability))
+      case Type.KindCase.TIMESTAMP_TZ =>
+        (TimestampType, isNullable(substraitType.getTimestampTz.getNullability))
       case Type.KindCase.DATE =>
         (DateType, isNullable(substraitType.getDate.getNullability))
       case Type.KindCase.DECIMAL =>

Original file line number	Diff line number	Diff line change
`@@ -107,7 +107,7 @@ SchemaPtr SerializedSchemaBuilder::build()`
`107`	`107`	`else if (type == "Timestamp")`
`108`	`108`	`{`
`109`	`109`	`auto * t = type_struct->mutable_types()->Add();`
`110`		`- t->mutable_timestamp()->set_nullability(`
	`110`	`+ t->mutable_timestamp_tz()->set_nullability(`
`111`	`111`	`this->nullability_map[name] ? substrait::Type_Nullability_NULLABILITY_NULLABLE`
`112`	`112`	`: substrait::Type_Nullability_NULLABILITY_REQUIRED);`
`113`	`113`	`}`
`@@ -256,7 +256,7 @@ std::shared_ptr<substrait::Type> SerializedPlanBuilder::buildType(const DB::Data`
`256`	`256`	`const auto * ch_type_datetime64 = checkAndGetDataType<DataTypeDateTime64>(ch_type_without_nullable.get());`
`257`	`257`	`if (ch_type_datetime64->getScale() != 6)`
`258`	`258`	`throw Exception(ErrorCodes::UNKNOWN_TYPE, "Spark doesn't support converting from {}", ch_type->getName());`
`259`		`- res->mutable_timestamp()->set_nullability(type_nullability);`
	`259`	`+ res->mutable_timestamp_tz()->set_nullability(type_nullability);`
`260`	`260`	`}`
`261`	`261`	`else if (which.isDate32())`
`262`	`262`	`res->mutable_date()->set_nullability(type_nullability);`
`@@ -365,7 +365,7 @@ substrait::Expression * literalTimestamp(int64_t value)`
`365`	`365`	`{`
`366`	`366`	`substrait::Expression * rel = new substrait::Expression();`
`367`	`367`	`auto * literal = rel->mutable_literal();`
`368`		`- literal->set_timestamp(value);`
	`368`	`+ literal->set_timestamp_tz(value);`
`369`	`369`	`return rel;`
`370`	`370`	`}`
`371`	`371`
Original file line number	Diff line number	Diff line change
`@@ -112,9 +112,9 @@ std::pair<DB::DataTypePtr, DB::Field> LiteralParser::parse(const substrait::Expr`
`112`	`112`	`field = literal.date();`
`113`	`113`	`break;`
`114`	`114`	`}`
`115`		`- case substrait::Expression_Literal::kTimestamp: {`
	`115`	`+ case substrait::Expression_Literal::kTimestampTz: {`
`116`	`116`	`type = std::make_shared<DB::DataTypeDateTime64>(6);`
`117`		`- field = DecimalField<DB::DateTime64>(literal.timestamp(), 6);`
	`117`	`+ field = DecimalField<DB::DateTime64>(literal.timestamp_tz(), 6);`
`118`	`118`	`break;`
`119`	`119`	`}`
`120`	`120`	`case substrait::Expression_Literal::kDecimal: {`
Original file line number	Diff line number	Diff line change
`@@ -153,10 +153,10 @@ DB::DataTypePtr TypeParser::parseType(const substrait::Type & substrait_type, st`
`153`	`153`	`ch_type = std::make_shared<DB::DataTypeFloat64>();`
`154`	`154`	`ch_type = tryWrapNullable(substrait_type.fp64().nullability(), ch_type);`
`155`	`155`	`}`
`156`		`- else if (substrait_type.has_timestamp())`
	`156`	`+ else if (substrait_type.has_timestamp_tz())`
`157`	`157`	`{`
`158`	`158`	`ch_type = std::make_shared<DB::DataTypeDateTime64>(6);`
`159`		`- ch_type = tryWrapNullable(substrait_type.timestamp().nullability(), ch_type);`
	`159`	`+ ch_type = tryWrapNullable(substrait_type.timestamp_tz().nullability(), ch_type);`
`160`	`160`	`}`
`161`	`161`	`else if (substrait_type.has_date())`
`162`	`162`	`{`
Original file line number	Diff line number	Diff line change
`@@ -131,7 +131,7 @@ TypePtr getScalarType(const ::substrait::Expression::Literal& literal) {`
`131`	`131`	`}`
`132`	`132`	`case ::substrait::Expression_Literal::LiteralTypeCase::kDate:`
`133`	`133`	`return DATE();`
`134`		`- case ::substrait::Expression_Literal::LiteralTypeCase::kTimestamp:`
	`134`	`+ case ::substrait::Expression_Literal::LiteralTypeCase::kTimestampTz:`
`135`	`135`	`return TIMESTAMP();`
`136`	`136`	`case ::substrait::Expression_Literal::LiteralTypeCase::kString:`
`137`	`137`	`return VARCHAR();`
Original file line number	Diff line number	Diff line change
`@@ -88,9 +88,9 @@ const ::substrait::Type& VeloxToSubstraitTypeConvertor::toSubstraitType(`
`88`	`88`	`break;`
`89`	`89`	`}`
`90`	`90`	`case velox::TypeKind::TIMESTAMP: {`
`91`		`- auto substraitTimestamp = google::protobuf::Arena::CreateMessage<::substrait::Type_Timestamp>(&arena);`
`92`		`- substraitTimestamp->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);`
`93`		`- substraitType->set_allocated_timestamp(substraitTimestamp);`
	`91`	`+ auto substraitTimestampTZ = google::protobuf::Arena::CreateMessage<::substrait::Type_TimestampTZ>(&arena);`
	`92`	`+ substraitTimestampTZ->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);`
	`93`	`+ substraitType->set_allocated_timestamp_tz(substraitTimestampTZ);`
`94`	`94`	`break;`
`95`	`95`	`}`
`96`	`96`	`case velox::TypeKind::ARRAY: {`
Original file line number	Diff line number	Diff line change
`@@ -32,6 +32,6 @@ public TimestampLiteralNode(Long value, TypeNode typeNode) {`
`32`	`32`
`33`	`33`	`@Override`
`34`	`34`	`protected void updateLiteralBuilder(Builder literalBuilder, Long value) {`
`35`		`- literalBuilder.setTimestamp(value);`
	`35`	`+ literalBuilder.setTimestampTz(value);`
`36`	`36`	`}`
`37`	`37`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,15 +26,15 @@ public TimestampTypeNode(Boolean nullable) {`
`26`	`26`
`27`	`27`	`@Override`
`28`	`28`	`public Type toProtobuf() {`
`29`		`- Type.Timestamp.Builder timestampBuilder = Type.Timestamp.newBuilder();`
	`29`	`+ Type.TimestampTZ.Builder timestampBuilder = Type.TimestampTZ.newBuilder();`
`30`	`30`	`if (nullable) {`
`31`	`31`	`timestampBuilder.setNullability(Type.Nullability.NULLABILITY_NULLABLE);`
`32`	`32`	`} else {`
`33`	`33`	`timestampBuilder.setNullability(Type.Nullability.NULLABILITY_REQUIRED);`
`34`	`34`	`}`
`35`	`35`
`36`	`36`	`Type.Builder builder = Type.newBuilder();`
`37`		`- builder.setTimestamp(timestampBuilder.build());`
	`37`	`+ builder.setTimestampTz(timestampBuilder.build());`
`38`	`38`	`return builder.build();`
`39`	`39`	`}`
`40`	`40`	`}`