This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ffc51fca16 [common] Convert variant timestamps with the requested zone
instead of the JVM default (#9939)
ffc51fca16 is described below
commit ffc51fca167599fa1997c6147ec26ae2251d771b
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Fri Sep 18 12:00:16 2026 +0800
[common] Convert variant timestamps with the requested zone instead of the
JVM default (#9939)
---
.../paimon/data/variant/BaseVariantReader.java | 3 +-
.../org/apache/paimon/data/variant/VariantGet.java | 88 +++++++++++++++++++++-
.../paimon/data/variant/GenericVariantTest.java | 81 ++++++++++++++++++++
.../data/variant/PaimonShreddingUtilsTest.java | 44 ++++++++++-
.../parquet/reader/VariantShreddingReadTest.java | 62 +++++++++++++++
.../apache/paimon/spark/sql/VariantTestBase.scala | 30 ++++++++
6 files changed, 303 insertions(+), 5 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java
b/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java
index e5c5011d20..807f3a901c 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java
@@ -543,7 +543,8 @@ public class BaseVariantReader {
if (noNeedCast) {
return i;
}
- Object result = VariantGet.castScalar(i, scalaType, targetType,
resolve);
+ Object result =
+ VariantGet.castScalar(i, scalaType, targetType, resolve,
castArgs.zoneId());
return result == null ? invalidCast(row, topLevelMetadata) :
result;
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java
b/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java
index 5a22fc1752..cf81143cc8 100644
--- a/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java
+++ b/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java
@@ -29,16 +29,21 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.variant.GenericVariantUtil.Type;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeChecks;
+import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.MapType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VariantType;
+import org.apache.paimon.utils.DateTimeUtils;
import javax.annotation.Nullable;
import java.math.BigDecimal;
import java.math.RoundingMode;
+import java.time.ZoneId;
import java.util.HashMap;
+import java.util.TimeZone;
/** Utils for variant get. */
public class VariantGet {
@@ -175,7 +180,7 @@ public class VariantGet {
CastExecutor<Object, Object> resolve =
(CastExecutor<Object, Object>)
CastExecutors.resolve(inputType, dataType);
- Object result = castScalar(input, inputType, dataType, resolve);
+ Object result = castScalar(input, inputType, dataType, resolve,
castArgs.zoneId());
return result == null ? invalidCast(v, dataType, castArgs) :
result;
}
}
@@ -183,14 +188,21 @@ public class VariantGet {
/**
* Casts a non-null scalar read from a variant to {@code targetType},
returning null when the
* cast is invalid. The generic cast rules wrap a numeric value that does
not fit the target, so
- * an out-of-range value is rejected here first, matching Spark's TRY cast
semantics.
+ * an out-of-range value is rejected here first, matching Spark's TRY cast
semantics. Casts that
+ * move between an instant and a local date or time use {@code zoneId},
the zone the query asked
+ * for, rather than the JVM default the generic rules fall back to.
*/
@Nullable
static Object castScalar(
Object input,
DataType inputType,
DataType targetType,
- @Nullable CastExecutor<Object, Object> executor) {
+ @Nullable CastExecutor<Object, Object> executor,
+ ZoneId zoneId) {
+ Object temporal = castTemporal(input, inputType, targetType, zoneId);
+ if (temporal != NOT_TEMPORAL) {
+ return temporal;
+ }
if (executor == null || !fitsIntegralTarget(input, inputType,
targetType)) {
return null;
}
@@ -201,6 +213,76 @@ public class VariantGet {
}
}
+ private static final Object NOT_TEMPORAL = new Object();
+
+ /**
+ * The casts whose result depends on a time zone: between a timestamp with
local time zone and a
+ * string, a timestamp without time zone or a date, and from a string to a
timestamp with local
+ * time zone. Returns {@link #NOT_TEMPORAL} for every other pair, and null
for a string that
+ * does not parse. A timestamp renders like Spark's cast, without trailing
fraction zeros.
+ */
+ @Nullable
+ private static Object castTemporal(
+ Object input, DataType inputType, DataType targetType, ZoneId
zoneId) {
+ TimeZone tz = TimeZone.getTimeZone(zoneId);
+ switch (inputType.getTypeRoot()) {
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ Timestamp instant = (Timestamp) input;
+ switch (targetType.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ return BinaryString.fromString(
+ DateTimeUtils.formatTimestamp(
+
DateTimeUtils.timestampWithLocalZoneToTimestamp(
+ instant, tz),
+ 0));
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return truncate(
+
DateTimeUtils.timestampWithLocalZoneToTimestamp(instant, tz),
+ targetType);
+ case DATE:
+ return
DateTimeUtils.timestampWithLocalZoneToDate(instant, tz);
+ default:
+ return NOT_TEMPORAL;
+ }
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ Timestamp local = (Timestamp) input;
+ switch (targetType.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ return
BinaryString.fromString(DateTimeUtils.formatTimestamp(local, 0));
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return truncate(
+
DateTimeUtils.timestampToTimestampWithLocalZone(local, tz),
+ targetType);
+ default:
+ return NOT_TEMPORAL;
+ }
+ case DATE:
+ if
(targetType.is(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+ return
DateTimeUtils.dateToTimestampWithLocalZone((Integer) input, tz);
+ }
+ return NOT_TEMPORAL;
+ case CHAR:
+ case VARCHAR:
+ if
(targetType.is(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+ try {
+ return DateTimeUtils.parseTimestampData(
+ input.toString(),
DataTypeChecks.getPrecision(targetType), tz);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+ return NOT_TEMPORAL;
+ default:
+ return NOT_TEMPORAL;
+ }
+ }
+
+ private static Timestamp truncate(Timestamp timestamp, DataType
targetType) {
+ return DateTimeUtils.truncate(timestamp,
DataTypeChecks.getPrecision(targetType));
+ }
+
/** Whether a numeric {@code input} lies within the range of an integral
{@code targetType}. */
private static boolean fitsIntegralTarget(
Object input, DataType inputType, DataType targetType) {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/variant/GenericVariantTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/variant/GenericVariantTest.java
index 46b5b8c992..12c54e54f8 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/variant/GenericVariantTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/variant/GenericVariantTest.java
@@ -23,6 +23,7 @@ import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -34,7 +35,11 @@ import java.io.ObjectStreamClass;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
+import java.time.ZoneId;
import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.TimeZone;
import static
org.apache.paimon.data.variant.PaimonShreddingUtils.assembleVariant;
import static
org.apache.paimon.data.variant.PaimonShreddingUtils.buildVariantSchema;
@@ -345,6 +350,82 @@ public class GenericVariantTest {
.isEqualTo(1.2345678901234567E19);
}
+ @Test
+ public void testVariantGetTemporalCastsUseRequestedZone() {
+ // 2023-11-14 22:13:20.5 UTC as an instant, the same wall time as a
local timestamp,
+ // and the date 2024-10-04
+ Map<String, Object> values = new HashMap<>();
+ values.put("ts", 1700000000500000L);
+ values.put("ntz", 1700000000500000L);
+ values.put("d", 20000);
+ values.put("s", "2023-11-14 22:13:20");
+ RowType valueType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(),
+ DataTypes.TIMESTAMP(),
+ DataTypes.DATE(),
+ DataTypes.STRING()
+ },
+ new String[] {"ts", "ntz", "d", "s"});
+ Variant variant = GenericVariantBuilderHelper.build(valueType, values);
+ VariantCastArgs shanghai = new VariantCastArgs(false,
ZoneId.of("Asia/Shanghai"));
+ VariantCastArgs utc = new VariantCastArgs(false, ZoneOffset.UTC);
+
+ // The generic cast rules use the JVM default zone; the requested zone
must win no
+ // matter what the JVM default is, so run the same assertions under
two defaults.
+ TimeZone original = TimeZone.getDefault();
+ try {
+ for (String jvmZone : new String[] {"UTC", "America/Los_Angeles"})
{
+ TimeZone.setDefault(TimeZone.getTimeZone(jvmZone));
+
+ // instant rendered in the requested zone, without trailing
fraction zeros
+ assertThat(variant.variantGet("$.ts", DataTypes.STRING(),
shanghai))
+ .isEqualTo(BinaryString.fromString("2023-11-15
06:13:20.5"));
+ assertThat(variant.variantGet("$.ts", DataTypes.STRING(), utc))
+ .isEqualTo(BinaryString.fromString("2023-11-14
22:13:20.5"));
+ assertThat(variant.variantGet("$.ts", DataTypes.TIMESTAMP(),
shanghai))
+ .isEqualTo(Timestamp.fromMicros(1700000000500000L + 8
* 3600_000_000L));
+ assertThat(variant.variantGet("$.ts", DataTypes.DATE(),
shanghai)).isEqualTo(19676);
+ assertThat(variant.variantGet("$.ts", DataTypes.DATE(),
utc)).isEqualTo(19675);
+
+ // a local timestamp has no zone to render, but gains one when
it becomes an
+ // instant
+ assertThat(variant.variantGet("$.ntz", DataTypes.STRING(),
shanghai))
+ .isEqualTo(BinaryString.fromString("2023-11-14
22:13:20.5"));
+ assertThat(
+ variant.variantGet(
+ "$.ntz",
+
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(),
+ shanghai))
+ .isEqualTo(Timestamp.fromMicros(1700000000500000L - 8
* 3600_000_000L));
+
+ // midnight of the date in the requested zone
+ assertThat(
+ variant.variantGet(
+ "$.d",
+
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(),
+ shanghai))
+ .isEqualTo(
+ Timestamp.fromMicros(20000L * 86400_000_000L -
8 * 3600_000_000L));
+
+ // a string is parsed as wall time in the requested zone
+ assertThat(
+ variant.variantGet(
+ "$.s",
+
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(),
+ shanghai))
+ .isEqualTo(Timestamp.fromMicros(1700000000000000L - 8
* 3600_000_000L));
+
+ // the target precision is still honoured
+ assertThat(variant.variantGet("$.ts", DataTypes.TIMESTAMP(0),
utc))
+ .isEqualTo(Timestamp.fromMicros(1700000000000000L));
+ }
+ } finally {
+ TimeZone.setDefault(original);
+ }
+ }
+
@Test
public void testObjectFieldOrderingCompatibility() {
String bmpKey = "\uE000";
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java
index a65d88c6dc..01915ed422 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java
@@ -46,6 +46,7 @@ import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.HashMap;
@@ -219,10 +220,11 @@ public class PaimonShreddingUtilsTest {
BinaryString.fromString("true"),
null,
BinaryString.fromString(DateTimeUtils.formatDate(20000)),
+ // rendered in the zone castArgs asks for, not the JVM
default
BinaryString.fromString(
DateTimeUtils.formatTimestamp(
Timestamp.fromMicros(1_234_567_890_123_456L),
- TimeZone.getDefault(),
+ DateTimeUtils.UTC_ZONE,
6)),
BinaryString.fromString(
DateTimeUtils.formatTimestamp(
@@ -340,6 +342,46 @@ public class PaimonShreddingUtilsTest {
}
}
+ @Test
+ public void testShreddedTimestampCastsUseRequestedZone() {
+ // A shredded typed_value timestamp goes through the scalar reader, an
unshredded one
+ // through VariantGet; both must convert with the requested zone, not
the JVM default.
+ Map<String, Object> values = new HashMap<>();
+ values.put("ts", 1700000000500000L); // 2023-11-14 22:13:20.5 UTC
+ RowType valueType =
+ RowType.of(
+ new DataType[]
{DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()},
+ new String[] {"ts"});
+ GenericVariant v = GenericVariantBuilderHelper.build(valueType,
values);
+ VariantCastArgs shanghai = new VariantCastArgs(true,
ZoneId.of("Asia/Shanghai"));
+
+ TimeZone original = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"));
+ for (RowType shape : new RowType[] {valueType, RowType.of()}) {
+ VariantSchema variantSchema =
buildVariantSchema(variantShreddingSchema(shape));
+ FieldToExtract[] fieldsToExtract = {
+ buildFieldsToExtract(DataTypes.STRING(), "$.ts", shanghai,
variantSchema),
+ buildFieldsToExtract(DataTypes.TIMESTAMP(), "$.ts",
shanghai, variantSchema),
+ buildFieldsToExtract(DataTypes.DATE(), "$.ts", shanghai,
variantSchema)
+ };
+ assertThat(
+ assembleVariantStruct(
+ castShredded(v, variantSchema),
+ variantSchema,
+ fieldsToExtract))
+ .as("shape %s", shape)
+ .isEqualTo(
+ GenericRow.of(
+ BinaryString.fromString("2023-11-15
06:13:20.5"),
+ Timestamp.fromMicros(1700000000500000L
+ 8 * 3600_000_000L),
+ 19676));
+ }
+ } finally {
+ TimeZone.setDefault(original);
+ }
+ }
+
@Test
public void testAssembleDecimalWithScaleAbovePrecision() {
// the unshredded leg extracts through VariantGet, which used to build
an invalid
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java
index 5b93ff3c73..12e5a60a1e 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java
@@ -23,8 +23,10 @@ import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.data.variant.GenericVariant;
+import org.apache.paimon.data.variant.GenericVariantBuilderHelper;
import org.apache.paimon.data.variant.VariantMetadataUtils;
import org.apache.paimon.format.FileFormatFactory;
import org.apache.paimon.format.FormatReaderContext;
@@ -38,6 +40,7 @@ import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -50,7 +53,10 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -594,6 +600,62 @@ public class VariantShreddingReadTest {
.hasMessageContaining("Invalid cast 99999999999 to INT");
}
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "null",
+
"{\"type\":\"ROW\",\"fields\":[{\"name\":\"v\",\"type\":{\"type\":\"ROW\",\"fields\":[{\"name\":\"ts\",\"type\":\"TIMESTAMP(6)
WITH LOCAL TIME ZONE\"}]}}]}"
+ })
+ public void testReadTimestampInRequestedZone(String shreddingSchema)
throws Exception {
+ // The extraction carries the zone the query runs in; a shredded
typed_value and an
+ // unshredded value must both convert with it rather than with the JVM
default.
+ Options options = new Options();
+ if (!shreddingSchema.equals("null")) {
+ options.set("parquet.variant.shreddingSchema", shreddingSchema);
+ }
+ ParquetFileFormat format =
+ new ParquetFileFormat(new
FileFormatFactory.FormatContext(options, 1024, 1024));
+
+ RowType writeType = DataTypes.ROW(DataTypes.FIELD(0, "v",
DataTypes.VARIANT()));
+ Map<String, Object> values = new HashMap<>();
+ values.put("ts", 1700000000500000L); // 2023-11-14 22:13:20.5 UTC
+ writeRows(
+ format.createWriterFactory(writeType),
+ GenericRow.of(
+ GenericVariantBuilderHelper.build(
+ RowType.of(
+ new DataType[]
{DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()},
+ new String[] {"ts"}),
+ values)));
+
+ RowType readType =
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 0,
+ "v",
+
VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.STRING(), "$.ts",
true, "Asia/Shanghai")
+ .field(DataTypes.DATE(), "$.ts", true,
"Asia/Shanghai")
+ .field(DataTypes.TIMESTAMP(), "$.ts",
true, "UTC")
+ .build()));
+ TimeZone original = TimeZone.getDefault();
+ try {
+ for (String jvmZone : new String[] {"UTC", "America/Los_Angeles"})
{
+ TimeZone.setDefault(TimeZone.getTimeZone(jvmZone));
+ List<InternalRow> result = readRows(format, readType);
+ assertThat(result.get(0).getRow(0, 3))
+ .as("JVM zone %s", jvmZone)
+ .isEqualTo(
+ GenericRow.of(
+ BinaryString.fromString("2023-11-15
06:13:20.5"),
+ 19676,
+
Timestamp.fromMicros(1700000000500000L)));
+ }
+ } finally {
+ TimeZone.setDefault(original);
+ }
+ }
+
protected List<InternalRow> readRows(ParquetFileFormat format, RowType
rowType)
throws IOException {
List<InternalRow> result = new ArrayList<>();
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
index 6bddc24fd4..fda16f8219 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
@@ -1130,6 +1130,36 @@ abstract class VariantTestBase extends
PaimonSparkTestBase {
Seq(Row(1, null, null), Row(2, 7, 1L)))
}
+ test("Paimon Variant pushdown: timestamp extraction follows the session time
zone") {
+ // The test JVM runs in America/Los_Angeles; the session zone below
differs from it, so a
+ // pushed-down extraction must convert with the session zone Spark hands
to the scan.
+ withSparkSQLConf("spark.sql.session.timeZone" -> "Asia/Shanghai") {
+ sql("CREATE TABLE T (id INT, v VARIANT)")
+ sql("""INSERT INTO T VALUES (1, to_variant_object(named_struct(
+ | 'ts', timestamp'2023-11-15 06:13:20.5',
+ | 'ntz', timestamp_ntz'2023-11-15 06:13:20.5',
+ | 'd', date'2024-10-04')))
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("""SELECT
+ | variant_get(v, '$.ts', 'string'),
+ | CAST(try_variant_get(v, '$.ts', 'timestamp_ntz') AS STRING),
+ | CAST(try_variant_get(v, '$.ts', 'date') AS STRING),
+ | CAST(try_variant_get(v, '$.ntz', 'timestamp') AS STRING),
+ | CAST(try_variant_get(v, '$.d', 'timestamp') AS STRING)
+ |FROM T""".stripMargin),
+ Seq(
+ Row(
+ "2023-11-15 06:13:20.5",
+ "2023-11-15 06:13:20.5",
+ "2023-11-15",
+ "2023-11-15 06:13:20.5",
+ "2024-10-04 00:00:00"))
+ )
+ }
+ }
+
test("Paimon Variant pushdown: nested variant column inside a struct") {
assume(gteqSpark4_1)
sql("CREATE TABLE T (id INT, nested STRUCT<v: VARIANT, x: INT>)")
