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 2c87f4a6ea [spark] Support timestamp_ntz and variant in generic row
access (#9493)
2c87f4a6ea is described below
commit 2c87f4a6ea8afb530fda23b464a5fe1042ab42fe
Author: shyjsarah <[email protected]>
AuthorDate: Mon Aug 31 18:14:25 2026 +0800
[spark] Support timestamp_ntz and variant in generic row access (#9493)
---
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 3 ++
.../spark/data/SparkInternalRowVariantTest.scala | 41 ++++++++++++++++++++++
.../paimon/spark/AbstractSparkInternalRow.java | 6 +++-
.../apache/spark/sql/paimon/shims/SparkShim.scala | 2 ++
.../apache/paimon/spark/SparkInternalRowTest.java | 12 +++++++
.../apache/spark/sql/paimon/shims/Spark3Shim.scala | 3 ++
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 3 ++
7 files changed, 69 insertions(+), 1 deletion(-)
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 8cdc6e453d..c348349fe9 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -375,6 +375,9 @@ class Spark4Shim extends SparkShim {
new GenericVariant(v.getValue, v.getMetadata)
}
+ override def toSparkVariant(variant: Variant): Object =
+ new VariantVal(variant.value(), variant.metadata())
+
override def isSparkVariantType(dataType:
org.apache.spark.sql.types.DataType): Boolean =
dataType.isInstanceOf[VariantType]
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/data/SparkInternalRowVariantTest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/data/SparkInternalRowVariantTest.scala
new file mode 100644
index 0000000000..ee00a9b4c5
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/data/SparkInternalRowVariantTest.scala
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.data
+
+import org.apache.paimon.data.GenericRow
+import org.apache.paimon.data.variant.GenericVariant
+import org.apache.paimon.types.{RowType, VariantType}
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.types.DataTypes
+import org.apache.spark.unsafe.types.VariantVal
+
+class SparkInternalRowVariantTest extends SparkFunSuite {
+
+ test("get variant with generic data type access") {
+ val variant = GenericVariant.fromJson("""{"id":1}""")
+ val rowType = RowType.of(new VariantType())
+ val row = SparkInternalRow.create(rowType).replace(GenericRow.of(variant))
+
+ val actual = row.get(0, DataTypes.VariantType).asInstanceOf[VariantVal]
+
+ assert(actual.getValue.sameElements(variant.value()))
+ assert(actual.getMetadata.sameElements(variant.metadata()))
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/AbstractSparkInternalRow.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/AbstractSparkInternalRow.java
index 46c48833d0..206fc0e20a 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/AbstractSparkInternalRow.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/AbstractSparkInternalRow.java
@@ -47,6 +47,7 @@ import org.apache.spark.sql.types.NullType;
import org.apache.spark.sql.types.ShortType;
import org.apache.spark.sql.types.StringType;
import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.types.TimestampNTZType;
import org.apache.spark.sql.types.TimestampType;
import org.apache.spark.sql.types.UserDefinedType;
import org.apache.spark.sql.types.VarcharType;
@@ -236,7 +237,7 @@ public abstract class AbstractSparkInternalRow extends
SparkInternalRow {
if (dataType instanceof DateType) {
return getInt(ordinal);
}
- if (dataType instanceof TimestampType) {
+ if (dataType instanceof TimestampType || dataType instanceof
TimestampNTZType) {
return getLong(ordinal);
}
if (dataType instanceof CalendarIntervalType) {
@@ -257,6 +258,9 @@ public abstract class AbstractSparkInternalRow extends
SparkInternalRow {
if (dataType instanceof UserDefinedType) {
return get(ordinal, ((UserDefinedType<?>) dataType).sqlType());
}
+ if (SparkShimLoader.shim().isSparkVariantType(dataType)) {
+ return
SparkShimLoader.shim().toSparkVariant(row.getVariant(ordinal));
+ }
if (SparkShimLoader.shim().isSparkGeometryType(dataType)) {
org.apache.paimon.types.GeometryType geometryType =
(org.apache.paimon.types.GeometryType)
rowType.getTypeAt(ordinal);
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
index 8d0cd71579..e4a0a315f2 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
@@ -265,6 +265,8 @@ trait SparkShim {
def toPaimonVariant(array: ArrayData, pos: Int): Variant
+ def toSparkVariant(variant: Variant): Object
+
def isSparkVariantType(dataType: org.apache.spark.sql.types.DataType):
Boolean
def SparkVariantType(): org.apache.spark.sql.types.DataType
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
index cf2886e947..e56a07a527 100644
---
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
@@ -30,6 +30,7 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.spark.data.SparkInternalRow;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.DateTimeUtils;
import org.apache.paimon.utils.UriReaderFactory;
@@ -178,6 +179,17 @@ public class SparkInternalRowTest {
.hasMessage("Spark MAP<X, BLOB> does not support null keys.");
}
+ @Test
+ public void testGetTimestampNtz() {
+ Timestamp timestamp =
+
Timestamp.fromLocalDateTime(LocalDateTime.parse("2026-08-31T10:15:30.123456"));
+ RowType rowType = RowType.of(DataTypes.TIMESTAMP());
+ SparkInternalRow row =
SparkInternalRow.create(rowType).replace(GenericRow.of(timestamp));
+
+ assertThat(row.get(0,
SparkTypeUtils.fromPaimonType(DataTypes.TIMESTAMP())))
+ .isEqualTo(timestamp.toMicros());
+ }
+
private String sparkRowToString(org.apache.spark.sql.Row row) {
return JavaConverters.seqAsJavaList(row.toSeq()).stream()
.map(
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
index 3ea437da35..9b9b617416 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
@@ -349,6 +349,9 @@ class Spark3Shim extends SparkShim {
override def toPaimonVariant(o: Object): Variant = throw new
UnsupportedOperationException()
+ override def toSparkVariant(variant: Variant): Object =
+ throw new UnsupportedOperationException("Variant requires Spark 4.0 or
later")
+
override def isSparkVariantType(dataType:
org.apache.spark.sql.types.DataType): Boolean = false
override def SparkVariantType(): org.apache.spark.sql.types.DataType =
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 4ae3bb1892..ae78e6eb79 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -353,6 +353,9 @@ class Spark4Shim extends SparkShim {
new GenericVariant(v.getValue, v.getMetadata)
}
+ override def toSparkVariant(variant: Variant): Object =
+ new VariantVal(variant.value(), variant.metadata())
+
override def isSparkVariantType(dataType:
org.apache.spark.sql.types.DataType): Boolean =
dataType.isInstanceOf[VariantType]
