This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 23633c3c44af931eb50fb06d41f885442678c1f4 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>)")
