diff --git a/src/main/java/com/snowflake/snowpark_java/Functions.java b/src/main/java/com/snowflake/snowpark_java/Functions.java
index 1c3552a5..74e0190a 100644
--- a/src/main/java/com/snowflake/snowpark_java/Functions.java
+++ b/src/main/java/com/snowflake/snowpark_java/Functions.java
@@ -360,8 +360,8 @@ public static Column stddev_pop(Column col) {
   }
 
   /**
-   * Returns the sum of non-NULL records in a group. If all records inside a group are NULL, the function returns
-   * NULL.
+   * Returns the sum of non-NULL records in a group. If all records inside a group are NULL, the
+   * function returns NULL.
    *
    * @since 0.9.0
    * @param col The input column
@@ -372,14 +372,16 @@ public static Column sum(Column col) {
   }
 
   /**
-   * Returns the sum of non-NULL records in a group. If all records inside a group are NULL, the function returns
-   * NULL.
+   * Returns the sum of non-NULL records in a group. If all records inside a group are NULL, the
+   * function returns NULL.
    *
    * @since 1.12.0
    * @param colName The input column name
    * @return The result column
    */
-  public static Column sum(String colName) { return sum(col(colName)); }
+  public static Column sum(String colName) {
+    return sum(col(colName));
+  }
 
   /**
    * Returns the sum of non-NULL distinct records in a group. You can use the DISTINCT keyword to
diff --git a/src/main/java/com/snowflake/snowpark_java/Row.java b/src/main/java/com/snowflake/snowpark_java/Row.java
index 596acb68..cff3489e 100644
--- a/src/main/java/com/snowflake/snowpark_java/Row.java
+++ b/src/main/java/com/snowflake/snowpark_java/Row.java
@@ -2,6 +2,7 @@
 
 import com.snowflake.snowpark.internal.JavaUtils;
 import com.snowflake.snowpark_java.types.Geography;
+import com.snowflake.snowpark_java.types.Geometry;
 import com.snowflake.snowpark_java.types.InternalUtils;
 import com.snowflake.snowpark_java.types.Variant;
 import java.io.Serializable;
@@ -53,6 +54,8 @@ private static Object[] javaObjectToScalaObject(Object[] input) {
       } else if (result[i] instanceof Geography) {
         result[i] =
             com.snowflake.snowpark.types.Geography.fromGeoJSON(((Geography) result[i]).asGeoJSON());
+      } else if (result[i] instanceof Geometry) {
+        result[i] = com.snowflake.snowpark.types.Geometry.fromGeoJSON(result[i].toString());
       }
     }
     return result;
@@ -134,6 +137,8 @@ public Object get(int index) {
       return InternalUtils.createVariant((com.snowflake.snowpark.types.Variant) result);
     } else if (result instanceof com.snowflake.snowpark.types.Geography) {
       return Geography.fromGeoJSON(((com.snowflake.snowpark.types.Geography) result).asGeoJSON());
+    } else if (result instanceof com.snowflake.snowpark.types.Geometry) {
+      return Geometry.fromGeoJSON(result.toString());
     } else if (result instanceof com.snowflake.snowpark.types.Variant[]) {
       com.snowflake.snowpark.types.Variant[] scalaVariantArray =
           (com.snowflake.snowpark.types.Variant[]) result;
@@ -325,6 +330,17 @@ public Geography getGeography(int index) {
     return Geography.fromGeoJSON(scalaRow.getGeography(index).asGeoJSON());
   }
 
+  /**
+   * Retrieves the value of the column at the given index as a Geometry value.
+   *
+   * @param index The index of target column
+   * @return The Geometry value of the column at the given index
+   * @since 1.12.0
+   */
+  public Geometry getGeometry(int index) {
+    return Geometry.fromGeoJSON(scalaRow.getGeometry(index).toString());
+  }
+
   /**
    * Retrieves the value of the column at the given index as a list of Variant.
    *
diff --git a/src/main/java/com/snowflake/snowpark_java/SessionBuilder.java b/src/main/java/com/snowflake/snowpark_java/SessionBuilder.java
index c590b955..1e9c2c49 100644
--- a/src/main/java/com/snowflake/snowpark_java/SessionBuilder.java
+++ b/src/main/java/com/snowflake/snowpark_java/SessionBuilder.java
@@ -82,8 +82,8 @@ public Session getOrCreate() {
   }
 
   /**
-   * Adds the app name to set in the query_tag after session creation.
-   * The query tag will be set with this format 'APPNAME=${appName}'.
+   * Adds the app name to set in the query_tag after session creation. The query tag will be set
+   * with this format 'APPNAME=${appName}'.
    *
    * @param appName Name of the app.
    * @return A {@code SessionBuilder} object
diff --git a/src/main/java/com/snowflake/snowpark_java/types/DataTypes.java b/src/main/java/com/snowflake/snowpark_java/types/DataTypes.java
index 30dbbf75..c88b8b50 100644
--- a/src/main/java/com/snowflake/snowpark_java/types/DataTypes.java
+++ b/src/main/java/com/snowflake/snowpark_java/types/DataTypes.java
@@ -80,6 +80,13 @@ private DataTypes() {}
    */
   public static final GeographyType GeographyType = new GeographyType();
 
+  /**
+   * Retrieves the GeometryType object.
+   *
+   * @since 1.12.0
+   */
+  public static final GeometryType GeometryType = new GeometryType();
+
   /**
    * Retrieves the StringType object.
    *
diff --git a/src/main/java/com/snowflake/snowpark_java/types/Geometry.java b/src/main/java/com/snowflake/snowpark_java/types/Geometry.java
new file mode 100644
index 00000000..0ab47bb9
--- /dev/null
+++ b/src/main/java/com/snowflake/snowpark_java/types/Geometry.java
@@ -0,0 +1,70 @@
+package com.snowflake.snowpark_java.types;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+
+/**
+ * Java representation of Snowflake Geometry data.
+ *
+ * @since 1.12.0
+ */
+public class Geometry implements Serializable {
+  private final String data;
+
+  private Geometry(String data) {
+    if (data == null) {
+      throw new UncheckedIOException(
+          new IOException("Cannot create geometry object from null input"));
+    }
+    this.data = data;
+  }
+
+  /**
+   * Checks whether two Geometry object are equal.
+   *
+   * @param other A Geometry object
+   * @return true if these two object are equal
+   * @since 1.12.0
+   */
+  @Override
+  public boolean equals(Object other) {
+    if (other instanceof Geometry) {
+      return data.equals(((Geometry) other).data);
+    }
+    return false;
+  }
+
+  /**
+   * Calculates the hash code of this Geometry Object.
+   *
+   * @return An int number representing the hash code value
+   * @since 1.12.0
+   */
+  @Override
+  public int hashCode() {
+    return this.data.hashCode();
+  }
+
+  /**
+   * Converts this Geometry object to a String value.
+   *
+   * @return A String value.
+   * @since 1.12.0
+   */
+  @Override
+  public String toString() {
+    return this.data;
+  }
+
+  /**
+   * Creates a Geometry object from a GeoJSON string.
+   *
+   * @param g GeoJSON String
+   * @return a new Geometry object
+   * @since 1.12.0
+   */
+  public static Geometry fromGeoJSON(String g) {
+    return new Geometry(g);
+  }
+}
diff --git a/src/main/java/com/snowflake/snowpark_java/types/GeometryType.java b/src/main/java/com/snowflake/snowpark_java/types/GeometryType.java
new file mode 100644
index 00000000..3b871596
--- /dev/null
+++ b/src/main/java/com/snowflake/snowpark_java/types/GeometryType.java
@@ -0,0 +1,10 @@
+package com.snowflake.snowpark_java.types;
+
+/**
+ * Geography data type. This maps to GEOMETRY data type in Snowflake.
+ *
+ * @since 1.12.0
+ */
+public class GeometryType extends AtomicType {
+  GeometryType() {}
+}
diff --git a/src/main/scala/com/snowflake/snowpark/Row.scala b/src/main/scala/com/snowflake/snowpark/Row.scala
index e5ee3fc2..84419555 100644
--- a/src/main/scala/com/snowflake/snowpark/Row.scala
+++ b/src/main/scala/com/snowflake/snowpark/Row.scala
@@ -1,9 +1,8 @@
 package com.snowflake.snowpark
 
 import java.sql.{Date, Time, Timestamp}
-
 import com.snowflake.snowpark.internal.ErrorMessage
-import com.snowflake.snowpark.types.{Geography, Variant}
+import com.snowflake.snowpark.types.{Geography, Geometry, Variant}
 
 import scala.util.hashing.MurmurHash3
 
@@ -245,6 +244,7 @@ class Row private (values: Array[Any]) extends Serializable {
     get(index) match {
       case variant: Variant => variant.toString
       case geo: Geography => geo.toString
+      case geo: Geometry => geo.toString
       case array: Array[_] => new Variant(array).toString
       case seq: Seq[_] => new Variant(seq).toString
       case map: Map[_, _] => new Variant(map).toString
@@ -301,6 +301,14 @@ class Row private (values: Array[Any]) extends Serializable {
    */
   def getGeography(index: Int): Geography = getAs[Geography](index)
 
+  /**
+   * Returns the value of the column at the given index as Geometry class
+   *
+   * @since 1.12.0
+   * @group getter
+   */
+  def getGeometry(index: Int): Geometry = getAs[Geometry](index)
+
   /**
    * Returns the value of the column at the given index as a Seq of Variant
    * @since 0.2.0
diff --git a/src/main/scala/com/snowflake/snowpark/Session.scala b/src/main/scala/com/snowflake/snowpark/Session.scala
index 321ef271..633c8e42 100644
--- a/src/main/scala/com/snowflake/snowpark/Session.scala
+++ b/src/main/scala/com/snowflake/snowpark/Session.scala
@@ -733,8 +733,8 @@ class Session private (private[snowpark] val conn: ServerConnection) extends Log
     val spAttrs = schema.map { field =>
       {
         val sfType = field.dataType match {
-          case _ @(VariantType | _: ArrayType | _: MapType | GeographyType | TimeType | DateType |
-              TimestampType) =>
+          case _ @(VariantType | _: ArrayType | _: MapType | GeographyType | GeometryType |
+              TimeType | DateType | TimestampType) =>
             StringType
           case other => other
         }
@@ -764,6 +764,7 @@ class Session private (private[snowpark] val conn: ServerConnection) extends Log
         case (value, _: AtomicType) => value
         case (value: Variant, VariantType) => value.asJsonString()
         case (value: Geography, GeographyType) => value.asGeoJSON()
+        case (value: Geometry, GeometryType) => value.toString
         case (value: Array[_], _: ArrayType) =>
           new Variant(value.toSeq).asJsonString()
         case (value: Map[_, _], _: MapType) => new Variant(value).asJsonString()
@@ -784,6 +785,7 @@ class Session private (private[snowpark] val conn: ServerConnection) extends Log
         case TimestampType => callUDF("to_timestamp", column(field.name)).as(field.name)
         case VariantType => to_variant(parse_json(column(field.name))).as(field.name)
         case GeographyType => callUDF("to_geography", column(field.name)).as(field.name)
+        case GeometryType => callUDF("to_geometry", column(field.name)).as(field.name)
         case _: ArrayType => to_array(parse_json(column(field.name))).as(field.name)
         case _: MapType => to_object(parse_json(column(field.name))).as(field.name)
         case _ => column(field.name)
@@ -1408,7 +1410,6 @@ object Session extends Logging {
       this
     }
 
-
     /**
      * Adds the app name to set in the query_tag after session creation.
      * The query tag will be set with this format 'APPNAME=${appName}'.
diff --git a/src/main/scala/com/snowflake/snowpark/internal/ErrorMessage.scala b/src/main/scala/com/snowflake/snowpark/internal/ErrorMessage.scala
index a88ebb35..e0df3d6b 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/ErrorMessage.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/ErrorMessage.scala
@@ -158,7 +158,8 @@ private[snowpark] object ErrorMessage {
     "0424" ->
       """Invalid input argument type, the input argument type of Explode function should be either Map or Array types.
         |The input argument type: %s
-        |""".stripMargin)
+        |""".stripMargin,
+    "0425" -> "Unsupported Geometry output format: %s. Please set session parameter GEOMETRY_OUTPUT_FORMAT to GeoJSON.")
   // scalastyle:on
 
   /*
@@ -405,6 +406,9 @@ private[snowpark] object ErrorMessage {
   def MISC_INVALID_EXPLODE_ARGUMENT_TYPE(argumentType: String): SnowparkClientException =
     createException("0424", argumentType)
 
+  def MISC_UNSUPPORTED_GEOMETRY_FORMAT(typeName: String): SnowparkClientException =
+    createException("0425", typeName)
+
   /**
    * Create Snowpark client Exception.
    *
diff --git a/src/main/scala/com/snowflake/snowpark/internal/JavaDataTypeUtils.scala b/src/main/scala/com/snowflake/snowpark/internal/JavaDataTypeUtils.scala
index f8f42795..79f24e2d 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/JavaDataTypeUtils.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/JavaDataTypeUtils.scala
@@ -13,6 +13,7 @@ import com.snowflake.snowpark_java.types.{
   DoubleType => JDoubleType,
   FloatType => JFloatType,
   GeographyType => JGeographyType,
+  GeometryType => JGeometryType,
   IntegerType => JIntegerType,
   LongType => JLongType,
   MapType => JMapType,
@@ -36,6 +37,7 @@ object JavaDataTypeUtils {
       case DoubleType => JDataTypes.DoubleType
       case FloatType => JDataTypes.FloatType
       case GeographyType => JDataTypes.GeographyType
+      case GeometryType => JDataTypes.GeometryType
       case IntegerType => JDataTypes.IntegerType
       case LongType => JDataTypes.LongType
       case MapType(keyType, valueType) =>
@@ -58,6 +60,7 @@ object JavaDataTypeUtils {
       case _: JDoubleType => DoubleType
       case _: JFloatType => FloatType
       case _: JGeographyType => GeographyType
+      case _: JGeometryType => GeometryType
       case _: JIntegerType => IntegerType
       case _: JLongType => LongType
       case mp: JMapType =>
diff --git a/src/main/scala/com/snowflake/snowpark/internal/JavaUtils.scala b/src/main/scala/com/snowflake/snowpark/internal/JavaUtils.scala
index 08a92b6b..58bd69e7 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/JavaUtils.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/JavaUtils.scala
@@ -25,7 +25,7 @@ import com.snowflake.snowpark.{
 }
 
 import java.io._
-import com.snowflake.snowpark.types.{Geography, Variant}
+import com.snowflake.snowpark.types.{Geography, Geometry, Variant}
 import com.snowflake.snowpark_java.types.InternalUtils
 import com.snowflake.snowpark_java.udtf._
 
@@ -180,14 +180,24 @@ object JavaUtils {
 
   def geographyToString(g: Geography): String = if (g == null) null else g.asGeoJSON()
 
+  def geometryToString(g: Geometry): String = if (g == null) null else g.toString()
+
   def geographyToString(g: com.snowflake.snowpark_java.types.Geography): String =
     if (g == null) null else g.asGeoJSON()
 
+  def geometryToString(g: com.snowflake.snowpark_java.types.Geometry): String =
+    if (g == null) null else g.toString
+
   def stringToGeography(g: String): Geography = if (g == null) null else Geography.fromGeoJSON(g)
 
+  def stringToGeometry(g: String): Geometry = if (g == null) null else Geometry.fromGeoJSON(g)
+
   def stringToJavaGeography(g: String): com.snowflake.snowpark_java.types.Geography =
     if (g == null) null else com.snowflake.snowpark_java.types.Geography.fromGeoJSON(g)
 
+  def stringToJavaGeometry(g: String): com.snowflake.snowpark_java.types.Geometry =
+    if (g == null) null else com.snowflake.snowpark_java.types.Geometry.fromGeoJSON(g)
+
   def variantToString(v: Variant): String = if (v == null) null else v.asJsonString()
 
   def variantToString(v: com.snowflake.snowpark_java.types.Variant): String =
diff --git a/src/main/scala/com/snowflake/snowpark/internal/ParameterUtils.scala b/src/main/scala/com/snowflake/snowpark/internal/ParameterUtils.scala
index adb9fd06..8ec95883 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/ParameterUtils.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/ParameterUtils.scala
@@ -19,6 +19,7 @@ private[snowpark] object ParameterUtils extends Logging {
   // client parameters
   private[snowpark] val SnowparkLazyAnalysis: String = "snowpark_lazy_analysis"
   private[snowpark] val GeographyOutputFormat: String = "geography_output_format"
+  private[snowpark] val GeometryOutputFormat: String = "geometry_output_format"
   private[snowpark] val SnowparkUseScopedTempObjects: String = "snowpark_use_scoped_temp_objects"
   private[snowpark] val SnowparkEnableClosureCleaner: String = "snowpark_enable_closure_cleaner"
   private[snowpark] val SnowparkRequestTimeoutInSeconds: String =
diff --git a/src/main/scala/com/snowflake/snowpark/internal/ScalaFunctions.scala b/src/main/scala/com/snowflake/snowpark/internal/ScalaFunctions.scala
index 4907da57..db07ab71 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/ScalaFunctions.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/ScalaFunctions.scala
@@ -55,6 +55,7 @@ object ScalaFunctions {
     case t if t =:= typeOf[scala.collection.mutable.Map[String, String]] => true
     case t if t =:= typeOf[scala.collection.mutable.Map[String, Variant]] => true
     case t if t =:= typeOf[Geography] => true
+    case t if t =:= typeOf[Geometry] => true
     case t if t =:= typeOf[Variant] => true
     case t if t <:< typeOf[scala.collection.Iterable[_]] =>
       throw new UnsupportedOperationException(
@@ -97,6 +98,7 @@ object ScalaFunctions {
     case t if t =:= typeOf[scala.collection.mutable.Map[String, Variant]] =>
       UdfColumnSchema(MapType(StringType, VariantType))
     case t if t =:= typeOf[Geography] => UdfColumnSchema(GeographyType)
+    case t if t =:= typeOf[Geometry] => UdfColumnSchema(GeometryType)
     case t if t =:= typeOf[Variant] => UdfColumnSchema(VariantType)
     case t => throw new UnsupportedOperationException(s"Unsupported type $t")
   }
diff --git a/src/main/scala/com/snowflake/snowpark/internal/ServerConnection.scala b/src/main/scala/com/snowflake/snowpark/internal/ServerConnection.scala
index 64e8cfb4..41a97a33 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/ServerConnection.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/ServerConnection.scala
@@ -80,6 +80,7 @@ private[snowpark] object ServerConnection {
       case "VARIANT" => VariantType
       case "OBJECT" => MapType(StringType, StringType)
       case "GEOGRAPHY" => GeographyType
+      case "GEOMETRY" => GeometryType
       case _ => getTypeFromJDBCType(sqlType, precision, scale, signed)
     }
   }
@@ -258,6 +259,7 @@ private[snowpark] class ServerConnection(
       val schema = ServerConnection.convertResultMetaToAttribute(data.getMetaData)
 
       lazy val geographyOutputFormat = getParameterValue(ParameterUtils.GeographyOutputFormat)
+      lazy val geometryOutputFormat = getParameterValue(ParameterUtils.GeometryOutputFormat)
 
       val iterator = new CloseableIterator[Row] {
         private var _currentRow: Row = _
@@ -299,6 +301,13 @@ private[snowpark] class ServerConnection(
                           throw ErrorMessage.MISC_UNSUPPORTED_GEOGRAPHY_FORMAT(
                             geographyOutputFormat)
                       }
+                    case GeometryType =>
+                      geometryOutputFormat match {
+                        case "GeoJSON" => Geometry.fromGeoJSON(data.getString(resultIndex))
+                        case _ =>
+                          throw ErrorMessage.MISC_UNSUPPORTED_GEOMETRY_FORMAT(
+                            geometryOutputFormat)
+                      }
                     case _ =>
                       // ArrayType, StructType, MapType
                       throw new UnsupportedOperationException(
diff --git a/src/main/scala/com/snowflake/snowpark/internal/TypeToSchemaConverter.scala b/src/main/scala/com/snowflake/snowpark/internal/TypeToSchemaConverter.scala
index 45dc780b..d17dc5e5 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/TypeToSchemaConverter.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/TypeToSchemaConverter.scala
@@ -84,6 +84,7 @@ object TypeToSchemaConverter {
 
       case t if t =:= typeOf[Variant] => (VariantType, true)
       case t if t =:= typeOf[Geography] => (GeographyType, true)
+      case t if t =:= typeOf[Geometry] => (GeometryType, true)
       case t if t =:= typeOf[Date] => (DateType, true)
       case t if t =:= typeOf[Timestamp] => (TimestampType, true)
       case t if t =:= typeOf[Time] => (TimeType, true)
diff --git a/src/main/scala/com/snowflake/snowpark/internal/UDXRegistrationHandler.scala b/src/main/scala/com/snowflake/snowpark/internal/UDXRegistrationHandler.scala
index 18a7ab49..acf9b62e 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/UDXRegistrationHandler.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/UDXRegistrationHandler.scala
@@ -546,6 +546,7 @@ class UDXRegistrationHandler(session: Session) extends Logging {
       s"""
          |import com.snowflake.snowpark.internal.JavaUtils;
          |import com.snowflake.snowpark.types.Geography;
+         |import com.snowflake.snowpark.types.Geometry;
          |import com.snowflake.snowpark.types.Variant;
          |import com.snowflake.snowpark.Row;
          |import java.util.Spliterator;
@@ -589,6 +590,7 @@ class UDXRegistrationHandler(session: Session) extends Logging {
       s"""
          |import com.snowflake.snowpark.internal.JavaUtils;
          |import com.snowflake.snowpark_java.types.Geography;
+         |import com.snowflake.snowpark_java.types.Geometry;
          |import com.snowflake.snowpark_java.types.Variant;
          |import com.snowflake.snowpark_java.Row;
          |import java.util.stream.Stream;
@@ -693,6 +695,7 @@ class UDXRegistrationHandler(session: Session) extends Logging {
       s"""
          |import com.snowflake.snowpark.internal.JavaUtils;
          |import com.snowflake.snowpark.types.Geography;
+         |import com.snowflake.snowpark.types.Geometry;
          |import com.snowflake.snowpark.types.Variant;
          |import scala.collection.JavaConverters;
          |import com.snowflake.snowpark.Session;
@@ -717,6 +720,7 @@ class UDXRegistrationHandler(session: Session) extends Logging {
       s"""
          |import com.snowflake.snowpark.internal.JavaUtils;
          |import com.snowflake.snowpark_java.types.Geography;
+         |import com.snowflake.snowpark_java.types.Geometry;
          |import com.snowflake.snowpark_java.types.Variant;
          |import com.snowflake.snowpark_java.Session;
          |import com.snowflake.snowpark_java.sproc.JavaSProc${numArgs - 1};
@@ -773,6 +777,7 @@ class UDXRegistrationHandler(session: Session) extends Logging {
       s"""
          |import com.snowflake.snowpark.internal.JavaUtils;
          |import com.snowflake.snowpark.types.Geography;
+         |import com.snowflake.snowpark.types.Geometry;
          |import com.snowflake.snowpark.types.Variant;
          |import scala.collection.JavaConverters;
          |
@@ -794,6 +799,7 @@ class UDXRegistrationHandler(session: Session) extends Logging {
       s"""
          |import com.snowflake.snowpark.internal.JavaUtils;
          |import com.snowflake.snowpark_java.types.Geography;
+         |import com.snowflake.snowpark_java.types.Geometry;
          |import com.snowflake.snowpark_java.types.Variant;
          |import com.snowflake.snowpark_java.udf.*;
          |
@@ -829,6 +835,8 @@ class UDXRegistrationHandler(session: Session) extends Logging {
         case ArrayType(VariantType) => s"JavaUtils.stringArrayToJavaVariantArray(${arg.name})"
         case GeographyType if isScalaUDF => s"JavaUtils.stringToGeography(${arg.name})"
         case GeographyType => s"JavaUtils.stringToJavaGeography(${arg.name})"
+        case GeometryType if isScalaUDF => s"JavaUtils.stringToGeometry(${arg.name})"
+        case GeometryType => s"JavaUtils.stringToJavaGeometry(${arg.name})"
         case VariantType if isScalaUDF => s"JavaUtils.stringToVariant(${arg.name})"
         case VariantType => s"JavaUtils.stringToJavaVariant(${arg.name})"
         case _ => arg.name
@@ -850,6 +858,7 @@ class UDXRegistrationHandler(session: Session) extends Logging {
   private def convertReturnValue(returnValue: UdfColumnSchema, value: String): String = {
     returnValue.dataType match {
       case GeographyType => s"JavaUtils.geographyToString($value)"
+      case GeometryType => s"JavaUtils.geometryToString($value)"
       case VariantType => s"JavaUtils.variantToString($value)"
       case MapType(_, VariantType) => s"JavaUtils.javaVariantMapToStringMap($value)"
       case ArrayType(VariantType) => s"JavaUtils.variantArrayToStringArray($value)"
diff --git a/src/main/scala/com/snowflake/snowpark/internal/analyzer/DataTypeMapper.scala b/src/main/scala/com/snowflake/snowpark/internal/analyzer/DataTypeMapper.scala
index 598dd166..2781e22c 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/analyzer/DataTypeMapper.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/analyzer/DataTypeMapper.scala
@@ -30,7 +30,8 @@ object DataTypeMapper {
       case None => "NULL"
       case Some(dt) =>
         (value, dt) match {
-          case (_, _: ArrayType | _: MapType | _: StructType | GeographyType) if value == null =>
+          case (_, _: ArrayType | _: MapType | _: StructType | GeographyType | GeometryType)
+              if value == null =>
             "NULL"
           case (_, IntegerType) if value == null => "NULL :: int"
           case (_, ShortType) if value == null => "NULL :: smallint"
@@ -88,6 +89,7 @@ object DataTypeMapper {
     if (isNullable) {
       dataType match {
         case GeographyType => "TRY_TO_GEOGRAPHY(NULL)"
+        case GeometryType => "TRY_TO_GEOMETRY(NULL)"
         case ArrayType(_) => "PARSE_JSON('NULL')::ARRAY"
         case _ => "NULL :: " + convertToSFType(dataType)
       }
@@ -104,6 +106,7 @@ object DataTypeMapper {
         case _: MapType => "to_object(parse_json('0'))"
         case VariantType => "to_variant(0)"
         case GeographyType => "to_geography('POINT(-122.35 37.55)')"
+        case GeometryType => "to_geometry('POINT(-122.35 37.55)')"
         case _ =>
           throw new UnsupportedOperationException(s"Unsupported data type: ${dataType.typeName}")
       }
diff --git a/src/main/scala/com/snowflake/snowpark/types/Geometry.scala b/src/main/scala/com/snowflake/snowpark/types/Geometry.scala
new file mode 100644
index 00000000..18e0050f
--- /dev/null
+++ b/src/main/scala/com/snowflake/snowpark/types/Geometry.scala
@@ -0,0 +1,60 @@
+package com.snowflake.snowpark.types
+
+import java.io.{IOException, UncheckedIOException}
+
+/**
+ * Companion object of Geometry class.
+ * @since 1.12.0
+ */
+object Geometry {
+
+  /**
+   * Creates a Geometry class from a GeoJSON string
+   *
+   * @param g GeoJSON string
+   * @return a Geometry class
+   * @since 1.12.0
+   */
+  def fromGeoJSON(g: String): Geometry = new Geometry(g)
+}
+
+/**
+ * Scala representation of Snowflake Geometry data.
+ * Only support GeoJSON format.
+ *
+ * @since 1.12.0
+ */
+class Geometry private (private val stringData: String) {
+  if (stringData == null) {
+    throw new UncheckedIOException(
+      new IOException("Cannot create geometry object from null input"))
+  }
+
+  /**
+   * Returns whether the Geometry object equals to the input object.
+   *
+   * @return GeoJSON string
+   * @since 1.12.0
+   */
+  override def equals(obj: Any): Boolean =
+    obj match {
+      case g: Geometry => stringData.equals(g.stringData)
+      case _ => false
+    }
+
+  /**
+   * Returns the hashCode of the stored GeoJSON string.
+   *
+   * @return hash code
+   * @since 1.12.0
+   */
+  override def hashCode(): Int = stringData.hashCode
+
+  /**
+   * Returns the underling string data for GeoJSON.
+   *
+   * @return GeoJSON string
+   * @since 1.12.0
+   */
+  override def toString: String = stringData
+}
diff --git a/src/main/scala/com/snowflake/snowpark/types/GeometryType.scala b/src/main/scala/com/snowflake/snowpark/types/GeometryType.scala
new file mode 100644
index 00000000..9088b663
--- /dev/null
+++ b/src/main/scala/com/snowflake/snowpark/types/GeometryType.scala
@@ -0,0 +1,11 @@
+package com.snowflake.snowpark.types
+
+/**
+ * Geometry data type. This maps to GEOMETRY data type in Snowflake.
+ * @since 1.12.0
+ */
+object GeometryType extends DataType {
+  override def toString: String = {
+    s"GeometryType"
+  }
+}
diff --git a/src/main/scala/com/snowflake/snowpark/types/Variant.scala b/src/main/scala/com/snowflake/snowpark/types/Variant.scala
index e297e05c..6ca52a12 100644
--- a/src/main/scala/com/snowflake/snowpark/types/Variant.scala
+++ b/src/main/scala/com/snowflake/snowpark/types/Variant.scala
@@ -64,6 +64,7 @@ private[snowpark] object Variant {
     obj match {
       case v: Variant => v.value
       case g: Geography => new Variant(g.asGeoJSON()).value
+      case g: Geometry => new Variant(g.toString).value
       case _ => MAPPER.valueToTree(obj)
     }
   }
diff --git a/src/main/scala/com/snowflake/snowpark/types/package.scala b/src/main/scala/com/snowflake/snowpark/types/package.scala
index 231d56d2..9f87d3d5 100644
--- a/src/main/scala/com/snowflake/snowpark/types/package.scala
+++ b/src/main/scala/com/snowflake/snowpark/types/package.scala
@@ -27,6 +27,7 @@ package object types {
       case ArrayType(StringType) => "String[]"
       case MapType(StringType, StringType) => "java.util.Map<String,String>"
       case GeographyType => "Geography"
+      case GeometryType => "Geometry"
       case VariantType => "Variant"
       // StructType is only used for defining schema
       // case StructType(_) => // Not Supported
@@ -40,6 +41,7 @@ package object types {
   private[snowpark] def toUDFArgumentType(datatype: DataType): String =
     datatype match {
       case GeographyType => classOf[java.lang.String].getCanonicalName
+      case GeometryType => classOf[java.lang.String].getCanonicalName
       case VariantType => classOf[java.lang.String].getCanonicalName
       case ArrayType(VariantType) => "String[]"
       case MapType(StringType, VariantType) => "java.util.Map<String,String>"
@@ -65,6 +67,7 @@ package object types {
       case MapType(_, _) => "OBJECT"
       case VariantType => "VARIANT"
       case GeographyType => "GEOGRAPHY"
+      case GeometryType => "GEOMETRY"
       case _ =>
         throw new UnsupportedOperationException(s"Unsupported data type: ${dataType.typeName}")
     }
diff --git a/src/test/java/com/snowflake/snowpark_test/JavaDataTypesSuite.java b/src/test/java/com/snowflake/snowpark_test/JavaDataTypesSuite.java
index 15ca9037..b56883ed 100644
--- a/src/test/java/com/snowflake/snowpark_test/JavaDataTypesSuite.java
+++ b/src/test/java/com/snowflake/snowpark_test/JavaDataTypesSuite.java
@@ -31,6 +31,8 @@ public void javaTypeToScalaTypeConversion() {
         .equals(com.snowflake.snowpark.types.DateType$.MODULE$);
     assert JavaDataTypeUtils.javaTypeToScalaType(DataTypes.GeographyType)
         .equals(com.snowflake.snowpark.types.GeographyType$.MODULE$);
+    assert JavaDataTypeUtils.javaTypeToScalaType(DataTypes.GeometryType)
+        .equals(com.snowflake.snowpark.types.GeometryType$.MODULE$);
     assert JavaDataTypeUtils.javaTypeToScalaType(DataTypes.StringType)
         .equals(com.snowflake.snowpark.types.StringType$.MODULE$);
     assert JavaDataTypeUtils.javaTypeToScalaType(DataTypes.TimestampType)
@@ -100,6 +102,8 @@ public void scalaTypeToJavaTypeConversion() {
     assert JavaDataTypeUtils.scalaTypeToJavaType(
             com.snowflake.snowpark.types.GeographyType$.MODULE$)
         .equals(DataTypes.GeographyType);
+    assert JavaDataTypeUtils.scalaTypeToJavaType(com.snowflake.snowpark.types.GeometryType$.MODULE$)
+        .equals(DataTypes.GeometryType);
     assert JavaDataTypeUtils.scalaTypeToJavaType(com.snowflake.snowpark.types.StringType$.MODULE$)
         .equals(DataTypes.StringType);
     assert JavaDataTypeUtils.scalaTypeToJavaType(
diff --git a/src/test/java/com/snowflake/snowpark_test/JavaGeographySuite.java b/src/test/java/com/snowflake/snowpark_test/JavaGeographySuite.java
index be8b73ed..ddb03e15 100644
--- a/src/test/java/com/snowflake/snowpark_test/JavaGeographySuite.java
+++ b/src/test/java/com/snowflake/snowpark_test/JavaGeographySuite.java
@@ -1,6 +1,7 @@
 package com.snowflake.snowpark_test;
 
 import com.snowflake.snowpark_java.types.Geography;
+import com.snowflake.snowpark_java.types.Geometry;
 import org.junit.Test;
 
 public class JavaGeographySuite {
@@ -8,10 +9,15 @@ public class JavaGeographySuite {
   private final String testData =
       "{\"geometry\":{\"type\":\"Point\",\"coordinates\":[125.6, 10.1]}}";
 
+  private final String testData2 =
+      "{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}";
+
   @Test
   public void testRoundTrip() {
     assert (Geography.fromGeoJSON(testData).asGeoJSON().equals(testData));
     assert (Geography.fromGeoJSON(testData).toString().equals(testData));
+
+    assert (Geometry.fromGeoJSON(testData2).toString().equals(testData2));
   }
 
   @Test
@@ -26,5 +32,8 @@ public void testEqual() {
     assert Geography.fromGeoJSON(testData).hashCode() == Geography.fromGeoJSON(testData).hashCode();
     assert Geography.fromGeoJSON(testData2).hashCode()
         != Geography.fromGeoJSON(testData).hashCode();
+
+    assert (Geometry.fromGeoJSON(this.testData2).equals(Geometry.fromGeoJSON(this.testData2)));
+    assert !(Geometry.fromGeoJSON(this.testData2).equals(Geometry.fromGeoJSON(testData2)));
   }
 }
diff --git a/src/test/java/com/snowflake/snowpark_test/JavaRowSuite.java b/src/test/java/com/snowflake/snowpark_test/JavaRowSuite.java
index 0982c790..bdde050e 100644
--- a/src/test/java/com/snowflake/snowpark_test/JavaRowSuite.java
+++ b/src/test/java/com/snowflake/snowpark_test/JavaRowSuite.java
@@ -2,6 +2,7 @@
 
 import com.snowflake.snowpark_java.Row;
 import com.snowflake.snowpark_java.types.Geography;
+import com.snowflake.snowpark_java.types.Geometry;
 import com.snowflake.snowpark_java.types.Variant;
 import java.math.BigDecimal;
 import java.sql.Date;
@@ -87,17 +88,24 @@ public void getters2() {
             binary,
             new Variant(3),
             Geography.fromGeoJSON("{\"type\":\"Point\",\"coordinates\":[30,10]}"),
-            new BigDecimal(12345));
+            new BigDecimal(12345),
+            Geometry.fromGeoJSON(
+                "{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}"));
 
-    assert row.size() == 4;
+    assert row.size() == 5;
     assert Arrays.equals(row.getBinary(0), binary);
     assert row.getVariant(1).equals(new Variant(3));
     assert row.getGeography(2)
         .equals(Geography.fromGeoJSON("{\"type\":\"Point\",\"coordinates\":[30,10]}"));
     assert row.getDecimal(3).equals(new BigDecimal(12345));
+    assert row.getGeometry(4)
+        .equals(
+            Geometry.fromGeoJSON(
+                "{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}"));
 
     assert row.toString()
-        .equals("Row[Binary(1,2),3,{\"type\":\"Point\",\"coordinates\":[30,10]},12345]");
+        .equals(
+            "Row[Binary(1,2),3,{\"type\":\"Point\",\"coordinates\":[30,10]},12345,{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}]");
   }
 
   @Test
@@ -132,10 +140,13 @@ public void getter5() {
         Row.create(
             new com.snowflake.snowpark.types.Variant(1),
             com.snowflake.snowpark.types.Geography.fromGeoJSON(
-                "{\"type\":\"Point\",\"coordinates\":[30,10]}"));
+                "{\"type\":\"Point\",\"coordinates\":[30,10]}"),
+            com.snowflake.snowpark.types.Geometry.fromGeoJSON(
+                "{\"coordinates\": [2.000000000000000e+01,4.000000000000000e+01],\"type\": \"Point\"}"));
 
     assert row.get(0) instanceof Variant;
     assert row.get(1) instanceof Geography;
+    assert row.get(2) instanceof Geometry;
   }
 
   @Test
diff --git a/src/test/scala/com/snowflake/code_verification/JavaScalaAPISuite.scala b/src/test/scala/com/snowflake/code_verification/JavaScalaAPISuite.scala
index 4d2aa737..d4394054 100644
--- a/src/test/scala/com/snowflake/code_verification/JavaScalaAPISuite.scala
+++ b/src/test/scala/com/snowflake/code_verification/JavaScalaAPISuite.scala
@@ -552,6 +552,20 @@ class JavaScalaAPISuite extends FunSuite {
         .containsSameFunctionNames(classOf[JavaGeographyType], ScalaGeograhyType.getClass))
   }
 
+  test("Geometry") {
+    import com.snowflake.snowpark_java.types.{Geometry => JavaGeometry}
+    import com.snowflake.snowpark.types.{Geometry => ScalaGeometry}
+    assert(ClassUtils.containsSameFunctionNames(classOf[JavaGeometry], classOf[ScalaGeometry]))
+  }
+
+  test("GeometryType") {
+    import com.snowflake.snowpark_java.types.{GeometryType => JavaGeometryType}
+    import com.snowflake.snowpark.types.{GeometryType => ScalaGeometryType}
+    assert(
+      ClassUtils
+        .containsSameFunctionNames(classOf[JavaGeometryType], ScalaGeometryType.getClass))
+  }
+
   test("IntegerType") {
     import com.snowflake.snowpark_java.types.{IntegerType => JavaIntegerType}
     import com.snowflake.snowpark.types.{IntegerType => ScalaIntegerType}
diff --git a/src/test/scala/com/snowflake/snowpark/ErrorMessageSuite.scala b/src/test/scala/com/snowflake/snowpark/ErrorMessageSuite.scala
index 2a3df7f7..937b93e6 100644
--- a/src/test/scala/com/snowflake/snowpark/ErrorMessageSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark/ErrorMessageSuite.scala
@@ -849,4 +849,14 @@ class ErrorMessageSuite extends FunSuite {
           "Explode function should be either Map or Array types.\n" +
           "The input argument type: Integer"))
   }
+
+  test("MISC_UNSUPPORTED_GEOMETRY_FORMAT") {
+    val ex = ErrorMessage.MISC_UNSUPPORTED_GEOMETRY_FORMAT("KWT")
+    assert(ex.telemetryMessage.equals(ErrorMessage.getMessage("0425")))
+    assert(
+      ex.message.startsWith(
+        "Error Code: 0425, Error message: " +
+          "Unsupported Geometry output format: KWT." +
+          " Please set session parameter GEOMETRY_OUTPUT_FORMAT to GeoJSON."))
+  }
 }
diff --git a/src/test/scala/com/snowflake/snowpark/SNTestBase.scala b/src/test/scala/com/snowflake/snowpark/SNTestBase.scala
index c938fdcc..3fd98755 100644
--- a/src/test/scala/com/snowflake/snowpark/SNTestBase.scala
+++ b/src/test/scala/com/snowflake/snowpark/SNTestBase.scala
@@ -76,7 +76,8 @@ trait SNTestBase extends FunSuite with BeforeAndAfterAll with SFTestUtils with S
     TypeMap("variant", "variant", Types.VARCHAR, VariantType),
     TypeMap("object", "object", Types.VARCHAR, MapType(StringType, StringType)),
     TypeMap("array", "array", Types.VARCHAR, ArrayType(StringType)),
-    TypeMap("geography", "geography", Types.VARCHAR, GeographyType))
+    TypeMap("geography", "geography", Types.VARCHAR, GeographyType),
+    TypeMap("geometry", "geometry", Types.VARCHAR, GeometryType))
 
   implicit lazy val session: Session = {
     TestUtils.tryToLoadFipsProvider()
diff --git a/src/test/scala/com/snowflake/snowpark/TestUtils.scala b/src/test/scala/com/snowflake/snowpark/TestUtils.scala
index d0514964..e1abedff 100644
--- a/src/test/scala/com/snowflake/snowpark/TestUtils.scala
+++ b/src/test/scala/com/snowflake/snowpark/TestUtils.scala
@@ -260,6 +260,8 @@ object TestUtils extends Logging {
       case (a: Geography, b: Geography) =>
         // todo: improve Geography.equals method
         a.asGeoJSON().replaceAll("\\s", "").equals(b.asGeoJSON().replaceAll("\\s", ""))
+      case (a: Geometry, b: Geometry) =>
+        a.toString.replaceAll("\\s", "").equals(b.toString.replaceAll("\\s", ""))
       case (a, b) => a == b
     }
     if (!res) {
diff --git a/src/test/scala/com/snowflake/snowpark/UDFInternalSuite.scala b/src/test/scala/com/snowflake/snowpark/UDFInternalSuite.scala
index 9a1eae3a..24137bb3 100644
--- a/src/test/scala/com/snowflake/snowpark/UDFInternalSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark/UDFInternalSuite.scala
@@ -29,10 +29,16 @@ class UDFInternalSuite extends TestData {
     super.afterAll
   }
 
-  test("Test temp udf not failing back to upload jar", JavaStoredProcExclude) {
+  // todo: re-enable in SNOW-1227362
+  // The new Geometry data type introduced in the release 1.12.0.
+  // The package suites use the latest Snowpark package on the server side,
+  // which doesn't have Geometry type, then, all package suites will fail
+  // before the server side release. So we have to temporarily disable those test suite
+  // until Snowpark 1.12.0 release.
+  ignore("Test temp udf not failing back to upload jar", JavaStoredProcExclude) {
     val newSession = Session.builder.configFile(defaultProfile).create
     val mockSession = spy(newSession)
-    TestUtils.addDepsToClassPath(mockSession, None, usePackages = true)
+    TestUtils.addDepsToClassPath(mockSession, None)
     val path = UDFClassPath.getPathForClass(classOf[com.snowflake.snowpark.Session]).get
 
     val doubleUDF = mockSession.udf.registerTemporary((x: Int) => x + x)
@@ -179,6 +185,13 @@ class UDFInternalSuite extends TestData {
   }
 }
 
+// todo: re-enable in SNOW-1227362
+// The new Geometry data type introduced in the release 1.12.0.
+// The package suites use the latest Snowpark package on the server side,
+// which doesn't have Geometry type, then, all package suites will fail
+// before the server side release. So we have to temporarily disable those test suite
+// until Snowpark 1.12.0 release.
+/*
 @UDFPackageTest
 class PackageUDFSuite extends UDFSuite {
   override def beforeAll: Unit = {
@@ -214,3 +227,4 @@ class PackageUDTFSuite extends UDTFSuite {
     super.afterAll()
   }
 }
+ */
diff --git a/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala b/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala
index b5f87c14..2067212f 100644
--- a/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala
@@ -135,11 +135,12 @@ class UtilsSuite extends SNTestBase {
         BooleanType))
     assert(TypeToSchemaConverter.inferSchema[Variant]().head.dataType == VariantType)
     assert(TypeToSchemaConverter.inferSchema[Geography]().head.dataType == GeographyType)
+    assert(TypeToSchemaConverter.inferSchema[Geometry]().head.dataType == GeometryType)
 
     // tuple
     assert(
       TypeToSchemaConverter
-        .inferSchema[(Int, Boolean, Double, Geography, Map[String, Boolean])]()
+        .inferSchema[(Int, Boolean, Double, Geography, Map[String, Boolean], Geometry)]()
         .treeString(0) ==
         """root
       | |--_1: Integer (nullable = false)
@@ -147,6 +148,7 @@ class UtilsSuite extends SNTestBase {
       | |--_3: Double (nullable = false)
       | |--_4: Geography (nullable = true)
       | |--_5: Map (nullable = true)
+      | |--_6: Geometry (nullable = true)
       |""".stripMargin)
 
     // case class
@@ -222,7 +224,8 @@ class UtilsSuite extends SNTestBase {
           Geography,
           Date,
           Time,
-          Timestamp)]()
+          Timestamp,
+          Geometry)]()
       .treeString(0) ==
       """root
           | |--_1: Integer (nullable = true)
@@ -242,6 +245,7 @@ class UtilsSuite extends SNTestBase {
           | |--_15: Date (nullable = true)
           | |--_16: Time (nullable = true)
           | |--_17: Timestamp (nullable = true)
+          | |--_18: Geometry (nullable = true)
           |""".stripMargin
   }
 
diff --git a/src/test/scala/com/snowflake/snowpark_test/DataFrameSuite.scala b/src/test/scala/com/snowflake/snowpark_test/DataFrameSuite.scala
index 65e141dc..e07b94a9 100644
--- a/src/test/scala/com/snowflake/snowpark_test/DataFrameSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark_test/DataFrameSuite.scala
@@ -1284,9 +1284,15 @@ trait DataFrameSuite extends TestData with BeforeAndAfterEach {
         StructField("array", ArrayType(null)),
         StructField("map", MapType(null, null)),
         StructField("variant", VariantType),
-        StructField("geography", GeographyType)))
+        StructField("geography", GeographyType),
+        StructField("geometry", GeometryType)))
     val data = Seq(
-      Row(Array("'", 2), Map("'" -> 1), new Variant(1), Geography.fromGeoJSON("POINT(30 10)")),
+      Row(
+        Array("'", 2),
+        Map("'" -> 1),
+        new Variant(1),
+        Geography.fromGeoJSON("POINT(30 10)"),
+        Geometry.fromGeoJSON("POINT(20 40)")),
       Row(null, null, null, null, null))
 
     val df = session.createDataFrame(data, schema)
@@ -1297,6 +1303,7 @@ trait DataFrameSuite extends TestData with BeforeAndAfterEach {
           | |--MAP: Map (nullable = true)
           | |--VARIANT: Variant (nullable = true)
           | |--GEOGRAPHY: Geography (nullable = true)
+          | |--GEOMETRY: Geometry (nullable = true)
           |""".stripMargin)
     df.show()
     val expected =
@@ -1311,8 +1318,15 @@ trait DataFrameSuite extends TestData with BeforeAndAfterEach {
                                   |    10
                                   |  ],
                                   |  "type": "Point"
-                                  |}""".stripMargin)),
-        Row(null, null, null, null))
+                                  |}""".stripMargin),
+          Geometry.fromGeoJSON("""{
+              |  "coordinates": [
+              |    2.000000000000000e+01,
+              |    4.000000000000000e+01
+              |  ],
+              |  "type": "Point"
+              |}""".stripMargin)),
+        Row(null, null, null, null, null))
     checkAnswer(df, expected, sort = false)
   }
 
@@ -1404,7 +1418,12 @@ trait DataFrameSuite extends TestData with BeforeAndAfterEach {
 
     // case class
     val df3 =
-      session.createDataFrame(Seq(Table1(new Variant(1), Geography.fromGeoJSON("point(10 10)"))))
+      session.createDataFrame(
+        Seq(
+          Table1(
+            new Variant(1),
+            Geography.fromGeoJSON("point(10 10)"),
+            Geometry.fromGeoJSON("point(20 40)"))))
     df3.schema.printTreeString()
     checkAnswer(
       df3,
@@ -1417,10 +1436,17 @@ trait DataFrameSuite extends TestData with BeforeAndAfterEach {
                                   |    10
                                   |  ],
                                   |  "type": "Point"
-                                  |}""".stripMargin))))
+                                  |}""".stripMargin),
+          Geometry.fromGeoJSON("""{
+              |  "coordinates": [
+              |    2.000000000000000e+01,
+              |    4.000000000000000e+01
+              |  ],
+              |  "type": "Point"
+              |}""".stripMargin))))
   }
 
-  case class Table1(variant: Variant, geography: Geography)
+  case class Table1(variant: Variant, geography: Geography, geometry: Geometry)
 
   test("create nullable dataFrame with schema inference") {
     val df = Seq((1, Some(1), None), (2, Some(3), Some(true)))
diff --git a/src/test/scala/com/snowflake/snowpark_test/IndependentClassSuite.scala b/src/test/scala/com/snowflake/snowpark_test/IndependentClassSuite.scala
index 661e29ac..7ffe6950 100644
--- a/src/test/scala/com/snowflake/snowpark_test/IndependentClassSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark_test/IndependentClassSuite.scala
@@ -15,7 +15,10 @@ class IndependentClassSuite extends FunSuite {
 
     checkDependencies(
       "target/classes/com/snowflake/snowpark/types/Variant$.class",
-      Seq("com.snowflake.snowpark.types.Variant", "com.snowflake.snowpark.types.Geography"))
+      Seq(
+        "com.snowflake.snowpark.types.Variant",
+        "com.snowflake.snowpark.types.Geography",
+        "com.snowflake.snowpark.types.Geometry"))
   }
 
   test("java variant") {
@@ -42,6 +45,22 @@ class IndependentClassSuite extends FunSuite {
       Seq("com.snowflake.snowpark_java.types.Geography"))
   }
 
+  test("scala geometry") {
+    checkDependencies(
+      "target/classes/com/snowflake/snowpark/types/Geometry.class",
+      Seq("com.snowflake.snowpark.types.Geometry"))
+
+    checkDependencies(
+      "target/classes/com/snowflake/snowpark/types/Geometry$.class",
+      Seq("com.snowflake.snowpark.types.Geometry"))
+  }
+
+  test("java geometry") {
+    checkDependencies(
+      "target/classes/com/snowflake/snowpark_java/types/Geometry.class",
+      Seq("com.snowflake.snowpark_java.types.Geometry"))
+  }
+
   // negative test, to make sure this test method works
   test("session") {
     assertThrows[TestFailedException] {
diff --git a/src/test/scala/com/snowflake/snowpark_test/JavaUtilsSuite.scala b/src/test/scala/com/snowflake/snowpark_test/JavaUtilsSuite.scala
index cba90ea4..7bec5be4 100644
--- a/src/test/scala/com/snowflake/snowpark_test/JavaUtilsSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark_test/JavaUtilsSuite.scala
@@ -19,12 +19,25 @@ class JavaUtilsSuite extends FunSuite {
       geographyToString(com.snowflake.snowpark_java.types.Geography.fromGeoJSON(data)) == data)
   }
 
+  test("geometry to string") {
+    val data =
+      "{\"coordinates\": [2.000000000000000e+01,4.000000000000000e+01],\"type\": \"Point\"}"
+    assert(geometryToString(com.snowflake.snowpark.types.Geometry.fromGeoJSON(data)) == data)
+    assert(geometryToString(com.snowflake.snowpark_java.types.Geometry.fromGeoJSON(data)) == data)
+  }
+
   test("string to geography") {
     val data = "{\"type\":\"Point\",\"coordinates\":[125.6, 10.1]}"
     assert(stringToGeography(data).isInstanceOf[com.snowflake.snowpark.types.Geography])
     assert(stringToJavaGeography(data).isInstanceOf[com.snowflake.snowpark_java.types.Geography])
   }
 
+  test("string to geometry") {
+    val data = "{\"type\":\"Point\",\"coordinates\":[125.6, 10.1]}"
+    assert(stringToGeometry(data).isInstanceOf[com.snowflake.snowpark.types.Geometry])
+    assert(stringToJavaGeometry(data).isInstanceOf[com.snowflake.snowpark_java.types.Geometry])
+  }
+
   test("variant to string") {
     assert(variantToString(new com.snowflake.snowpark.types.Variant(1)) == "1")
     assert(variantToString(new com.snowflake.snowpark_java.types.Variant(1)) == "1")
diff --git a/src/test/scala/com/snowflake/snowpark_test/LargeDataFrameSuite.scala b/src/test/scala/com/snowflake/snowpark_test/LargeDataFrameSuite.scala
index acc20885..a2f70f98 100644
--- a/src/test/scala/com/snowflake/snowpark_test/LargeDataFrameSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark_test/LargeDataFrameSuite.scala
@@ -171,7 +171,8 @@ class LargeDataFrameSuite extends TestData {
         StructField("array", ArrayType(null)),
         StructField("map", MapType(null, null)),
         StructField("variant", VariantType),
-        StructField("geography", GeographyType)))
+        StructField("geography", GeographyType),
+        StructField("geometry", GeometryType)))
 
     val rowCount = 350
     val largeData = new ArrayBuffer[Row]()
@@ -182,7 +183,8 @@ class LargeDataFrameSuite extends TestData {
           Array("'", 2),
           Map("'" -> 1),
           new Variant(1),
-          Geography.fromGeoJSON("POINT(30 10)")))
+          Geography.fromGeoJSON("POINT(30 10)"),
+          Geometry.fromGeoJSON("POINT(20 40)")))
     }
     largeData.append(Row(rowCount, null, null, null, null, null))
 
@@ -195,6 +197,7 @@ class LargeDataFrameSuite extends TestData {
           | |--MAP: Map (nullable = true)
           | |--VARIANT: Variant (nullable = true)
           | |--GEOGRAPHY: Geography (nullable = true)
+          | |--GEOMETRY: Geometry (nullable = true)
           |""".stripMargin)
 
     val expected = new ArrayBuffer[Row]()
@@ -211,9 +214,16 @@ class LargeDataFrameSuite extends TestData {
                                   |    10
                                   |  ],
                                   |  "type": "Point"
-                                  |}""".stripMargin)))
+                                  |}""".stripMargin),
+          Geometry.fromGeoJSON("""{
+            |  "coordinates": [
+            |    2.000000000000000e+01,
+            |    4.000000000000000e+01
+            |  ],
+            |  "type": "Point"
+            |}""".stripMargin)))
     }
-    expected.append(Row(rowCount, null, null, null, null))
+    expected.append(Row(rowCount, null, null, null, null, null))
     checkAnswer(df.sort(col("id")), expected, sort = false)
   }
 
diff --git a/src/test/scala/com/snowflake/snowpark_test/ResultSchemaSuite.scala b/src/test/scala/com/snowflake/snowpark_test/ResultSchemaSuite.scala
index a81d61de..a06500eb 100644
--- a/src/test/scala/com/snowflake/snowpark_test/ResultSchemaSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark_test/ResultSchemaSuite.scala
@@ -1,7 +1,7 @@
 package com.snowflake.snowpark_test
 
 import com.snowflake.snowpark.functions._
-import com.snowflake.snowpark.types.{GeographyType, TimeType}
+import com.snowflake.snowpark.types.{GeographyType, Geometry, GeometryType, TimeType}
 import com.snowflake.snowpark.{TestData, UnstableTest}
 
 import java.sql.Types
@@ -145,7 +145,7 @@ class ResultSchemaSuite extends TestData {
     statement.close()
   }
 
-  test("verify Geometry schema type") {
+  test("verify Geography schema type") {
     try {
       runQuery(s"alter session set GEOGRAPHY_OUTPUT_FORMAT='GeoJSON'", session)
       var statement = runQueryReturnStatement(s"select geography from $fullTypesTable2", session)
@@ -176,6 +176,37 @@ class ResultSchemaSuite extends TestData {
     }
   }
 
+  test("verify Geometry schema type") {
+    try {
+      runQuery(s"alter session set GEOMETRY_OUTPUT_FORMAT='GeoJSON'", session)
+      var statement = runQueryReturnStatement(s"select geometry from $fullTypesTable2", session)
+      var resultMeta = statement.getResultSet.getMetaData
+      var tsSchema = session.table(fullTypesTable2).select(col("geometry")).schema
+      assert(resultMeta.getColumnType(1) == Types.VARCHAR)
+      assert(tsSchema.head.dataType == GeometryType)
+      statement.close()
+
+      runQuery(s"alter session set GEOMETRY_OUTPUT_FORMAT='WKT'", session)
+      statement = runQueryReturnStatement(s"select geometry from $fullTypesTable2", session)
+      resultMeta = statement.getResultSet.getMetaData
+      tsSchema = session.table(fullTypesTable2).select(col("geometry")).schema
+      assert(resultMeta.getColumnType(1) == Types.VARCHAR)
+      assert(tsSchema.head.dataType == GeometryType)
+      statement.close()
+
+      runQuery(s"alter session set GEOMETRY_OUTPUT_FORMAT='WKB'", session)
+      statement = runQueryReturnStatement(s"select geometry from $fullTypesTable2", session)
+      resultMeta = statement.getResultSet.getMetaData
+      tsSchema = session.table(fullTypesTable2).select(col("geometry")).schema
+      assert(resultMeta.getColumnType(1) == Types.BINARY)
+      assert(tsSchema.head.dataType == GeometryType)
+      statement.close()
+    } finally {
+      // Assign output format to the default value
+      runQuery(s"alter session set GEOMETRY_OUTPUT_FORMAT='GeoJSON'", session)
+    }
+  }
+
   test("verify Time schema type") {
     val statement = runQueryReturnStatement(s"select time from $fullTypesTable2", session)
     val resultMeta = statement.getResultSet.getMetaData
diff --git a/src/test/scala/com/snowflake/snowpark_test/RowSuite.scala b/src/test/scala/com/snowflake/snowpark_test/RowSuite.scala
index 57bb688c..54aba687 100644
--- a/src/test/scala/com/snowflake/snowpark_test/RowSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark_test/RowSuite.scala
@@ -1,6 +1,6 @@
 package com.snowflake.snowpark_test
 
-import com.snowflake.snowpark.types.{Geography, Variant}
+import com.snowflake.snowpark.types.{Geography, Geometry, Variant}
 import com.snowflake.snowpark.{Row, SNTestBase, SnowparkClientException}
 
 import java.sql.{Date, Timestamp}
@@ -55,9 +55,11 @@ class RowSuite extends SNTestBase {
       Map("a" -> "b"),
       Row(1, 2, 3),
       Array[Byte](1, 9),
-      Geography.fromGeoJSON("{\"type\":\"Point\",\"coordinates\":[30,10]}"))
+      Geography.fromGeoJSON("{\"type\":\"Point\",\"coordinates\":[30,10]}"),
+      Geometry.fromGeoJSON(
+        "{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}"))
 
-    assert(row.length == 19)
+    assert(row.length == 20)
     assert(row.isNullAt(0))
     assert(row.getBoolean(1))
     assert(row.getByte(2) == 1.toByte)
@@ -79,6 +81,13 @@ class RowSuite extends SNTestBase {
         Geography.fromGeoJSON("{\"type\":\"Point\",\"coordinates\":[30,10]}"))
     assertThrows[ClassCastException](row.getBinary(18))
     assert(row.getString(18) == "{\"type\":\"Point\",\"coordinates\":[30,10]}")
+    assert(
+      row.getGeometry(19) ==
+        Geometry.fromGeoJSON(
+          "{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}"))
+    assert(
+      row.getString(19) ==
+        "{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}")
   }
 
   test("number getters") {
diff --git a/src/test/scala/com/snowflake/snowpark_test/UDFSuite.scala b/src/test/scala/com/snowflake/snowpark_test/UDFSuite.scala
index d2e1b734..61ed76a0 100644
--- a/src/test/scala/com/snowflake/snowpark_test/UDFSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark_test/UDFSuite.scala
@@ -2,7 +2,7 @@ package com.snowflake.snowpark_test
 
 import com.snowflake.snowpark._
 import com.snowflake.snowpark.functions.{col, _}
-import com.snowflake.snowpark.types.{Geography, Variant}
+import com.snowflake.snowpark.types.{Geography, Geometry, Variant}
 
 import java.io.{NotSerializableException, Serializable}
 import java.sql.{Date, Time, Timestamp}
@@ -431,6 +431,44 @@ trait UDFSuite extends TestData {
         Row(null)))
   }
 
+  test("Test for Geometry data type") {
+    createTable(tableName, "g geometry")
+    runQuery(s"insert into $tableName values ('POINT(20 40)'), ('POINT(50 60)'), (null)", session)
+    val df = session.table(tableName)
+
+    val geometryUDF = udf((g: Geometry) => {
+      if (g == null) {
+        null
+      } else {
+        if (g.toString.contains("2.000000000000000e+01")) {
+          Geometry.fromGeoJSON(g.toString)
+        } else {
+          Geometry.fromGeoJSON(
+            "{\"coordinates\": [3.000000000000000e+01,1.000000000000000e+01],\"type\": \"Point\"}")
+        }
+      }
+    })
+
+    checkAnswer(
+      df.select(geometryUDF(col("g"))),
+      Seq(
+        Row(Geometry.fromGeoJSON("""{
+              |  "coordinates": [
+              |    2.000000000000000e+01,
+              |    4.000000000000000e+01
+              |  ],
+              |  "type": "Point"
+              |}""".stripMargin)),
+        Row(Geometry.fromGeoJSON("""{
+              |  "coordinates": [
+              |    3.000000000000000e+01,
+              |    1.000000000000000e+01
+              |  ],
+              |  "type": "Point"
+              |}""".stripMargin)),
+        Row(null)))
+  }
+
   // Excluding this test for known Timezone issue in stored proc
   test("Test for Variant Timestamp input", JavaStoredProcExclude) {
     val variantTimestampUDF = udf((v: Variant) => {