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 a39e3e8ed6 [spark] Use jackson shaded by Paimon instead of Spark's
built-in one (#8479)
a39e3e8ed6 is described below
commit a39e3e8ed6493f4e9e0173b5192e98ce761941ba
Author: Kerwin Zhang <[email protected]>
AuthorDate: Mon Jul 6 20:44:36 2026 +0800
[spark] Use jackson shaded by Paimon instead of Spark's built-in one (#8479)
---
.../paimon/spark/sources/PaimonSourceOffset.scala | 19 ++++++++--
.../org/apache/paimon/spark/util/JsonUtils.scala | 42 ----------------------
paimon-spark/pom.xml | 2 +-
3 files changed, 17 insertions(+), 46 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala
index 5d8f4baafb..9b89729dd1 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSourceOffset.scala
@@ -18,8 +18,8 @@
package org.apache.paimon.spark.sources
-import org.apache.paimon.spark.util.JsonUtils
import org.apache.paimon.table.source.snapshot.StartingContext
+import org.apache.paimon.utils.JsonSerdeUtil
import org.apache.spark.sql.connector.read.streaming.Offset
@@ -38,7 +38,11 @@ case class PaimonSourceOffset(snapshotId: Long, index: Long,
scanSnapshot: Boole
with Comparable[PaimonSourceOffset] {
override def json(): String = {
- JsonUtils.toJson(this)
+ val node = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.createObjectNode()
+ node.put(PaimonSourceOffset.FIELD_SNAPSHOT_ID, snapshotId)
+ node.put(PaimonSourceOffset.FIELD_INDEX, index)
+ node.put(PaimonSourceOffset.FIELD_SCAN_SNAPSHOT, scanSnapshot)
+ node.toString
}
override def compareTo(o: PaimonSourceOffset): Int = {
@@ -58,6 +62,10 @@ object PaimonSourceOffset {
// index of the init offset, for we filter offset by (startOffset,
endOffset]
val INIT_OFFSET_INDEX: Long = -1L
+ private val FIELD_SNAPSHOT_ID = "snapshotId"
+ private val FIELD_INDEX = "index"
+ private val FIELD_SCAN_SNAPSHOT = "scanSnapshot"
+
def apply(version: Long, index: Long, scanSnapshot: Boolean):
PaimonSourceOffset = {
new PaimonSourceOffset(
version,
@@ -69,7 +77,12 @@ object PaimonSourceOffset {
def apply(offset: Any): PaimonSourceOffset = {
offset match {
case o: PaimonSourceOffset => o
- case json: String => JsonUtils.fromJson[PaimonSourceOffset](json)
+ case json: String =>
+ val node = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(json)
+ PaimonSourceOffset(
+ node.get(FIELD_SNAPSHOT_ID).asLong(),
+ node.get(FIELD_INDEX).asLong(),
+ node.get(FIELD_SCAN_SNAPSHOT).asBoolean())
case sc: StartingContext =>
PaimonSourceOffset(sc.getSnapshotId, INIT_OFFSET_INDEX,
sc.getScanFullSnapshot)
case _ => throw new IllegalArgumentException(s"Can't parse $offset to
PaimonSourceOffset.")
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/JsonUtils.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/JsonUtils.scala
deleted file mode 100644
index b0cf962e98..0000000000
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/JsonUtils.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.util
-
-import com.fasterxml.jackson.annotation.JsonInclude.Include
-import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
-import com.fasterxml.jackson.module.scala.{DefaultScalaModule,
ScalaObjectMapper}
-
-object JsonUtils {
-
- lazy val mapper = {
- val _mapper = new ObjectMapper with ScalaObjectMapper {}
- _mapper.setSerializationInclusion(Include.NON_ABSENT)
- _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
- _mapper.registerModule(DefaultScalaModule)
- _mapper
- }
-
- def toJson[T: Manifest](obj: T): String = {
- mapper.writeValueAsString(obj)
- }
-
- def fromJson[T: Manifest](json: String): T = {
- mapper.readValue[T](json)
- }
-}
diff --git a/paimon-spark/pom.xml b/paimon-spark/pom.xml
index e843edbef6..124e548cc1 100644
--- a/paimon-spark/pom.xml
+++ b/paimon-spark/pom.xml
@@ -45,7 +45,7 @@ under the License.
<artifactId>paimon-format</artifactId>
<version>${project.version}</version>
<exclusions>
- <!-- todo(fix me): we should use the jackson libraries
shaded by Paimon. -->
+ <!-- Paimon relies on the jackson libraries shaded into
its bundle, exclude the unshaded ones. -->
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>*</artifactId>
