File tree Expand file tree Collapse file tree
spark/common/src/main/scala/org/apache Expand file tree Collapse file tree Original file line number Diff line number Diff line change @@ -31,9 +31,13 @@ object RasterRegistrator {
3131 def registerAll (sparkSession : SparkSession ): Unit = {
3232 if (isGeoToolsAvailable) {
3333 RasterUdtRegistratorWrapper .registerAll(gridClassName)
34- sparkSession.udf.register(
35- RasterUdafCatalog .rasterAggregateExpression.getClass.getSimpleName,
36- functions.udaf(RasterUdafCatalog .rasterAggregateExpression))
34+ val functionName = RasterUdafCatalog .rasterAggregateExpression.getClass.getSimpleName
35+ val functionIdentifier = FunctionIdentifier (functionName)
36+ if (! sparkSession.sessionState.functionRegistry.functionExists(functionIdentifier)) {
37+ sparkSession.udf.register(
38+ functionName,
39+ functions.udaf(RasterUdafCatalog .rasterAggregateExpression))
40+ }
3741 }
3842 }
3943
Original file line number Diff line number Diff line change @@ -23,6 +23,8 @@ import org.apache.spark.sql.types.UDTRegistration
2323object RasterUdtRegistratorWrapper {
2424
2525 def registerAll (gridClassName : String ): Unit = {
26- UDTRegistration .register(gridClassName, classOf [RasterUDT ].getName)
26+ if (! UDTRegistration .exists(gridClassName)) {
27+ UDTRegistration .register(gridClassName, classOf [RasterUDT ].getName)
28+ }
2729 }
2830}
Original file line number Diff line number Diff line change @@ -26,8 +26,14 @@ import org.locationtech.jts.index.SpatialIndex
2626object UdtRegistratorWrapper {
2727
2828 def registerAll (): Unit = {
29- UDTRegistration .register(classOf [Geometry ].getName, classOf [GeometryUDT ].getName)
30- UDTRegistration .register(classOf [Geography ].getName, classOf [GeographyUDT ].getName)
31- UDTRegistration .register(classOf [SpatialIndex ].getName, classOf [IndexUDT ].getName)
29+ registerIfNotExists(classOf [Geometry ].getName, classOf [GeometryUDT ].getName)
30+ registerIfNotExists(classOf [Geography ].getName, classOf [GeographyUDT ].getName)
31+ registerIfNotExists(classOf [SpatialIndex ].getName, classOf [IndexUDT ].getName)
32+ }
33+
34+ private def registerIfNotExists (userClass : String , udtClass : String ): Unit = {
35+ if (! UDTRegistration .exists(userClass)) {
36+ UDTRegistration .register(userClass, udtClass)
37+ }
3238 }
3339}
Original file line number Diff line number Diff line change @@ -24,9 +24,13 @@ import org.apache.spark.sql.types.UDTRegistration
2424
2525object UdtRegistratorWrapper {
2626 def registerAll (): Unit = {
27- UDTRegistration .register(
28- classOf [ImageSerializableWrapper ].getName,
29- classOf [ImageWrapperUDT ].getName)
30- UDTRegistration .register(classOf [Pixel ].getName, classOf [PixelUDT ].getName)
27+ if (! UDTRegistration .exists(classOf [ImageSerializableWrapper ].getName)) {
28+ UDTRegistration .register(
29+ classOf [ImageSerializableWrapper ].getName,
30+ classOf [ImageWrapperUDT ].getName)
31+ }
32+ if (! UDTRegistration .exists(classOf [Pixel ].getName)) {
33+ UDTRegistration .register(classOf [Pixel ].getName, classOf [PixelUDT ].getName)
34+ }
3135 }
3236}
You can’t perform that action at this time.
0 commit comments